crates/nornir-flight/src/door.rs
//! The Flight SQL **door**: who may talk to a nordisk Flight SQL server, and
//! the handshake that answers them.
//!
//! ONE copy for every Flight SQL server in the constellation (skade-flight,
//! knut's graphar-flight, korp's mount of it). Before this module each server
//! carried its own Basic decoder, its own constant-time compare and its own
//! handshake, and the same two client bugs were found and fixed twice.
//!
//! Two credentials, both fed by the ONE nordisk token ladder
//! ([`nornir_service::resolve_token`]: flag → env → token file):
//!
//! - **Bearer** — `authorization: Bearer <token>`; what every nordisk client
//! and every ADBC client sends after a handshake. Empty token and no Basic
//! credential ⇒ OPEN server (dev bring-up).
//! - **Basic** — `authorization: Basic <base64(user:pass)>`; what Power BI's
//! connector and the Arrow Flight SQL ODBC driver send on the Flight
//! `Handshake`. The handshake answers with the bearer token, which the
//! client replays on every later call. Basic is accepted on ordinary calls
//! too, because the ODBC driver sends it there as well.
//!
//! No second compare: [`nornir_service::token_matches`] is the constant-time
//! equality every nordisk server uses.
use std::pin::Pin;
use std::sync::Arc;
use arrow_flight::HandshakeResponse;
use base64::Engine as _;
use base64::alphabet::STANDARD as STANDARD_ALPHABET;
use base64::engine::DecodePaddingMode;
use base64::engine::general_purpose::{GeneralPurpose, GeneralPurposeConfig};
use futures::Stream;
use tonic::{Request, Response, Status};
/// The ONE constant-time compare, re-exported so a door with richer rules
/// (token sets, identities) reaches it through this crate, not a copy.
pub use nornir_service::token_matches;
/// The decoder for a Basic payload: standard alphabet, padding OPTIONAL.
///
/// RFC 4648 says encoders MUST pad; clients do not all read RFC 4648.
/// arrow-go's driver, and a client in the Flight SQL TPC-H comparison, send
/// `Basic YWRtaW46cHc` (no trailing `=`), which the strict STANDARD engine
/// refuses as `InvalidPadding` — a valid user with a valid password, told
/// "invalid credentials". Accepting either form loses nothing: the decoded
/// bytes are the same, and the constant-time comparison decides.
pub const BASIC_PAYLOAD: GeneralPurpose = GeneralPurpose::new(
&STANDARD_ALPHABET,
GeneralPurposeConfig::new().with_decode_padding_mode(DecodePaddingMode::Indifferent),
);
/// Decode a Basic payload (the part after `Basic `) into `(user, pass)`.
/// `None` for anything that is not base64 of UTF-8 `user:pass`.
pub fn decode_basic(b64: &str) -> Option<(String, String)> {
let bytes = BASIC_PAYLOAD.decode(b64.trim()).ok()?;
let text = std::str::from_utf8(&bytes).ok()?;
let (user, pass) = text.split_once(':')?;
Some((user.to_string(), pass.to_string()))
}
/// Does a Basic payload carry exactly `user` / `pass`? Both compares always
/// run (`&`, not `&&`) so timing does not reveal which half differed.
pub fn basic_matches(b64: &str, user: &str, pass: &str) -> bool {
let Some((u, p)) = decode_basic(b64) else { return false };
token_matches(&u, user) & token_matches(&p, pass)
}
/// The door's credentials: one bearer token and an optional Basic pair the
/// handshake exchanges for it.
#[derive(Debug, Clone, Default)]
pub struct DoorAuth {
/// The bearer token. Empty = no bearer.
token: String,
/// Optional Basic credential the handshake exchanges for the token.
basic: Option<(String, String)>,
}
impl DoorAuth {
/// Open server: no credentials checked, handshake issues a placeholder.
pub fn open() -> Self {
Self::default()
}
/// Bearer-only.
pub fn bearer(token: impl Into<String>) -> Self {
Self { token: token.into(), basic: None }
}
/// Add a Basic username/password the handshake accepts.
pub fn with_basic(mut self, user: impl Into<String>, pass: impl Into<String>) -> Self {
self.basic = Some((user.into(), pass.into()));
self
}
pub fn is_open(&self) -> bool {
self.token.is_empty() && self.basic.is_none()
}
/// The token the handshake hands out.
pub fn issued_token(&self) -> &str {
&self.token
}
/// Validate an `authorization` header value.
pub fn check_header(&self, header: Option<&str>) -> Result<(), Status> {
if self.is_open() {
return Ok(());
}
let Some(h) = header else {
return Err(Status::unauthenticated("missing authorization header"));
};
if let Some(t) = h.strip_prefix("Bearer ").or_else(|| h.strip_prefix("bearer ")) {
if !self.token.is_empty() && token_matches(t.trim(), &self.token) {
return Ok(());
}
return Err(Status::unauthenticated("invalid bearer token"));
}
if let Some(b64) = h.strip_prefix("Basic ").or_else(|| h.strip_prefix("basic ")) {
if let Some((user, pass)) = &self.basic {
if basic_matches(b64, user, pass) {
return Ok(());
}
}
return Err(Status::unauthenticated("invalid username or password"));
}
Err(Status::unauthenticated("unsupported authorization scheme"))
}
/// Check a tonic request's `authorization` metadata.
pub fn check_request<T>(&self, req: &Request<T>) -> Result<(), Status> {
self.check_header(authorization(req))
}
}
/// The `authorization` metadata of a request, if present and ASCII.
pub fn authorization<T>(req: &Request<T>) -> Option<&str> {
req.metadata().get("authorization").and_then(|v| v.to_str().ok())
}
/// The per-call tonic interceptor. Installed on the Flight SQL service so
/// every RPC (the handshake included, which carries Basic) is checked once.
pub fn interceptor(auth: Arc<DoorAuth>) -> impl Fn(Request<()>) -> Result<Request<()>, Status> + Clone {
move |req: Request<()>| {
auth.check_request(&req)?;
Ok(req)
}
}
/// The `Handshake` response stream type every Flight service names.
pub type HandshakeStream = Pin<Box<dyn Stream<Item = Result<HandshakeResponse, Status>> + Send>>;
/// The handshake answer: an `authorization: Bearer <token>` response header
/// and NO `HandshakeResponse` message. An empty token (open server) issues the
/// fixed placeholder `open`, which an open server's check ignores.
///
/// ALWAYS a header: the C++ driver takes the bearer from the response metadata
/// and, finding none, pushes an EMPTY authorization header onto every later
/// call.
///
/// NEVER a message. MEASURED 2026-09-09 on Windows 11 with Dremio's Arrow
/// Flight SQL ODBC driver (build 2025-07-24): it vendors an arrow-cpp whose
/// AuthenticateInternal calls WritesDone() then Finish() WITHOUT draining the
/// read side (upstream added the drain later: "otherwise gRPC Finish() will
/// hang"). A queued HandshakeResponse hangs that client until its deadline —
/// `Deadline Exceeded` on connect, before a single query. Reproduced on both
/// skade-flight and graphar-flight; a server-side drain of the request stream
/// did NOT help. Nothing needs the payload: arrow-cpp reads the bearer from
/// the response header, arrow-rs stores its token from the same header, and
/// arrow-go drains to EOF and reads the header.
pub fn handshake_response(issued_token: &str) -> Result<Response<HandshakeStream>, Status> {
let token = if issued_token.is_empty() { "open" } else { issued_token };
let out = futures::stream::empty::<Result<HandshakeResponse, Status>>();
let mut resp: Response<HandshakeStream> = Response::new(Box::pin(out));
let bearer = format!("Bearer {token}")
.parse()
.map_err(|_| Status::internal("token not header-safe"))?;
resp.metadata_mut().insert("authorization", bearer);
Ok(resp)
}
#[cfg(test)]
mod tests {
use super::*;
use base64::engine::general_purpose::{STANDARD, STANDARD_NO_PAD};
use futures::StreamExt;
fn basic(u: &str, p: &str) -> String {
format!("Basic {}", STANDARD.encode(format!("{u}:{p}")))
}
#[test]
fn open_accepts_anything() {
let a = DoorAuth::open();
assert!(a.check_header(None).is_ok());
assert!(a.check_header(Some("Bearer nope")).is_ok());
assert!(a.check_header(Some("garbage")).is_ok());
}
#[test]
fn bearer_accepts_and_rejects() {
let a = DoorAuth::bearer("s3cret");
assert!(a.check_header(Some("Bearer s3cret")).is_ok());
assert!(a.check_header(Some("bearer s3cret")).is_ok());
assert!(a.check_header(Some("Bearer wrong")).is_err());
assert!(a.check_header(None).is_err());
// Basic without a configured pair is refused, not a bypass.
assert!(a.check_header(Some(&basic("x", "y"))).is_err());
}
/// RED direction: decode with the strict STANDARD engine again and the
/// unpadded arm fails.
#[test]
fn basic_padded_and_unpadded() {
let a = DoorAuth::bearer("s3cret").with_basic("admin", "pw");
let padded = basic("admin", "pw");
assert!(padded.ends_with('='), "precondition: `admin:pw` pads to {padded}");
assert!(a.check_header(Some(&padded)).is_ok());
let unpadded = format!("Basic {}", STANDARD_NO_PAD.encode("admin:pw"));
assert_ne!(padded, unpadded);
assert!(a.check_header(Some(&unpadded)).is_ok());
assert!(a.check_header(Some(&basic("admin", "wrong"))).is_err());
assert!(a.check_header(Some(&basic("nope", "pw"))).is_err());
assert!(a.check_header(Some("Basic !!!notbase64")).is_err());
assert_eq!(a.issued_token(), "s3cret");
}
#[test]
fn decode_basic_splits_on_first_colon() {
let b64 = STANDARD.encode("pbi:pass:with:colons");
assert_eq!(decode_basic(&b64), Some(("pbi".into(), "pass:with:colons".into())));
assert_eq!(decode_basic(&STANDARD.encode("nocolon")), None);
}
/// RED direction: queue one HandshakeResponse and the stream is no longer
/// empty — the Dremio ODBC driver build hangs on exactly that.
#[tokio::test]
async fn handshake_is_header_only() {
let resp = handshake_response("issued-token").unwrap();
assert_eq!(resp.metadata().get("authorization").unwrap(), "Bearer issued-token");
let mut stream = resp.into_inner();
assert!(stream.next().await.is_none(), "handshake must send NO message");
let open = handshake_response("").unwrap();
assert_eq!(open.metadata().get("authorization").unwrap(), "Bearer open");
}
}