crates/graphar-flight/src/auth.rs
//! Authentication and authorization for the Cypher Flight SQL server.
//!
//! ## Authentication — *who are you?*
//!
//! Three methods, any combination, to cover the clients that matter:
//!
//! - **Bearer token** — a shared secret in `authorization: Bearer <token>`.
//! More than one token can be valid at once (a **token set**) so a secret can
//! be **rotated without downtime**: add the new token, let clients migrate,
//! then drop the old one — both are accepted during the overlap window.
//! - **Basic (username / password)** — `authorization: Basic <base64>`. This
//! is what generic Flight SQL clients on Windows use: Power BI's ADBC
//! connector and the Arrow Flight SQL ODBC driver send username/password,
//! first through the Flight `Handshake` (which issues a bearer token) and
//! then on every call.
//! - **mutual TLS** — client-certificate auth, configured at the transport in
//! [`crate::server`] (`TlsOptions::client_ca_pem`); orthogonal to the
//! header methods here.
//!
//! The same [`AuthConfig`] drives both the per-call interceptor and the
//! handshake, so the two can never disagree about who is allowed in.
//!
//! ## Authorization — *what may you run?*
//!
//! Authentication answers *who*; the [`Authorizer`] answers *what*. After a
//! query is resolved to its registered table name, the authorizer is consulted
//! with the caller's [`Identity`] and may deny it (`permission_denied`). The
//! default is [`Authorizer::AllowAll`] — no behavior change — but
//! [`Authorizer::AllowList`] restricts each identity to a named subset of
//! registered queries, and [`Authorizer::Custom`] takes an arbitrary closure.
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
// The Basic decoder (padding optional) and the constant-time compare are the
// ONE copy in `nornir_flight::door`, shared with skade-flight. This module keeps
// only what is graphar's own: token sets, identities and the authorizer.
use nornir_flight::door::{decode_basic, token_matches};
use tonic::Status;
/// A username / password pair accepted via Basic auth.
#[derive(Debug, Clone)]
pub struct BasicCredential {
pub username: String,
pub password: String,
}
/// The authenticated caller, as resolved from the `authorization` header.
///
/// Carried as a request extension by the interceptor (and returned by
/// [`AuthConfig::authenticate`]) so the per-query [`Authorizer`] can key its
/// decision on *who* is asking. The variants name the credential method:
///
/// - [`Identity::Token`] — a bearer token authenticated; the string is the
/// token's mapped name (from [`AuthConfig::with_named_bearer_tokens`]) or the
/// token value itself when unnamed.
/// - [`Identity::Basic`] — Basic auth authenticated; the string is the username.
/// - [`Identity::Anonymous`] — an open server with no auth configured.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Identity {
/// A bearer token authenticated; carries the token's identity name.
Token(String),
/// Basic auth authenticated; carries the username.
Basic(String),
/// No authentication configured (open server).
Anonymous,
}
impl Identity {
/// The identity's name for authz lookups: the token-identity name, the
/// Basic username, or `"*"` for the anonymous (open-server) identity.
pub fn name(&self) -> &str {
match self {
Identity::Token(name) | Identity::Basic(name) => name,
Identity::Anonymous => "*",
}
}
}
/// Per-query authorization hook, consulted *after* a query resolves to its
/// registered name. Dependency-light by design — no external policy engine.
///
/// The default [`Authorizer::AllowAll`] permits everything (so wiring it in is
/// a no-op for existing servers). [`Authorizer::AllowList`] keys on
/// [`Identity::name`] and permits only the named queries listed for that
/// identity. [`Authorizer::Custom`] defers to a closure for arbitrary policy.
#[derive(Clone, Default)]
pub enum Authorizer {
/// Permit every identity to run every query (the default).
#[default]
AllowAll,
/// Permit each identity only the registered-query names listed for it.
/// An identity absent from the map is denied every query; the special key
/// `"*"` is consulted as a fallback for identities not otherwise listed.
AllowList(HashMap<String, HashSet<String>>),
/// Arbitrary policy: `(identity, resolved_query_name) -> allowed`.
Custom(Arc<dyn Fn(&Identity, &str) -> bool + Send + Sync>),
}
impl std::fmt::Debug for Authorizer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Authorizer::AllowAll => f.write_str("Authorizer::AllowAll"),
Authorizer::AllowList(m) => f.debug_tuple("Authorizer::AllowList").field(m).finish(),
Authorizer::Custom(_) => f.write_str("Authorizer::Custom(<fn>)"),
}
}
}
impl Authorizer {
/// Build an [`Authorizer::AllowList`] from `(identity_name, &[query_name])`
/// pairs. Each identity is permitted exactly the listed registered-query
/// names (table names from `register_named`, or the Cypher key otherwise).
pub fn allow_list<I, S1, S2>(entries: I) -> Self
where
I: IntoIterator<Item = (S1, Vec<S2>)>,
S1: Into<String>,
S2: Into<String>,
{
let map = entries
.into_iter()
.map(|(id, queries)| (id.into(), queries.into_iter().map(Into::into).collect()))
.collect();
Authorizer::AllowList(map)
}
/// Wrap a closure as an [`Authorizer::Custom`].
pub fn custom<F>(f: F) -> Self
where
F: Fn(&Identity, &str) -> bool + Send + Sync + 'static,
{
Authorizer::Custom(Arc::new(f))
}
/// Is `identity` allowed to run the query resolved to `query_name`?
pub fn is_allowed(&self, identity: &Identity, query_name: &str) -> bool {
match self {
Authorizer::AllowAll => true,
Authorizer::AllowList(map) => {
let allowed = |key: &str| map.get(key).is_some_and(|qs| qs.contains(query_name));
// The identity's own allowances, then the wildcard fallback.
allowed(identity.name()) || allowed("*")
}
Authorizer::Custom(f) => f(identity, query_name),
}
}
/// Authorize or reject, mapping a denial to `permission_denied`.
pub fn authorize(&self, identity: &Identity, query_name: &str) -> Result<(), Status> {
if self.is_allowed(identity, query_name) {
Ok(())
} else {
Err(Status::permission_denied(format!(
"identity '{}' is not permitted to run query '{query_name}'",
identity.name()
)))
}
}
}
/// Which credentials the server accepts. An empty config (the default) means
/// the server is open — no interceptor is installed.
#[derive(Debug, Clone, Default)]
pub struct AuthConfig {
/// Accepted `Bearer` tokens, keyed by token value → identity name. Any
/// token in the set authenticates (rotation: old + new both valid). The
/// first inserted token is the one issued by the handshake.
bearer_tokens: Vec<(String, String)>,
/// Accepted Basic username/password, if any.
pub basic: Option<BasicCredential>,
/// Per-query authorization policy, consulted after a query resolves. The
/// default [`Authorizer::AllowAll`] is a no-op (no behavior change).
pub authorizer: Authorizer,
}
impl AuthConfig {
/// A single accepted bearer token (back-compat shape). Equivalent to
/// [`with_bearer_tokens`] with one element; the token names itself.
///
/// [`with_bearer_tokens`]: AuthConfig::with_bearer_tokens
pub fn with_bearer_token(mut self, token: impl Into<String>) -> Self {
self.add_bearer_token(token);
self
}
/// A **set** of accepted bearer tokens for zero-downtime rotation. Any
/// listed token authenticates; the first is issued by the handshake. Each
/// token names itself as its [`Identity`] (use
/// [`with_named_bearer_tokens`] to map tokens to authz identities).
///
/// [`with_named_bearer_tokens`]: AuthConfig::with_named_bearer_tokens
pub fn with_bearer_tokens<I, S>(mut self, tokens: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
for t in tokens {
self.add_bearer_token(t);
}
self
}
/// A set of accepted bearer tokens mapped to authz **identity names**, so
/// several tokens can share one identity (e.g. an old and a new token for
/// the same `analyst` role rotate transparently). Pairs are `(token,
/// identity_name)`; the first pair's token is issued by the handshake.
pub fn with_named_bearer_tokens<I, S1, S2>(mut self, tokens: I) -> Self
where
I: IntoIterator<Item = (S1, S2)>,
S1: Into<String>,
S2: Into<String>,
{
for (token, name) in tokens {
self.add_named_bearer_token(token, name);
}
self
}
/// Add one accepted bearer token (self-named). Keeps any already added —
/// the building block for rotation. Returns `&mut Self` for chaining.
pub fn add_bearer_token(&mut self, token: impl Into<String>) -> &mut Self {
let token = token.into();
let name = token.clone();
self.add_named_bearer_token(token, name)
}
/// Add one accepted bearer token mapped to an authz identity name.
pub fn add_named_bearer_token(
&mut self,
token: impl Into<String>,
name: impl Into<String>,
) -> &mut Self {
self.bearer_tokens.push((token.into(), name.into()));
self
}
/// Accept Basic `username` / `password`.
pub fn with_basic(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
self.basic = Some(BasicCredential {
username: username.into(),
password: password.into(),
});
self
}
/// Attach a per-query [`Authorizer`]. Default is [`Authorizer::AllowAll`].
pub fn with_authorizer(mut self, authorizer: Authorizer) -> Self {
self.authorizer = authorizer;
self
}
/// Authorize an already-resolved query for an identity (delegates to the
/// configured [`Authorizer`]). Denial → `permission_denied`.
pub fn authorize(&self, identity: &Identity, query_name: &str) -> Result<(), Status> {
self.authorizer.authorize(identity, query_name)
}
/// All accepted bearer token values, in insertion order.
pub fn bearer_tokens(&self) -> impl Iterator<Item = &str> {
self.bearer_tokens.iter().map(|(t, _)| t.as_str())
}
/// Any credential method configured? When false, the server requires no
/// authentication and the interceptor is skipped entirely.
pub fn is_enabled(&self) -> bool {
!self.bearer_tokens.is_empty() || self.basic.is_some()
}
/// The token issued to a client after a successful handshake — the first
/// configured bearer token, which the client then replays on each call.
pub fn issued_token(&self) -> Option<&str> {
self.bearer_tokens.first().map(|(t, _)| t.as_str())
}
/// Validate an `authorization` header value against every configured
/// method. Open servers accept anything (including a missing header).
///
/// A thin wrapper over [`authenticate`](Self::authenticate) that discards
/// the resolved [`Identity`]; kept for call sites that only gate access.
pub fn check_header(&self, header: Option<&str>) -> Result<(), Status> {
self.authenticate(header).map(|_| ())
}
/// Validate an `authorization` header and return the resolved [`Identity`].
///
/// Open servers (no method configured) authenticate anything as
/// [`Identity::Anonymous`]. Otherwise a valid `Bearer` token resolves to its
/// mapped [`Identity::Token`] and valid Basic creds to [`Identity::Basic`];
/// anything else is `unauthenticated`.
pub fn authenticate(&self, header: Option<&str>) -> Result<Identity, Status> {
if !self.is_enabled() {
return Ok(Identity::Anonymous);
}
let value = header.ok_or_else(|| {
Status::unauthenticated(
"missing 'authorization' header (Bearer token or Basic credentials)",
)
})?;
if let Some(token) = value.strip_prefix("Bearer ") {
// Constant-time compare against every accepted token (rotation).
for (expected, name) in &self.bearer_tokens {
if token_matches(token, expected) {
return Ok(Identity::Token(name.clone()));
}
}
} else if let Some(user) = value
.strip_prefix("Basic ")
.and_then(|b64| self.check_basic(b64))
{
return Ok(Identity::Basic(user));
}
Err(Status::unauthenticated("invalid credentials"))
}
/// Validate a base64-encoded `user:pass` Basic payload, returning the
/// username on success.
pub fn check_basic(&self, b64: &str) -> Option<String> {
let expected = self.basic.as_ref()?;
let (user, pass) = decode_basic(b64)?;
// Compare both fields in constant time; `&` (not `&&`) so timing does
// not reveal which field differed.
let ok = token_matches(&user, &expected.username) & token_matches(&pass, &expected.password);
ok.then_some(user)
}
}
#[cfg(test)]
mod tests {
use super::*;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64;
fn basic_header(user: &str, pass: &str) -> String {
format!("Basic {}", BASE64.encode(format!("{user}:{pass}")))
}
/// **An UNPADDED Basic payload authenticates the same user.** RFC 4648
/// wants the `=`; a client in the Flight SQL comparison did not send it
/// and was refused as "invalid credentials". RED direction: decode with
/// the strict STANDARD engine again and the unpadded arm fails.
#[test]
fn basic_payload_is_accepted_with_or_without_padding() {
let auth = AuthConfig::default().with_basic("admin", "pw");
let padded = basic_header("admin", "pw");
assert!(padded.ends_with('='), "precondition: `admin:pw` pads to {padded}");
let unpadded = padded.trim_end_matches('=').to_string();
assert_ne!(padded, unpadded);
assert_eq!(
auth.authenticate(Some(&unpadded)).unwrap(),
Identity::Basic("admin".into()),
"the bytes are the same bytes"
);
assert_eq!(auth.authenticate(Some(&padded)).unwrap(), Identity::Basic("admin".into()));
// Still a decode, not a bypass: garbage stays refused, wrong password stays refused.
assert!(auth.check_header(Some("Basic !!!notbase64")).is_err());
let wrong = basic_header("admin", "wrong").trim_end_matches('=').to_string();
assert!(auth.check_header(Some(&wrong)).is_err());
}
#[test]
fn open_server_accepts_anything() {
let auth = AuthConfig::default();
assert!(!auth.is_enabled());
assert!(auth.check_header(None).is_ok());
assert!(auth.check_header(Some("garbage")).is_ok());
assert_eq!(auth.authenticate(None).unwrap(), Identity::Anonymous);
}
#[test]
fn bearer_token_accepted_and_rejected() {
let auth = AuthConfig::default().with_bearer_token("s3cret");
assert!(auth.check_header(Some("Bearer s3cret")).is_ok());
assert!(auth.check_header(Some("Bearer nope")).is_err());
assert!(auth.check_header(None).is_err());
assert_eq!(
auth.authenticate(Some("Bearer s3cret")).unwrap(),
Identity::Token("s3cret".into())
);
}
#[test]
fn multiple_tokens_all_accepted_for_rotation() {
// Old + new token both valid during the overlap window.
let auth = AuthConfig::default().with_bearer_tokens(["old-tok", "new-tok"]);
assert!(auth.check_header(Some("Bearer old-tok")).is_ok());
assert!(auth.check_header(Some("Bearer new-tok")).is_ok());
// A stale/never-valid token is still rejected.
assert!(auth.check_header(Some("Bearer retired-tok")).is_err());
// The handshake issues the first (primary) token.
assert_eq!(auth.issued_token(), Some("old-tok"));
}
#[test]
fn add_bearer_token_grows_the_set() {
let mut auth = AuthConfig::default().with_bearer_token("t1");
auth.add_bearer_token("t2");
assert!(auth.check_header(Some("Bearer t1")).is_ok());
assert!(auth.check_header(Some("Bearer t2")).is_ok());
assert_eq!(auth.bearer_tokens().collect::<Vec<_>>(), vec!["t1", "t2"]);
}
#[test]
fn named_tokens_share_an_identity() {
// Two tokens (rotation) that both authenticate as "analyst".
let auth = AuthConfig::default()
.with_named_bearer_tokens([("tok-a", "analyst"), ("tok-b", "analyst")]);
assert_eq!(
auth.authenticate(Some("Bearer tok-a")).unwrap(),
Identity::Token("analyst".into())
);
assert_eq!(
auth.authenticate(Some("Bearer tok-b")).unwrap(),
Identity::Token("analyst".into())
);
}
#[test]
fn basic_credentials_accepted_and_rejected() {
let auth = AuthConfig::default().with_basic("admin", "pw");
assert!(
auth.check_header(Some(&basic_header("admin", "pw")))
.is_ok()
);
assert!(
auth.check_header(Some(&basic_header("admin", "wrong")))
.is_err()
);
assert!(auth.check_header(Some(&basic_header("eve", "pw"))).is_err());
assert!(auth.check_header(Some("Basic !!!notbase64")).is_err());
assert_eq!(
auth.authenticate(Some(&basic_header("admin", "pw")))
.unwrap(),
Identity::Basic("admin".into())
);
}
#[test]
fn both_methods_accepted_when_both_configured() {
let auth = AuthConfig::default()
.with_bearer_token("tok")
.with_basic("u", "p");
assert!(auth.check_header(Some("Bearer tok")).is_ok());
assert!(auth.check_header(Some(&basic_header("u", "p"))).is_ok());
assert!(auth.check_header(Some("Bearer bad")).is_err());
}
#[test]
fn issued_token_is_the_first_bearer_token() {
let auth = AuthConfig::default().with_bearer_tokens(["tok", "tok2"]);
assert_eq!(auth.issued_token(), Some("tok"));
assert_eq!(AuthConfig::default().issued_token(), None);
}
#[test]
fn authorizer_allow_all_permits_everything() {
let authz = Authorizer::AllowAll;
assert!(authz.is_allowed(&Identity::Token("anyone".into()), "q1"));
assert!(authz.is_allowed(&Identity::Anonymous, "q2"));
assert!(
authz
.authorize(&Identity::Basic("u".into()), "anything")
.is_ok()
);
}
#[test]
fn authorizer_allow_list_allows_and_denies_per_identity() {
// analyst may run q1 but not q2.
let authz = Authorizer::allow_list([("analyst", vec!["q1"])]);
let analyst = Identity::Token("analyst".into());
assert!(authz.is_allowed(&analyst, "q1"));
assert!(!authz.is_allowed(&analyst, "q2"));
assert!(authz.authorize(&analyst, "q1").is_ok());
let denied = authz.authorize(&analyst, "q2").unwrap_err();
assert_eq!(denied.code(), tonic::Code::PermissionDenied);
// An identity absent from the list is denied everything.
assert!(!authz.is_allowed(&Identity::Token("stranger".into()), "q1"));
}
#[test]
fn authorizer_allow_list_wildcard_fallback() {
// "*" grants q_public to every identity not otherwise listed.
let authz = Authorizer::allow_list([("analyst", vec!["q1"]), ("*", vec!["q_public"])]);
let analyst = Identity::Token("analyst".into());
assert!(authz.is_allowed(&analyst, "q1"));
assert!(authz.is_allowed(&analyst, "q_public")); // wildcard fallback
let other = Identity::Token("other".into());
assert!(authz.is_allowed(&other, "q_public"));
assert!(!authz.is_allowed(&other, "q1"));
}
#[test]
fn authorizer_custom_closure() {
let authz = Authorizer::custom(|id, q| id.name() == "root" || q == "q_open");
assert!(authz.is_allowed(&Identity::Token("root".into()), "anything"));
assert!(authz.is_allowed(&Identity::Token("nobody".into()), "q_open"));
assert!(!authz.is_allowed(&Identity::Token("nobody".into()), "q_secret"));
}
}