graphar-flight 0.1.2

Apache Arrow Flight SQL service over FalkorDB — Cypher in, Arrow out
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
//! 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;

use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64;
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 constant_time_eq(token.as_bytes(), expected.as_bytes()) {
                    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 decoded = BASE64.decode(b64).ok()?;
        let text = std::str::from_utf8(&decoded).ok()?;
        let (user, pass) = text.split_once(':')?;
        // Compare both fields in constant time; `&` (not `&&`) so timing does
        // not reveal which field differed.
        let ok = constant_time_eq(user.as_bytes(), expected.username.as_bytes())
            & constant_time_eq(pass.as_bytes(), expected.password.as_bytes());
        ok.then(|| user.to_string())
    }
}

/// Length-aware constant-time byte comparison.
pub(crate) fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
}

#[cfg(test)]
mod tests {
    use super::*;

    fn basic_header(user: &str, pass: &str) -> String {
        format!("Basic {}", BASE64.encode(format!("{user}:{pass}")))
    }

    #[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"));
    }
}