skade-flight/src/ticket.rs
//! **Opaque, signed statement handles** — the bytes inside a
//! `TicketStatementQuery.statement_handle` and a
//! `CommandPreparedStatementQuery.prepared_statement_handle`.
//!
//! They used to be the raw SQL text. That is stateless and replica-friendly,
//! and it is also replayable by anything that has seen the bytes, legible on
//! the path, and usable by a principal it was not issued to. A Flight SQL
//! handle is a capability the server minted for one caller; it should look
//! like one.
//!
//! A handle is a [`Claims`] envelope — the statement (SQL + bound parameter
//! values), the principal it was issued to, when, and when it expires — SEALED
//! with XChaCha20-Poly1305 (authenticated encryption: nothing on the path can
//! read it, and nothing can change it). The server keeps NO table: any replica
//! holding the same key opens any handle, so statelessness is kept; a byte
//! flipped anywhere, a handle past its expiry, or a handle presented by
//! another principal is `UNAUTHENTICATED`, never executed.
//!
//! The key is derived from the server's bearer secret when it has one (so
//! replicas sharing a token share handles) and random per process otherwise
//! (an open server's handles die with the process — nothing else has any
//! business holding them).
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use arrow_array::RecordBatch;
use bytes::{Buf, BufMut, Bytes, BytesMut};
use chacha20poly1305::aead::{Aead, KeyInit, Payload};
use chacha20poly1305::{XChaCha20Poly1305, XNonce};
use datafusion::scalar::ScalarValue;
use sha2::{Digest, Sha256};
use tonic::Status;
use crate::engine::Statement;
const VERSION: u8 = 1;
const NONCE_LEN: usize = 24;
const TAG_LEN: usize = 16;
/// Bound to the ciphertext as associated data: a handle sealed under another
/// version never opens as this one.
const AAD: &[u8] = b"skade-flight statement handle v1";
/// Default lifetime of a statement ticket: the `DoGet` follows its
/// `GetFlightInfo` within the same refresh, in practice within seconds; an
/// hour covers a paused BI refresh without leaving a capability alive all day.
pub const TICKET_TTL: Duration = Duration::from_secs(60 * 60);
/// Default lifetime of a prepared-statement handle: a driver keeps one open
/// for a session, which for a dashboard host is the working day.
pub const PREPARED_TTL: Duration = Duration::from_secs(24 * 60 * 60);
/// What a handle asserts.
#[derive(Debug, Clone, PartialEq)]
pub struct Claims {
pub statement: Statement,
/// Who the handle was issued to (see [`principal_of_header`]).
pub principal: String,
pub issued_at: u64,
pub expires_at: u64,
}
/// Mints and opens handles with one key.
#[derive(Clone)]
pub struct Signer {
key: [u8; 32],
}
impl std::fmt::Debug for Signer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Signer").finish_non_exhaustive()
}
}
impl Signer {
/// A key every replica configured with the same bearer secret derives
/// identically — so a handle minted by one is honoured by another.
pub fn from_secret(secret: &str) -> Self {
let mut h = Sha256::new();
h.update(b"skade-flight statement handle v1\0");
h.update(secret.as_bytes());
Self { key: h.finalize().into() }
}
/// A key nobody else has: this process's handles are its own.
pub fn random() -> Self {
let mut key = [0u8; 32];
getrandom::fill(&mut key).expect("the OS random source is available");
Self { key }
}
/// The signer a server with this bearer secret uses (`""` = open server).
pub fn for_secret(secret: &str) -> Self {
if secret.is_empty() { Self::random() } else { Self::from_secret(secret) }
}
pub fn sign(&self, claims: &Claims) -> Result<Bytes, Status> {
let mut plain = BytesMut::new();
plain.put_u64(claims.issued_at);
plain.put_u64(claims.expires_at);
put_bytes(&mut plain, claims.principal.as_bytes());
put_bytes(&mut plain, claims.statement.sql.as_bytes());
put_bytes(&mut plain, &encode_params(&claims.statement.params)?);
let mut nonce = [0u8; NONCE_LEN];
getrandom::fill(&mut nonce).map_err(|e| Status::internal(format!("random nonce: {e}")))?;
let cipher = XChaCha20Poly1305::new((&self.key).into());
let sealed = cipher
.encrypt(XNonce::from_slice(&nonce), Payload { msg: &plain, aad: AAD })
.map_err(|_| Status::internal("seal statement handle"))?;
let mut out = BytesMut::with_capacity(1 + NONCE_LEN + sealed.len());
out.put_u8(VERSION);
out.put_slice(&nonce);
out.put_slice(&sealed);
Ok(out.freeze())
}
/// Mint a handle for `statement`, issued to `principal`, valid for `ttl` from now.
pub fn issue(&self, statement: Statement, principal: &str, ttl: Duration) -> Result<Bytes, Status> {
let now = now_secs();
self.sign(&Claims { statement, principal: principal.to_string(), issued_at: now, expires_at: now + ttl.as_secs() })
}
/// Open a handle presented by `principal` at the current time.
pub fn verify(&self, handle: &[u8], principal: &str) -> Result<Claims, Status> {
self.verify_at(handle, principal, now_secs())
}
/// [`Self::verify`] with the clock passed in (pure; what the tests drive).
pub fn verify_at(&self, handle: &[u8], principal: &str, now: u64) -> Result<Claims, Status> {
let unauth = |why: &str| Status::unauthenticated(format!("statement handle refused: {why}"));
if handle.len() < 1 + NONCE_LEN + TAG_LEN + 8 + 8 + 4 * 3 {
return Err(unauth("too short to be one this server issued"));
}
if handle[0] != VERSION {
return Err(unauth("unknown handle version"));
}
let (nonce, sealed) = handle[1..].split_at(NONCE_LEN);
// The AEAD tag decides before anything is parsed: a foreign key, a
// flipped byte or a different version's AAD all fail here, identically.
let cipher = XChaCha20Poly1305::new((&self.key).into());
let plain = cipher
.decrypt(XNonce::from_slice(nonce), Payload { msg: sealed, aad: AAD })
.map_err(|_| unauth("not issued by this server (seal does not open)"))?;
let mut b = plain.as_slice();
if b.len() < 16 {
return Err(unauth("truncated"));
}
let issued_at = b.get_u64();
let expires_at = b.get_u64();
let principal_bytes = get_bytes(&mut b).ok_or_else(|| unauth("truncated"))?;
let sql = get_bytes(&mut b).ok_or_else(|| unauth("truncated"))?;
let params = get_bytes(&mut b).ok_or_else(|| unauth("truncated"))?;
let issued_to = String::from_utf8(principal_bytes.to_vec()).map_err(|_| unauth("principal not UTF-8"))?;
if issued_to != principal {
return Err(unauth("issued to a different principal"));
}
if now >= expires_at {
return Err(unauth("expired — ask for the statement again (GetFlightInfo)"));
}
let sql = String::from_utf8(sql.to_vec()).map_err(|_| unauth("statement not UTF-8"))?;
let params = decode_params(params)?;
Ok(Claims { statement: Statement { sql, params }, principal: issued_to, issued_at, expires_at })
}
}
fn put_bytes(buf: &mut BytesMut, b: &[u8]) {
buf.put_u32(b.len() as u32);
buf.put_slice(b);
}
fn get_bytes<'a>(b: &mut &'a [u8]) -> Option<&'a [u8]> {
if b.len() < 4 {
return None;
}
let n = b.get_u32() as usize;
if b.len() < n {
return None;
}
let (head, rest) = b.split_at(n);
*b = rest;
Some(head)
}
fn now_secs() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
}
/// Bound parameter values as one one-row Arrow IPC stream (`$1`, `$2`, … as
/// column names) — Arrow is the wire's own type system, so a bound value
/// round-trips with its type and nothing is stringified.
pub fn encode_params(params: &[ScalarValue]) -> Result<Vec<u8>, Status> {
if params.is_empty() {
return Ok(Vec::new());
}
let batch = params_batch(params)?;
let mut out = Vec::new();
{
let mut w = arrow_ipc::writer::StreamWriter::try_new(&mut out, batch.schema().as_ref())
.map_err(|e| Status::internal(format!("encode parameters: {e}")))?;
w.write(&batch).map_err(|e| Status::internal(format!("encode parameters: {e}")))?;
w.finish().map_err(|e| Status::internal(format!("encode parameters: {e}")))?;
}
Ok(out)
}
/// One row: column `$i` holds parameter `i`.
pub fn params_batch(params: &[ScalarValue]) -> Result<RecordBatch, Status> {
let mut fields = Vec::with_capacity(params.len());
let mut columns = Vec::with_capacity(params.len());
for (i, v) in params.iter().enumerate() {
let arr = v.to_array().map_err(|e| Status::internal(format!("parameter ${}: {e}", i + 1)))?;
fields.push(arrow_schema::Field::new(format!("${}", i + 1), arr.data_type().clone(), true));
columns.push(arr);
}
RecordBatch::try_new(std::sync::Arc::new(arrow_schema::Schema::new(fields)), columns)
.map_err(|e| Status::internal(format!("parameters batch: {e}")))
}
pub fn decode_params(bytes: &[u8]) -> Result<Vec<ScalarValue>, Status> {
if bytes.is_empty() {
return Ok(Vec::new());
}
let reader = arrow_ipc::reader::StreamReader::try_new(std::io::Cursor::new(bytes), None)
.map_err(|e| Status::unauthenticated(format!("statement handle refused: parameters: {e}")))?;
let mut out = Vec::new();
for batch in reader {
let batch = batch.map_err(|e| Status::unauthenticated(format!("statement handle refused: parameters: {e}")))?;
out.extend(params_from_batch(&batch)?);
}
Ok(out)
}
/// The bound values a client sent (`DoPutPreparedStatementQuery`): the first
/// row of the batch, one column per placeholder, in column order.
pub fn params_from_batch(batch: &RecordBatch) -> Result<Vec<ScalarValue>, Status> {
if batch.num_rows() == 0 {
return Ok(Vec::new());
}
batch
.columns()
.iter()
.map(|c| ScalarValue::try_from_array(c.as_ref(), 0).map_err(|e| Status::invalid_argument(format!("bind: {e}"))))
.collect()
}
/// **Who is calling** — from the `authorization` header a handle is bound to:
/// the Basic user name, `bearer` for a shared bearer token, empty on an open
/// server. Coarse on purpose: the bearer is one shared secret, so every
/// bearer caller is the same principal; the Basic user is a name.
pub fn principal_of_header(header: Option<&str>) -> String {
use base64::Engine as _;
let Some(h) = header else { return String::new() };
if h.starts_with("Bearer ") || h.starts_with("bearer ") {
return "bearer".to_string();
}
if let Some(b64) = h.strip_prefix("Basic ").or_else(|| h.strip_prefix("basic ")) {
let decoded = base64::engine::general_purpose::STANDARD
.decode(b64.trim())
.or_else(|_| base64::engine::general_purpose::STANDARD_NO_PAD.decode(b64.trim()));
if let Ok(bytes) = decoded {
if let Ok(s) = std::str::from_utf8(&bytes) {
return s.split_once(':').map(|(u, _)| u).unwrap_or(s).to_string();
}
}
return "basic".to_string();
}
String::new()
}
#[cfg(test)]
mod tests {
use super::*;
/// Known-answer test (2026-10-02, W3 crypto bump): a handle sealed by the
/// PRE-bump stack (chacha20poly1305 0.10 / sha2 0.10 / getrandom 0.3) under a
/// key derived from a fixed secret still opens to the same claims on every
/// later stack, so a rolling upgrade does not refuse handles minted a second
/// before it. Regenerate ONLY on purpose:
/// `SKADE_TICKET_KAT_REGEN=1 cargo test -p skade-flight ticket_kat`.
#[test]
fn ticket_kat_old_handle_opens() {
let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("testdata/ticket-kat/handle.bin");
let signer = Signer::from_secret("skade ticket kat secret");
if std::env::var_os("SKADE_TICKET_KAT_REGEN").is_some() {
let claims = Claims { statement: stmt(), principal: "kat".into(), issued_at: 1_000, expires_at: 4_000_000_000 };
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
std::fs::write(&path, signer.sign(&claims).unwrap()).unwrap();
}
let got = signer.verify_at(&std::fs::read(&path).unwrap(), "kat", 2_000).unwrap();
assert_eq!((got.statement.sql.as_str(), got.principal.as_str(), got.issued_at, got.expires_at),
(stmt().sql.as_str(), "kat", 1_000, 4_000_000_000));
}
fn stmt() -> Statement {
Statement { sql: "SELECT * FROM people WHERE id = $1".into(), params: vec![ScalarValue::Int64(Some(2))] }
}
/// **A handle round-trips exactly, and is opaque**: the SQL is not legible
/// in the bytes, the whole envelope (parameters included) comes back equal,
/// and a foreign key does not open it.
#[test]
fn a_handle_round_trips_and_a_foreign_key_refuses_it() {
let s = Signer::from_secret("s3cret");
let h = s.issue(stmt(), "pbi", TICKET_TTL).unwrap();
assert!(!String::from_utf8_lossy(&h).contains("people"), "sealed, not signed-in-the-clear: {h:?}");
assert!(!h.windows(3).any(|w| w == b"pbi"), "the principal is sealed too");
let c = s.verify(&h, "pbi").unwrap();
assert_eq!(c.statement, stmt());
assert_eq!(c.principal, "pbi");
assert_eq!(c.expires_at - c.issued_at, TICKET_TTL.as_secs());
let other = Signer::from_secret("other");
let err = other.verify(&h, "pbi").unwrap_err();
assert_eq!(err.code(), tonic::Code::Unauthenticated, "{err}");
assert!(err.message().contains("not issued by this server"), "{err}");
// Two issues of the same claims are two different byte strings (fresh
// nonce), and both open.
let h2 = s.issue(stmt(), "pbi", TICKET_TTL).unwrap();
assert_ne!(h, h2);
assert_eq!(s.verify(&h2, "pbi").unwrap().statement, stmt());
// Two replicas with the same secret honour each other's handles.
assert!(Signer::from_secret("s3cret").verify(&h, "pbi").is_ok());
// Two random signers never do.
let r = Signer::random().issue(stmt(), "", TICKET_TTL).unwrap();
assert!(Signer::random().verify(&r, "").is_err());
}
/// **One flipped byte anywhere is UNAUTHENTICATED** — version, nonce,
/// ciphertext (SQL, parameters, expiry, principal) or tag. RED direction:
/// drop the AAD/tag check and the flipped-ciphertext cases open as garbage.
#[test]
fn a_tampered_handle_is_unauthenticated_wherever_the_byte_is() {
let s = Signer::from_secret("s3cret");
let h = s.issue(stmt(), "pbi", TICKET_TTL).unwrap();
for i in 0..h.len() {
let mut bad = h.to_vec();
bad[i] ^= 0x01;
let err = s.verify(&bad, "pbi").unwrap_err();
assert_eq!(err.code(), tonic::Code::Unauthenticated, "byte {i}: {err}");
}
}
/// **A handle is bound to its principal and to its expiry.**
#[test]
fn a_handle_is_refused_for_another_principal_and_after_expiry() {
let s = Signer::from_secret("s3cret");
let h = s.issue(stmt(), "pbi", Duration::from_secs(10)).unwrap();
let err = s.verify(&h, "mallory").unwrap_err();
assert!(err.message().contains("different principal"), "{err}");
let issued = s.verify(&h, "pbi").unwrap().issued_at;
assert!(s.verify_at(&h, "pbi", issued + 9).is_ok());
let err = s.verify_at(&h, "pbi", issued + 10).unwrap_err();
assert!(err.message().contains("expired"), "{err}");
}
#[test]
fn principal_is_the_basic_user_or_the_bearer_class() {
use base64::Engine as _;
let b = format!("Basic {}", base64::engine::general_purpose::STANDARD.encode("pbi:hunter2"));
assert_eq!(principal_of_header(Some(&b)), "pbi");
assert_eq!(principal_of_header(Some("Bearer t0k3n")), "bearer");
assert_eq!(principal_of_header(None), "");
}
#[test]
fn parameters_round_trip_typed() {
let params = vec![
ScalarValue::Int64(Some(7)),
ScalarValue::Utf8(Some("Umeå".into())),
ScalarValue::Float64(None),
ScalarValue::Date32(Some(20_000)),
];
let bytes = encode_params(¶ms).unwrap();
assert_eq!(decode_params(&bytes).unwrap(), params);
assert!(decode_params(&[]).unwrap().is_empty());
}
}