skade-flight/src/bin/skade-flight.rs
//! `skade-flight` — the standalone skade remote: Flight SQL over a warehouse.
//!
//! ```text
//! skade-flight --warehouse /var/lib/skade [--listen 0.0.0.0:10009] [--token T] [--basic USER:PASS]
//! skade-flight --nessie-uri https://nessie:19120/api/v2 --s3-bucket B --s3-endpoint URL (read-only)
//! skade-flight --unit | --install | --uninstall | --start | --stop | --status [--system] [--env K=V]…
//! ```
//!
//! Token ladder (nornir-service, the ONE ladder): `--token` → `$SKADE_FLIGHT_TOKEN`
//! → `<home>/.skade/token`. Empty everywhere ⇒ OPEN server (say so loudly).
//!
//! Lifecycle verbs render and install the systemd unit through skidbladnir
//! (LAW: a nordisk service installs through its own door, never a hand-copied
//! binary or a hand-written drop-in).
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context, Result, bail};
use skade::connect::CatalogConnector;
use skade::tls::ClientTls;
use skade_flight::engine::EngineOptions;
use skade_flight::{Auth, Engine, ServeOptions};
/// Where the server listens when neither `--listen` nor `$SKADE_FLIGHT_ADDR`
/// says: every interface, port 10009 — the service exposes itself.
const DEFAULT_LISTEN: &str = "0.0.0.0:10009";
const USAGE: &str = "\
skade-flight — Apache Arrow Flight SQL over an Iceberg catalog
Catalog (pick ONE; --warehouse-uri, --nessie-uri and --catalog-uri are exclusive):
--warehouse DIR skade warehouse directory (env SKADE_FLIGHT_WAREHOUSE; default ~/.skade/warehouse)
With --warehouse-uri this is the LOCAL redb catalog dir only.
--warehouse-uri U object-store warehouse root the writer used, e.g. s3://bucket/prefix
(feature `object-store`). Needs --s3-bucket, and --s3-endpoint for MinIO.
--nessie-uri URL Nessie API v2, http:// or https://, e.g. http://nessie:19120/api/v2
(env SKADE_FLIGHT_NESSIE_URI; feature `nessie`). READ-ONLY.
Token from SKADE_FLIGHT_NESSIE_TOKEN — never a flag; none set → no Authorization header.
--nessie-ref REF Nessie branch or tag (env SKADE_FLIGHT_NESSIE_REF; default main)
--catalog-uri URL Iceberg REST catalog, http:// or https://, e.g. http://nessie:19120/iceberg
(env SKADE_FLIGHT_CATALOG_URI; feature `rest-catalog`). READ-ONLY.
Token from SKADE_FLIGHT_CATALOG_TOKEN — never a flag.
--catalog-warehouse W the REST catalog's warehouse name (env SKADE_FLIGHT_CATALOG_WAREHOUSE)
--catalog-ref REF Nessie branch/tag behind the REST catalog (env SKADE_FLIGHT_CATALOG_REF)
--s3-bucket B --s3-endpoint URL --s3-region R --s3-virtual-hosted
where the table files are (with --warehouse-uri, --nessie-uri or --catalog-uri);
keys from SKADE_S3_ACCESS_KEY_ID|AWS_ACCESS_KEY_ID|MINIO_ACCESS_KEY and
SKADE_S3_SECRET_ACCESS_KEY|AWS_SECRET_ACCESS_KEY|MINIO_SECRET_KEY — never a flag
--namespace NS schema unqualified table names resolve in (env SKADE_FLIGHT_NAMESPACE; default main)
TLS, per leg. The mode is the URL scheme plus which files are given:
http:// → plain, https:// → tls, https:// + client cert and key → mtls. Each flag has a SKADE_FLIGHT_* env twin.
--s3-ca F --s3-client-cert F --s3-client-key F S3 / MinIO (SKADE_FLIGHT_S3_CA, …_S3_CLIENT_CERT, …_S3_CLIENT_KEY)
--nessie-ca F --nessie-client-cert F --nessie-client-key F Nessie (SKADE_FLIGHT_NESSIE_CA, …_NESSIE_CLIENT_CERT, …_NESSIE_CLIENT_KEY)
--catalog-ca F --catalog-client-cert F --catalog-client-key F REST catalog (SKADE_FLIGHT_CATALOG_CA, …_CATALOG_CLIENT_CERT, …_CATALOG_CLIENT_KEY)
--tls-cert F --tls-key F [--tls-client-ca F] Power BI side (feature `tls`; absent = plaintext)
A client cert without its key, or a key without its cert, is refused.
Server:
--listen ADDR bind address (env SKADE_FLIGHT_ADDR; default 0.0.0.0:10009 — every interface;
an OPEN server on a non-loopback address is warned about loudly)
--token T bearer token (env SKADE_FLIGHT_TOKEN, file ~/.skade/token); empty = OPEN
--basic USER:PASS Basic credential the Handshake exchanges for the token (env SKADE_FLIGHT_BASIC)
--refresh-secs N rebuild the session when older than N seconds (default 5)
--partitions N DataFusion target_partitions (default: cores)
--unit print the systemd unit --install writes
--install render + install + enable + start the unit via skidbladnir
--uninstall stop, disable, remove the unit (data root untouched unless --purge)
--start/--stop/--status
--system system scope (User=skade); default is --user scope
--env K=V (with --install, repeatable) operator env rendered into the unit
";
fn flag(args: &[String], name: &str) -> Option<String> {
let mut it = args.iter();
while let Some(a) = it.next() {
if a == name {
return it.next().cloned();
}
if let Some(v) = a.strip_prefix(&format!("{name}=")) {
return Some(v.to_string());
}
}
None
}
fn home() -> PathBuf {
std::env::var_os("HOME").map(PathBuf::from).unwrap_or_else(|| PathBuf::from("/"))
}
fn skade_home() -> PathBuf {
std::env::var_os("SKADE_FLIGHT_HOME").map(PathBuf::from).unwrap_or_else(|| home().join(".skade"))
}
/// The process environment, as [`connector_from_args`] reads it.
fn process_env(name: &str) -> Option<String> {
std::env::var(name).ok()
}
fn spec(args: &[String], addr: &str, warehouse: &std::path::Path) -> Result<skidbladnir::service::ServiceSpec> {
use skidbladnir::service::{AgentMode, ServiceSpec};
use skidbladnir::systemd::{RestartPolicy, Scope};
let system = args.iter().any(|a| a == "--system");
let exec = std::env::current_exe()
.map(|p| p.to_string_lossy().into_owned())
.unwrap_or_else(|_| "/usr/local/bin/skade-flight".into());
let mut spec = ServiceSpec::new("skade-flight", "skade-flight Flight SQL server", exec)
.scope(if system { Scope::System } else { Scope::User })
.working_dir(skade_home().to_string_lossy())
.env("SKADE_FLIGHT_ADDR", addr)
.env("SKADE_FLIGHT_WAREHOUSE", warehouse.to_string_lossy())
.restart(RestartPolicy::OnFailure)
.agent(AgentMode::Fat);
if system {
spec = spec.user("skade");
}
for (k, v) in skidbladnir::service::env_args_from_argv(args)? {
spec = spec.operator_env(k, v);
}
match skidbladnir::uninstall::DataRoot::from_env("SKADE_FLIGHT_DATA", skade_home()) {
Ok(dr) => spec = spec.data_root(&dr),
Err(e) => eprintln!("skade-flight: {e} — uninstall would erase nothing; not declaring a root"),
}
Ok(spec)
}
/// `--warehouse` → `$SKADE_FLIGHT_WAREHOUSE` → `~/.skade/warehouse`.
fn warehouse_dir(args: &[String], env: &dyn Fn(&str) -> Option<String>) -> PathBuf {
flag(args, "--warehouse")
.or_else(|| env("SKADE_FLIGHT_WAREHOUSE"))
.map(PathBuf::from)
.unwrap_or_else(|| skade_home().join("warehouse"))
}
#[tokio::main]
async fn main() -> Result<()> {
let args: Vec<String> = std::env::args().skip(1).collect();
if args.iter().any(|a| a == "-h" || a == "--help") {
print!("{USAGE}");
return Ok(());
}
let addr_str = flag(&args, "--listen")
.or_else(|| std::env::var("SKADE_FLIGHT_ADDR").ok())
.unwrap_or_else(|| DEFAULT_LISTEN.into());
let warehouse = warehouse_dir(&args, &process_env);
// ── lifecycle door ───────────────────────────────────────────────────
let want = |v: &str| args.iter().any(|a| a == v);
if want("--unit") || want("--install") || want("--uninstall") || want("--start") || want("--stop")
|| want("--status")
{
let lc = skidbladnir::service::Lifecycle::new(spec(&args, &addr_str, &warehouse)?);
if want("--unit") {
print!("{}", lc.unit_text());
} else if want("--install") {
lc.install().context("install")?;
lc.enable().context("enable")?;
lc.start().context("start")?;
println!("skade-flight: INSTALLED and started ({})", lc.unit());
} else if want("--uninstall") {
let r = lc.uninstall().context("uninstall")?;
println!("skade-flight: unit removed = {}, erased {} path(s)", r.unit_removed, r.erased.len());
} else if want("--start") {
lc.start()?;
} else if want("--stop") {
lc.stop()?;
} else if want("--status") {
println!("{}", if lc.status()? { "active" } else { "inactive" });
}
return Ok(());
}
// ── serve ────────────────────────────────────────────────────────────
let addr: SocketAddr = addr_str.parse().with_context(|| format!("--listen {addr_str}"))?;
let connector = connector_from_args(&args, &process_env)?;
let legs = tls_legs(&args, &process_env)?;
eprintln!(
"skade-flight: TLS — {}",
legs.iter().map(|(leg, mode)| format!("{leg}: {mode}")).collect::<Vec<_>>().join(", ")
);
let token_file = skade_home().join("token");
let (token, whence) = nornir_service::resolve_token_whence(
flag(&args, "--token").as_deref(),
"SKADE_FLIGHT_TOKEN",
Some(token_file.as_path()),
);
let mut auth = Auth::bearer(token.clone());
let basic = flag(&args, "--basic").or_else(|| std::env::var("SKADE_FLIGHT_BASIC").ok());
if let Some(b) = &basic {
let Some((u, p)) = b.split_once(':') else { bail!("--basic wants USER:PASS") };
auth = auth.with_basic(u, p);
}
if token.is_empty() {
eprintln!("skade-flight: no bearer token ({}) — serving OPEN", whence.describe());
} else {
eprintln!("skade-flight: bearer from {}", whence.describe());
}
if let Some(warning) = exposure_warning(&addr, token.is_empty() && basic.is_none()) {
eprintln!("{warning}");
}
let refresh = flag(&args, "--refresh-secs").and_then(|s| s.parse().ok()).unwrap_or(5u64);
let parts = flag(&args, "--partitions")
.and_then(|s| s.parse().ok())
.unwrap_or_else(|| std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1));
let mut opts =
EngineOptions { refresh_after: Duration::from_secs(refresh), target_partitions: parts, ..Default::default() };
if let Some(ns) = flag(&args, "--namespace").or_else(|| process_env("SKADE_FLIGHT_NAMESPACE")) {
opts.default_namespace = ns;
}
let namespace = opts.default_namespace.clone();
let engine = Arc::new(
Engine::with_connector(Arc::clone(&connector), opts)
.await
.with_context(|| format!("open {} catalog: {}", connector.kind(), connector.describe()))?,
);
let n = engine.tables().await?.len();
eprintln!(
"skade-flight: {} — {} — {n} table(s), namespace {namespace}{} — Flight SQL on {addr} (partitions {parts}, refresh {refresh}s)",
connector.kind(),
connector.describe(),
if engine.options().read_only { ", READ-ONLY" } else { "" },
);
#[allow(unused_mut)]
let mut serve_opts = ServeOptions::default();
#[cfg(feature = "tls")]
if let (Some(c), Some(k)) = (flag(&args, "--tls-cert"), flag(&args, "--tls-key")) {
serve_opts.tls = Some(skade_flight::server::TlsOptions {
cert_pem: std::fs::read(&c).with_context(|| c.clone())?,
key_pem: std::fs::read(&k).with_context(|| k.clone())?,
client_ca_pem: match flag(&args, "--tls-client-ca") {
Some(ca) => Some(std::fs::read(&ca).with_context(|| ca.clone())?),
None => None,
},
});
}
let shutdown = async {
let _ = tokio::signal::ctrl_c().await;
eprintln!("skade-flight: shutting down");
};
skade_flight::serve(engine, Arc::new(auth), addr, serve_opts, shutdown).await?;
Ok(())
}
/// **The ONE place flags become a catalog.** Every backend choice and every
/// refusal lives here; `main` only calls `Engine::with_connector` on the
/// result. `env` is the process environment in the binary and a fixed map in
/// the tests.
///
/// - `--nessie-uri` → Nessie API v2 (read-only)
/// - `--catalog-uri` → an Iceberg REST catalog (read-only)
/// - `--warehouse-uri` → skade's catalog on local disk, table files in S3
/// - none of them → skade's local warehouse directory
fn connector_from_args(args: &[String], env: &dyn Fn(&str) -> Option<String>) -> Result<Arc<dyn CatalogConnector>> {
let pick = |f: &str, e: &str| flag(args, f).or_else(|| env(e)).filter(|v| !v.is_empty());
for secret in ["--nessie-token", "--catalog-token"] {
if args.iter().any(|a| a == secret || a.starts_with(&format!("{secret}="))) {
let var = if secret == "--nessie-token" { "SKADE_FLIGHT_NESSIE_TOKEN" } else { "SKADE_FLIGHT_CATALOG_TOKEN" };
bail!("{secret} is not a flag: a token on the command line is readable by every user — set {var}");
}
}
let warehouse_uri = pick("--warehouse-uri", "SKADE_FLIGHT_WAREHOUSE_URI");
let nessie_uri = pick("--nessie-uri", "SKADE_FLIGHT_NESSIE_URI");
let catalog_uri = pick("--catalog-uri", "SKADE_FLIGHT_CATALOG_URI");
let chosen: Vec<&str> = [
("--warehouse-uri", warehouse_uri.is_some()),
("--nessie-uri", nessie_uri.is_some()),
("--catalog-uri", catalog_uri.is_some()),
]
.into_iter()
.filter_map(|(name, set)| set.then_some(name))
.collect();
if chosen.len() > 1 {
bail!("{} are exclusive — a server fronts ONE catalog; pick one", chosen.join(" and "));
}
// Every leg's TLS files are checked up front — a half client pair, or a
// leg's files with no door that leg belongs to, is refused whatever else is set.
let s3_tls = client_tls(args, env, &S3_LEG)?;
let nessie_tls = client_tls(args, env, &NESSIE_LEG)?;
let catalog_tls = client_tls(args, env, &CATALOG_LEG)?;
if nessie_tls.is_some() && nessie_uri.is_none() {
bail!("{} need --nessie-uri: they configure TLS to Nessie", NESSIE_LEG.flags());
}
if catalog_tls.is_some() && catalog_uri.is_none() {
bail!("{} need --catalog-uri: they configure TLS to the REST catalog", CATALOG_LEG.flags());
}
if let Some(uri) = nessie_uri {
http_uri("--nessie-uri", &uri)?;
leg_mode("--nessie-uri", &uri, nessie_tls.as_ref(), &NESSIE_LEG)?;
#[cfg(feature = "nessie")]
{
return Ok(Arc::new(skade::connect::NessieV2 {
reference: pick("--nessie-ref", "SKADE_FLIGHT_NESSIE_REF").unwrap_or_else(|| "main".into()),
token: env("SKADE_FLIGHT_NESSIE_TOKEN").filter(|t| !t.is_empty()),
tls: nessie_tls,
s3: s3_config(args, env, "--nessie-uri")?,
uri,
}));
}
#[cfg(not(feature = "nessie"))]
bail!("--nessie-uri {uri} needs a build with the `nessie` feature (cargo build --features nessie)");
}
if let Some(uri) = catalog_uri {
http_uri("--catalog-uri", &uri)?;
leg_mode("--catalog-uri", &uri, catalog_tls.as_ref(), &CATALOG_LEG)?;
#[cfg(feature = "rest-catalog")]
{
return Ok(Arc::new(skade::connect::IcebergRest {
warehouse: pick("--catalog-warehouse", "SKADE_FLIGHT_CATALOG_WAREHOUSE"),
prefix_or_ref: pick("--catalog-ref", "SKADE_FLIGHT_CATALOG_REF"),
token: env("SKADE_FLIGHT_CATALOG_TOKEN").filter(|t| !t.is_empty()),
tls: catalog_tls,
s3: s3_config(args, env, "--catalog-uri")?,
read_only: true,
sigv4: None,
uri,
..Default::default()
}));
}
#[cfg(not(feature = "rest-catalog"))]
bail!("--catalog-uri {uri} needs a build with the `rest-catalog` feature (cargo build --features rest-catalog)");
}
let warehouse = warehouse_dir(args, env);
if let Some(uri) = warehouse_uri {
#[cfg(feature = "object-store")]
{
return Ok(Arc::new(skade::connect::SkadeObjectStore {
db_path: warehouse.join("catalog.redb"),
store: s3_store(s3_config(args, env, "--warehouse-uri")?)?,
warehouse_uri: uri,
}));
}
#[cfg(not(feature = "object-store"))]
bail!("--warehouse-uri {uri} needs a build with the `object-store` feature (cargo build --features object-store)");
}
if s3_tls.is_some()
|| ["--s3-bucket", "--s3-endpoint", "--s3-region"].iter().any(|f| flag(args, f).is_some())
|| ["SKADE_FLIGHT_S3_BUCKET", "SKADE_FLIGHT_S3_ENDPOINT", "SKADE_FLIGHT_S3_REGION"].iter().any(|e| env(e).is_some())
{
bail!("--s3-* flags need a catalog that reads S3: --warehouse-uri, --nessie-uri or --catalog-uri");
}
// Say BY NAME why the catalog cannot be opened before redb says it in
// its own words: a bind-mounted warehouse under a container's service
// user was the measured shape (2026-09-09), and the redb error sent the
// tester to `root`. See `skade_flight::dataroot`.
#[cfg(unix)]
if let Some(why) = skade_flight::dataroot::refusal_for(&warehouse) {
bail!("{why}");
}
Ok(Arc::new(skade::connect::SkadeLocal { dir: warehouse }))
}
/// The S3 store behind a catalog door (`door` names the flag in refusals).
#[cfg(any(feature = "object-store", feature = "nessie", feature = "rest-catalog"))]
fn s3_config(args: &[String], env: &dyn Fn(&str) -> Option<String>, door: &str) -> Result<skade::object_store::S3Config> {
let pick = |f: &str, e: &str| flag(args, f).or_else(|| env(e)).filter(|v| !v.is_empty());
let any = |names: &[&str]| names.iter().find_map(|n| env(n).filter(|v| !v.is_empty()));
let Some(bucket) = pick("--s3-bucket", "SKADE_FLIGHT_S3_BUCKET") else {
bail!("{door} needs --s3-bucket NAME (and --s3-endpoint for MinIO): the table files live there");
};
let access_key_id = any(&["SKADE_S3_ACCESS_KEY_ID", "AWS_ACCESS_KEY_ID", "MINIO_ACCESS_KEY"])
.context("no S3 access key in the environment (SKADE_S3_ACCESS_KEY_ID | AWS_ACCESS_KEY_ID | MINIO_ACCESS_KEY)")?;
let secret_access_key = any(&["SKADE_S3_SECRET_ACCESS_KEY", "AWS_SECRET_ACCESS_KEY", "MINIO_SECRET_KEY"])
.context("no S3 secret key in the environment (SKADE_S3_SECRET_ACCESS_KEY | AWS_SECRET_ACCESS_KEY | MINIO_SECRET_KEY)")?;
let endpoint = pick("--s3-endpoint", "SKADE_FLIGHT_S3_ENDPOINT").unwrap_or_default();
let tls = client_tls(args, env, &S3_LEG)?;
leg_mode("--s3-endpoint", &s3_url(&endpoint), tls.as_ref(), &S3_LEG)?;
Ok(skade::object_store::S3Config {
endpoint,
region: pick("--s3-region", "SKADE_FLIGHT_S3_REGION").unwrap_or_else(|| "us-east-1".to_string()),
bucket,
access_key_id,
secret_access_key,
path_style: !args.iter().any(|a| a == "--s3-virtual-hosted"),
tls,
})
}
// ── TLS, per leg ───────────────────────────────────────────────────────────
/// The object store behind `--warehouse-uri`. `RustS3Store` refuses TLS
/// material, so an S3 leg with `--s3-ca` / `--s3-client-cert` goes through the
/// aegir store, which carries it (the same store the Nessie and REST connectors build).
#[cfg(feature = "object-store")]
fn s3_store(s3: skade::object_store::S3Config) -> Result<skade::object_store::ObjectStoreConfig> {
if s3.tls.is_none() {
return Ok(skade::object_store::ObjectStoreConfig::RustS3(s3));
}
#[cfg(any(feature = "nessie", feature = "rest-catalog"))]
return Ok(skade::object_store::ObjectStoreConfig::Aegir(s3));
#[cfg(not(any(feature = "nessie", feature = "rest-catalog")))]
bail!(
"--warehouse-uri with --s3-ca / --s3-client-cert needs the aegir S3 store, which carries TLS: \
build with --features nessie or rest-catalog"
)
}
/// One outbound TLS leg: its name at startup, and its flag / env prefixes
/// (`--nessie` → `--nessie-ca`, `SKADE_FLIGHT_NESSIE` → `SKADE_FLIGHT_NESSIE_CA`).
struct Leg {
name: &'static str,
flag: &'static str,
env: &'static str,
}
impl Leg {
fn flags(&self) -> String {
format!("{0}-ca / {0}-client-cert / {0}-client-key", self.flag)
}
}
const S3_LEG: Leg = Leg { name: "s3", flag: "--s3", env: "SKADE_FLIGHT_S3" };
const NESSIE_LEG: Leg = Leg { name: "nessie", flag: "--nessie", env: "SKADE_FLIGHT_NESSIE" };
const CATALOG_LEG: Leg = Leg { name: "catalog", flag: "--catalog", env: "SKADE_FLIGHT_CATALOG" };
/// A leg's `-ca` / `-client-cert` / `-client-key` files (flag, then env).
/// `None` when none is given. A client cert without its key, or a key without
/// its cert, is refused by name: mutual TLS needs the pair.
fn client_tls(args: &[String], env: &dyn Fn(&str) -> Option<String>, leg: &Leg) -> Result<Option<ClientTls>> {
let pick = |suffix: &str, env_suffix: &str| {
flag(args, &format!("{}-{suffix}", leg.flag))
.or_else(|| env(&format!("{}_{env_suffix}", leg.env)))
.filter(|v| !v.is_empty())
.map(PathBuf::from)
};
let ca = pick("ca", "CA");
let cert = pick("client-cert", "CLIENT_CERT");
let key = pick("client-key", "CLIENT_KEY");
match (&cert, &key) {
(Some(_), None) => bail!(
"{0}-client-cert without {0}-client-key: mutual TLS needs the private key the certificate was issued for",
leg.flag
),
(None, Some(_)) => {
bail!("{0}-client-key without {0}-client-cert: mutual TLS needs the certificate for that key", leg.flag)
}
_ => {}
}
if ca.is_none() && cert.is_none() {
return Ok(None);
}
Ok(Some(ClientTls { ca_pem: ca, client_cert_pem: cert, client_key_pem: key }))
}
/// A leg's mode: `plain` over `http://`, `tls` over `https://`, `mtls` over
/// `https://` with a client cert and key. TLS files on an `http://` URL are
/// refused by name — they would be silently ignored.
fn leg_mode(url_flag: &str, url: &str, tls: Option<&ClientTls>, leg: &Leg) -> Result<&'static str> {
let https = url.starts_with("https://");
if tls.is_some() && !https {
bail!("{} given, but {url_flag} {url} is not https:// — TLS files need an https:// URL", leg.flags());
}
Ok(match tls {
Some(t) if t.client_cert_pem.is_some() => "mtls",
_ if https => "tls",
_ => "plain",
})
}
/// The URL an S3 leg's mode is read from: an empty endpoint is real AWS (https).
fn s3_url(endpoint: &str) -> String {
if endpoint.is_empty() { "https://s3.amazonaws.com".to_string() } else { endpoint.to_string() }
}
/// The Power BI side: `--tls-cert` + `--tls-key` → tls, plus `--tls-client-ca`
/// → mtls, none → plain. Half a pair, or a client CA with no server TLS, is refused.
fn server_mode(args: &[String]) -> Result<&'static str> {
let (cert, key, client_ca) = (flag(args, "--tls-cert"), flag(args, "--tls-key"), flag(args, "--tls-client-ca"));
match (&cert, &key) {
(Some(_), None) => bail!("--tls-cert without --tls-key: the server needs the private key for its certificate"),
(None, Some(_)) => bail!("--tls-key without --tls-cert: the server needs the certificate for that key"),
(None, None) if client_ca.is_some() => {
bail!("--tls-client-ca needs --tls-cert and --tls-key: client certificates ride on server TLS")
}
(None, None) => return Ok("plain"),
_ => {}
}
if cfg!(not(feature = "tls")) {
bail!("--tls-cert needs a build with the `tls` feature (cargo build --features tls)");
}
Ok(if client_ca.is_some() { "mtls" } else { "tls" })
}
/// Every leg this invocation uses, with its resolved mode — what the server
/// prints at startup. Paths are never read here, let alone printed.
fn tls_legs(args: &[String], env: &dyn Fn(&str) -> Option<String>) -> Result<Vec<(&'static str, &'static str)>> {
let pick = |f: &str, e: &str| flag(args, f).or_else(|| env(e)).filter(|v| !v.is_empty());
let mut legs = vec![("powerbi", server_mode(args)?)];
let nessie = pick("--nessie-uri", "SKADE_FLIGHT_NESSIE_URI");
let catalog = pick("--catalog-uri", "SKADE_FLIGHT_CATALOG_URI");
if let Some(uri) = &nessie {
legs.push((NESSIE_LEG.name, leg_mode("--nessie-uri", uri, client_tls(args, env, &NESSIE_LEG)?.as_ref(), &NESSIE_LEG)?));
}
if let Some(uri) = &catalog {
legs.push((
CATALOG_LEG.name,
leg_mode("--catalog-uri", uri, client_tls(args, env, &CATALOG_LEG)?.as_ref(), &CATALOG_LEG)?,
));
}
if nessie.is_some() || catalog.is_some() || pick("--warehouse-uri", "SKADE_FLIGHT_WAREHOUSE_URI").is_some() {
let endpoint = pick("--s3-endpoint", "SKADE_FLIGHT_S3_ENDPOINT").unwrap_or_default();
legs.push((S3_LEG.name, leg_mode("--s3-endpoint", &s3_url(&endpoint), client_tls(args, env, &S3_LEG)?.as_ref(), &S3_LEG)?));
}
Ok(legs)
}
/// A catalog URI must be `http://` or `https://`; anything else is refused by name.
fn http_uri(door: &str, uri: &str) -> Result<()> {
if uri.starts_with("http://") || uri.starts_with("https://") {
return Ok(());
}
bail!("{door} {uri}: want an http:// or https:// URL")
}
/// The loud warning for an OPEN server (no token, no Basic) on an address other
/// machines can reach. `None` when the address is loopback or the server is gated.
fn exposure_warning(addr: &SocketAddr, open: bool) -> Option<String> {
if !open || addr.ip().is_loopback() {
return None;
}
Some(format!(
"skade-flight: ⚠ WARNING ⚠ listening on {addr}, which other machines can reach, AND the server is OPEN \
(no --token / SKADE_FLIGHT_TOKEN / token file, no --basic): anyone who can reach this port reads every table. \
Set a token, or bind loopback with --listen 127.0.0.1:{}.",
addr.port()
))
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use super::*;
fn argv(s: &str) -> Vec<String> {
s.split_whitespace().map(String::from).collect()
}
/// A fixed environment: S3 keys present, nothing else unless given.
fn env_with(extra: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> {
let mut m: HashMap<String, String> = HashMap::from([
("SKADE_S3_ACCESS_KEY_ID".into(), "AKIA-TEST".into()),
("SKADE_S3_SECRET_ACCESS_KEY".into(), "very-secret-key".into()),
]);
for (k, v) in extra {
m.insert((*k).into(), (*v).into());
}
move |k: &str| m.get(k).cloned()
}
fn err(args: &str, env: &dyn Fn(&str) -> Option<String>) -> String {
match connector_from_args(&argv(args), env) {
Ok(c) => panic!("expected a refusal, got {} — {}", c.kind(), c.describe()),
Err(e) => format!("{e:#}"),
}
}
fn legs(args: &str, env: &dyn Fn(&str) -> Option<String>) -> Vec<(&'static str, &'static str)> {
tls_legs(&argv(args), env).unwrap_or_else(|e| panic!("{args}: {e:#}"))
}
fn mode_of(args: &str, leg: &str) -> &'static str {
legs(args, &env_with(&[])).into_iter().find(|(l, _)| *l == leg).unwrap_or_else(|| panic!("no {leg} leg for {args}")).1
}
// ── catalog doors ──────────────────────────────────────────────────────
#[test]
fn no_catalog_flag_is_the_local_skade_warehouse() {
let c = connector_from_args(&argv("--warehouse /nonexistent/wh"), &env_with(&[])).unwrap();
assert_eq!(c.kind(), "skade-local");
assert!(c.describe().contains("/nonexistent/wh"), "{}", c.describe());
assert!(!c.read_only());
}
#[test]
fn s3_flags_without_a_catalog_that_reads_s3_are_refused() {
let m = err("--s3-bucket lake", &env_with(&[]));
assert!(m.contains("--warehouse-uri, --nessie-uri or --catalog-uri"), "{m}");
let m = err("--s3-ca /etc/pki/ca.pem", &env_with(&[]));
assert!(m.contains("--warehouse-uri, --nessie-uri or --catalog-uri"), "{m}");
}
#[test]
fn the_three_catalog_doors_are_exclusive_and_refused_by_name() {
let env = env_with(&[]);
let m = err("--nessie-uri http://n:19120/api/v2 --catalog-uri http://n:19120/iceberg --s3-bucket b", &env);
assert!(m.contains("--nessie-uri and --catalog-uri are exclusive"), "{m}");
let m = err("--warehouse-uri s3://b/wh --nessie-uri http://n:19120/api/v2 --s3-bucket b", &env);
assert!(m.contains("--warehouse-uri and --nessie-uri are exclusive"), "{m}");
// An env-supplied door counts the same as a flag.
let env = env_with(&[("SKADE_FLIGHT_CATALOG_URI", "http://n:19120/iceberg")]);
let m = err("--warehouse-uri s3://b/wh --s3-bucket b", &env);
assert!(m.contains("--warehouse-uri and --catalog-uri are exclusive"), "{m}");
}
#[test]
fn a_token_flag_is_refused_and_names_the_env_var() {
let m = err("--nessie-uri http://n:19120/api/v2 --nessie-token abc --s3-bucket b", &env_with(&[]));
assert!(m.contains("SKADE_FLIGHT_NESSIE_TOKEN"), "{m}");
let m = err("--catalog-uri http://n:19120/iceberg --catalog-token=abc --s3-bucket b", &env_with(&[]));
assert!(m.contains("SKADE_FLIGHT_CATALOG_TOKEN"), "{m}");
}
#[test]
fn a_catalog_uri_that_is_not_http_or_https_is_refused_by_name() {
let env = env_with(&[]);
let m = err("--nessie-uri nessie:19120/api/v2 --s3-bucket b", &env);
assert!(m.contains("--nessie-uri nessie:19120/api/v2: want an http:// or https:// URL"), "{m}");
let m = err("--catalog-uri ftp://n/iceberg --s3-bucket b", &env);
assert!(m.contains("--catalog-uri ftp://n/iceberg: want an http:// or https:// URL"), "{m}");
}
#[cfg(feature = "nessie")]
#[test]
fn nessie_uri_is_a_read_only_nessie_v2_connector_on_main_by_default() {
let env = env_with(&[("SKADE_FLIGHT_NESSIE_TOKEN", "tok-should-not-show")]);
let c = connector_from_args(
&argv("--nessie-uri http://nessie:19120/api/v2 --s3-bucket lake --s3-endpoint http://minio:9000"),
&env,
)
.unwrap();
assert_eq!(c.kind(), "nessie-v2");
assert!(c.read_only());
let d = c.describe();
assert!(d.contains("http://nessie:19120/api/v2") && d.contains("ref main") && d.contains("lake"), "{d}");
assert!(!d.contains("tok-should-not-show") && !d.contains("very-secret-key"), "describe leaks a secret: {d}");
let c = connector_from_args(&argv("--nessie-uri https://nessie:19120/api/v2 --nessie-ref dev --s3-bucket lake"), &env)
.unwrap();
assert!(c.describe().contains("ref dev"), "{}", c.describe());
}
#[cfg(feature = "nessie")]
#[test]
fn nessie_uri_without_a_bucket_is_refused_by_name() {
let m = err("--nessie-uri http://nessie:19120/api/v2", &env_with(&[]));
assert!(m.contains("--nessie-uri needs --s3-bucket"), "{m}");
}
#[cfg(feature = "rest-catalog")]
#[test]
fn catalog_uri_is_a_read_only_iceberg_rest_connector() {
let env = env_with(&[("SKADE_FLIGHT_CATALOG_TOKEN", "tok-should-not-show")]);
let c = connector_from_args(
&argv("--catalog-uri http://nessie:19120/iceberg --catalog-warehouse warehouse --catalog-ref main --s3-bucket lake"),
&env,
)
.unwrap();
assert_eq!(c.kind(), "iceberg-rest");
assert!(c.read_only());
let d = c.describe();
assert!(d.contains("http://nessie:19120/iceberg") && d.contains("warehouse warehouse") && d.contains("prefix main"), "{d}");
assert!(!d.contains("tok-should-not-show") && !d.contains("very-secret-key"), "describe leaks a secret: {d}");
}
#[cfg(feature = "object-store")]
#[test]
fn warehouse_uri_is_skade_over_an_object_store_with_the_local_redb() {
let c = connector_from_args(&argv("--warehouse /var/lib/sk --warehouse-uri s3://lake/wh --s3-bucket lake"), &env_with(&[]))
.unwrap();
assert_eq!(c.kind(), "skade-object-store");
assert!(!c.read_only());
let d = c.describe();
assert!(d.contains("/var/lib/sk/catalog.redb") && d.contains("s3://lake/wh"), "{d}");
}
// ── listen ─────────────────────────────────────────────────────────────
#[test]
fn the_default_listen_is_every_interface_on_port_10009() {
assert_eq!(DEFAULT_LISTEN, "0.0.0.0:10009");
let addr: SocketAddr = DEFAULT_LISTEN.parse().unwrap();
assert!(addr.ip().is_unspecified());
assert_eq!(addr.port(), 10009);
assert!(USAGE.contains("default 0.0.0.0:10009"), "--help must state the default");
}
#[test]
fn an_open_server_on_a_reachable_address_is_warned_about_by_name() {
let exposed: SocketAddr = "0.0.0.0:10009".parse().unwrap();
let w = exposure_warning(&exposed, true).expect("open + 0.0.0.0 must warn");
assert!(w.contains("0.0.0.0:10009") && w.contains("OPEN") && w.contains("other machines"), "{w}");
assert!(exposure_warning(&exposed, false).is_none(), "a gated server does not warn");
let loopback: SocketAddr = "127.0.0.1:10009".parse().unwrap();
assert!(exposure_warning(&loopback, true).is_none(), "loopback is not exposure");
let v6: SocketAddr = "[::]:10009".parse().unwrap();
assert!(exposure_warning(&v6, true).is_some(), "[::] is every interface too");
}
// ── TLS, per leg: plain | tls | mtls ───────────────────────────────────
#[test]
fn the_nessie_leg_resolves_plain_tls_and_mtls() {
assert_eq!(mode_of("--nessie-uri http://n:19120/api/v2", "nessie"), "plain");
assert_eq!(mode_of("--nessie-uri https://n:19120/api/v2", "nessie"), "tls");
assert_eq!(mode_of("--nessie-uri https://n:19120/api/v2 --nessie-ca /pki/ca.pem", "nessie"), "tls");
assert_eq!(
mode_of("--nessie-uri https://n:19120/api/v2 --nessie-client-cert /pki/c.pem --nessie-client-key /pki/k.pem", "nessie"),
"mtls"
);
}
#[test]
fn the_catalog_leg_resolves_plain_tls_and_mtls() {
assert_eq!(mode_of("--catalog-uri http://n:19120/iceberg", "catalog"), "plain");
assert_eq!(mode_of("--catalog-uri https://n:19120/iceberg --catalog-ca /pki/ca.pem", "catalog"), "tls");
assert_eq!(
mode_of(
"--catalog-uri https://n:19120/iceberg --catalog-ca /pki/ca.pem --catalog-client-cert /pki/c.pem --catalog-client-key /pki/k.pem",
"catalog"
),
"mtls"
);
}
#[test]
fn the_s3_leg_resolves_plain_tls_and_mtls() {
let door = "--nessie-uri http://n:19120/api/v2";
assert_eq!(mode_of(&format!("{door} --s3-endpoint http://minio:9000"), "s3"), "plain");
assert_eq!(mode_of(&format!("{door} --s3-endpoint https://minio:9000 --s3-ca /pki/ca.pem"), "s3"), "tls");
assert_eq!(mode_of(&format!("{door}"), "s3"), "tls", "no endpoint is real AWS, over https");
assert_eq!(
mode_of(&format!("{door} --s3-endpoint https://minio:9000 --s3-client-cert /pki/c.pem --s3-client-key /pki/k.pem"), "s3"),
"mtls"
);
// From the environment too.
let env = env_with(&[
("SKADE_FLIGHT_S3_ENDPOINT", "https://minio:9000"),
("SKADE_FLIGHT_S3_CLIENT_CERT", "/pki/c.pem"),
("SKADE_FLIGHT_S3_CLIENT_KEY", "/pki/k.pem"),
]);
assert!(legs(door, &env).contains(&("s3", "mtls")));
}
#[test]
fn the_power_bi_leg_resolves_plain_tls_and_mtls() {
assert_eq!(mode_of("", "powerbi"), "plain");
#[cfg(feature = "tls")]
{
assert_eq!(mode_of("--tls-cert /pki/s.pem --tls-key /pki/s.key", "powerbi"), "tls");
assert_eq!(mode_of("--tls-cert /pki/s.pem --tls-key /pki/s.key --tls-client-ca /pki/ca.pem", "powerbi"), "mtls");
}
#[cfg(not(feature = "tls"))]
{
let e = tls_legs(&argv("--tls-cert /pki/s.pem --tls-key /pki/s.key"), &env_with(&[])).unwrap_err();
assert!(format!("{e}").contains("`tls` feature"), "{e}");
}
let e = tls_legs(&argv("--tls-cert /pki/s.pem"), &env_with(&[])).unwrap_err();
assert!(format!("{e}").contains("--tls-cert without --tls-key"), "{e}");
let e = tls_legs(&argv("--tls-client-ca /pki/ca.pem"), &env_with(&[])).unwrap_err();
assert!(format!("{e}").contains("--tls-client-ca needs --tls-cert and --tls-key"), "{e}");
}
#[test]
fn only_the_legs_in_use_are_listed() {
assert_eq!(legs("--warehouse /nonexistent/wh", &env_with(&[])), vec![("powerbi", "plain")]);
let l = legs("--nessie-uri http://n:19120/api/v2 --s3-endpoint http://minio:9000", &env_with(&[]));
assert_eq!(l, vec![("powerbi", "plain"), ("nessie", "plain"), ("s3", "plain")]);
}
#[test]
fn half_a_client_pair_is_refused_by_name_on_every_leg() {
let env = env_with(&[]);
let m = err("--nessie-uri https://n:19120/api/v2 --s3-bucket b --s3-client-cert /pki/c.pem", &env);
assert!(m.contains("--s3-client-cert without --s3-client-key"), "{m}");
let m = err("--nessie-uri https://n:19120/api/v2 --s3-bucket b --nessie-client-key /pki/k.pem", &env);
assert!(m.contains("--nessie-client-key without --nessie-client-cert"), "{m}");
let m = err("--catalog-uri https://n:19120/iceberg --s3-bucket b --catalog-client-cert /pki/c.pem", &env);
assert!(m.contains("--catalog-client-cert without --catalog-client-key"), "{m}");
// From the environment, and before any catalog door is chosen.
let env = env_with(&[("SKADE_FLIGHT_S3_CLIENT_CERT", "/pki/c.pem")]);
let m = err("", &env);
assert!(m.contains("--s3-client-cert without --s3-client-key"), "{m}");
}
#[test]
fn tls_files_on_an_http_url_are_refused_by_name() {
let env = env_with(&[]);
let m = err("--nessie-uri http://n:19120/api/v2 --nessie-ca /pki/ca.pem --s3-bucket b", &env);
assert!(m.contains("--nessie-uri http://n:19120/api/v2 is not https://"), "{m}");
let m = err("--catalog-uri https://n:19120/iceberg --s3-bucket b --s3-endpoint http://minio:9000 --s3-ca /pki/ca.pem", &env);
assert!(m.contains("--s3-endpoint http://minio:9000 is not https://"), "{m}");
}
#[test]
fn a_legs_tls_files_without_its_door_are_refused_by_name() {
let env = env_with(&[]);
let m = err("--nessie-ca /pki/ca.pem", &env);
assert!(m.contains("--nessie-ca / --nessie-client-cert / --nessie-client-key need --nessie-uri"), "{m}");
let m = err("--nessie-uri https://n:19120/api/v2 --s3-bucket b --catalog-client-cert /pki/c.pem --catalog-client-key /pki/k.pem", &env);
assert!(m.contains("need --catalog-uri"), "{m}");
}
#[cfg(any(feature = "object-store", feature = "nessie", feature = "rest-catalog"))]
#[test]
fn s3_tls_flags_and_env_feed_the_s3_config() {
let env = env_with(&[("SKADE_FLIGHT_S3_CLIENT_KEY", "/pki/k.pem")]);
let s = s3_config(
&argv("--s3-bucket b --s3-endpoint https://minio:9000 --s3-ca /pki/ca.pem --s3-client-cert /pki/c.pem"),
&env,
"--nessie-uri",
)
.unwrap();
let tls = s.tls.expect("tls material given");
assert_eq!(tls.ca_pem, Some(PathBuf::from("/pki/ca.pem")));
assert_eq!(tls.client_cert_pem, Some(PathBuf::from("/pki/c.pem")));
assert_eq!(tls.client_key_pem, Some(PathBuf::from("/pki/k.pem")));
let s = s3_config(&argv("--s3-bucket b --s3-endpoint http://minio:9000"), &env_with(&[]), "--nessie-uri").unwrap();
assert!(s.tls.is_none(), "no TLS flags → platform roots, no client cert");
}
/// RustS3Store refuses TLS material; an S3 leg with TLS must reach the aegir store.
#[cfg(all(feature = "object-store", any(feature = "nessie", feature = "rest-catalog")))]
#[test]
fn a_warehouse_uri_s3_leg_with_tls_uses_the_aegir_store() {
use skade::object_store::ObjectStoreConfig;
let env = env_with(&[]);
let with_tls = s3_config(&argv("--s3-bucket b --s3-endpoint https://minio:9000 --s3-ca /pki/ca.pem"), &env, "--warehouse-uri")
.unwrap();
assert!(matches!(s3_store(with_tls).unwrap(), ObjectStoreConfig::Aegir(_)), "TLS material must go through aegir");
let plain = s3_config(&argv("--s3-bucket b --s3-endpoint http://minio:9000"), &env, "--warehouse-uri").unwrap();
assert!(matches!(s3_store(plain).unwrap(), ObjectStoreConfig::RustS3(_)), "no TLS keeps rust-s3");
}
}