crates/knut-bifrost/src/catalog_args.rs
//! Which Iceberg catalog a `--rest-uri` (or `--grpc-uri`) lake is, and how to
//! reach it — the flags shared by `bifrost watch` and `bifrost doris`.
//!
//! | `--catalog` | skade connector | knut feature |
//! |---|---|---|
//! | `rest` (default) | `IcebergRest` | `skade-rest` |
//! | `aistor` | `AIStor` (MinIO AIStor, `/_iceberg`, SigV4 `s3tables`) | `catalog-aistor` |
//! | `polaris` | `Polaris` (Apache Polaris, `/api/catalog`, OAuth2) | `catalog-polaris` |
//! | `unity` | `UnityCatalog` (Unity Catalog OSS / Databricks) | `catalog-unity` |
//! | `gravitino` | `Gravitino` (Apache Gravitino's Iceberg REST) | `catalog-gravitino` |
//! | `--grpc-uri` | `SkadeGrpc` (skade-katalog's own gRPC wire) | `skade-grpc` |
//!
//! Each vendor is its own feature, so a build carries only the catalogs it
//! names; a `--catalog` this build lacks fails naming the feature to rebuild
//! with. Every vendor preset is an Iceberg REST catalog underneath — it becomes
//! an `IcebergRest`, so a lake's cursor key (`rest:<uri>#<warehouse>#<table>`)
//! does not depend on which preset spelled it.
//!
//! # Secrets never go on the command line
//!
//! The OAuth2 client credential (`client_id:client_secret`) is read from
//! `BIFROST_CATALOG_CREDENTIAL` and a bearer token from `BIFROST_CATALOG_TOKEN`
//! — flags land in `/proc/<pid>/cmdline`. `--rest-header` refuses
//! `Authorization` for the same reason.
use std::collections::BTreeMap;
use std::path::PathBuf;
use clap::{Args, ValueEnum};
/// The OAuth2 client credential, `client_id:client_secret`.
pub const CREDENTIAL_ENV: &str = "BIFROST_CATALOG_CREDENTIAL";
/// A bearer token for the catalog.
pub const TOKEN_ENV: &str = "BIFROST_CATALOG_TOKEN";
/// The catalog secrets, for an installer's secret channel.
pub const SECRET_ENV: [&str; 2] = [CREDENTIAL_ENV, TOKEN_ENV];
/// The REST catalog's vendor.
#[derive(ValueEnum, Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum CatalogKind {
/// Any Iceberg REST catalog, spelled out by the flags.
#[default]
Rest,
/// MinIO AIStor Tables (`<minio>/_iceberg`, SigV4 `s3tables`, warehouse = bucket).
Aistor,
/// Apache Polaris (`/api/catalog`, OAuth2, vended credentials).
Polaris,
/// Unity Catalog OSS / Databricks (`/api/2.1/unity-catalog/iceberg`).
Unity,
/// Apache Gravitino's Iceberg REST service (`/iceberg`).
Gravitino,
}
impl CatalogKind {
pub fn as_str(self) -> &'static str {
match self {
CatalogKind::Rest => "rest",
CatalogKind::Aistor => "aistor",
CatalogKind::Polaris => "polaris",
CatalogKind::Unity => "unity",
CatalogKind::Gravitino => "gravitino",
}
}
}
/// The catalog flags. Flattened into a lake's flags beside `--rest-uri`.
#[derive(Args, Debug, Clone, Default, PartialEq, Eq)]
#[group(skip)]
pub struct CatalogArgs {
/// The `--rest-uri` catalog's vendor; each knows its own path and auth habits.
/// Give `--rest-uri` as the server's base URL (the vendor path is added).
#[arg(long, value_enum, default_value_t = CatalogKind::Rest)]
pub catalog: CatalogKind,
/// OAuth2 scope for the client credential (Polaris default `PRINCIPAL_ROLE:ALL`).
#[arg(long, value_name = "SCOPE")]
pub rest_scope: Option<String>,
/// OAuth2 token endpoint, when it is not the catalog's own `/v1/oauth/tokens`.
#[arg(long, value_name = "URL")]
pub rest_oauth2_uri: Option<String>,
/// Extra HTTP header on every catalog call, `Name=Value` (repeatable). Not for
/// secrets: `Authorization` is refused — use `BIFROST_CATALOG_TOKEN`.
#[arg(long = "rest-header", value_name = "NAME=VALUE")]
pub rest_headers: Vec<String>,
/// Read the data with per-table credentials the catalog vends (`true`/`false`).
/// Default: on for Polaris and Unity, off otherwise.
#[arg(long, value_name = "BOOL")]
pub vended_credentials: Option<bool>,
/// Gravitino `simple` auth: the user name (sent as Basic `user:dummy`).
#[arg(long, value_name = "USER")]
pub gravitino_user: Option<String>,
/// A remote skade-katalog over its gRPC wire, e.g. `https://katalog:50051`
/// (`skade-katalog serve --grpc`). A mapping table is `ns.name` in it. The
/// data is read through `--s3-endpoint` if given, else the local filesystem.
/// Feature `skade-grpc`.
#[arg(long, value_name = "URL")]
pub grpc_uri: Option<String>,
/// CA (PEM file) to verify an `https://` gRPC catalog with.
#[arg(long, value_name = "PEM")]
pub grpc_ca: Option<PathBuf>,
/// Client certificate (PEM) for a mutual-TLS gRPC catalog.
#[arg(long, value_name = "PEM", requires = "grpc_key")]
pub grpc_cert: Option<PathBuf>,
/// Client private key (PEM) for a mutual-TLS gRPC catalog.
#[arg(long, value_name = "PEM", requires = "grpc_cert")]
pub grpc_key: Option<PathBuf>,
}
#[cfg(feature = "skade-rest")]
fn env(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|v| !v.is_empty())
}
#[cfg(feature = "skade-rest")]
fn missing(kind: CatalogKind) -> String {
format!(
"--catalog {}: this build has no {0} connector - rebuild with --features catalog-{0}",
kind.as_str()
)
}
impl CatalogArgs {
/// `--rest-header` as a map; `Authorization` refused (it is a secret).
pub fn headers(&self) -> Result<BTreeMap<String, String>, String> {
let mut out = BTreeMap::new();
for h in &self.rest_headers {
let (k, v) = h.split_once('=').ok_or_else(|| format!("--rest-header {h:?}: want Name=Value"))?;
let k = k.trim();
if k.is_empty() {
return Err(format!("--rest-header {h:?}: empty name"));
}
if k.eq_ignore_ascii_case("authorization") {
return Err("--rest-header Authorization: a secret never goes on the command line - set BIFROST_CATALOG_TOKEN".into());
}
out.insert(k.to_string(), v.trim().to_string());
}
Ok(out)
}
/// The `--rest-uri` catalog as an `IcebergRest`, through the `--catalog`
/// vendor's preset. Secrets come from [`CREDENTIAL_ENV`] / [`TOKEN_ENV`].
#[cfg(feature = "skade-rest")]
pub fn rest(
&self,
uri: &str,
warehouse: Option<String>,
sigv4: Option<String>,
s3: skade::S3Config,
read_only: bool,
) -> Result<skade::connect::IcebergRest, String> {
self.rest_with(uri, warehouse, sigv4, s3, read_only, env(CREDENTIAL_ENV), env(TOKEN_ENV))
}
#[cfg(feature = "skade-rest")]
#[allow(clippy::too_many_arguments)]
fn rest_with(
&self,
uri: &str,
warehouse: Option<String>,
sigv4: Option<String>,
s3: skade::S3Config,
read_only: bool,
credential: Option<String>,
token: Option<String>,
) -> Result<skade::connect::IcebergRest, String> {
use skade::connect::IcebergRest;
let headers = self.headers()?;
if self.gravitino_user.is_some() && self.catalog != CatalogKind::Gravitino {
return Err("--gravitino-user needs --catalog gravitino".into());
}
#[allow(unused_variables)]
let need = |what: &str| {
warehouse
.clone()
.filter(|w| !w.trim().is_empty())
.ok_or_else(|| format!("--catalog {} needs --rest-warehouse ({what})", self.catalog.as_str()))
};
let mut rest = match self.catalog {
CatalogKind::Rest => IcebergRest {
uri: uri.to_string(),
warehouse,
token,
credential,
scope: self.rest_scope.clone(),
oauth2_server_uri: self.rest_oauth2_uri.clone(),
tls: s3.tls.clone(),
vended_credentials: self.vended_credentials.unwrap_or(false),
s3,
read_only,
..IcebergRest::default()
},
#[cfg(feature = "catalog-aistor")]
CatalogKind::Aistor => {
let mut r = skade::connect::AIStor { s3, warehouse: need("the AIStor warehouse")?, read_only }.rest();
// `--rest-uri` names the MinIO; AIStor's catalog is its `/_iceberg`.
r.uri = format!("{}/_iceberg", uri.trim_end_matches('/').trim_end_matches("/_iceberg"));
r.token = token;
r.vended_credentials = self.vended_credentials.unwrap_or(false);
r
}
#[cfg(feature = "catalog-polaris")]
CatalogKind::Polaris => {
let mut p = skade::connect::Polaris::new(uri, need("the Polaris catalog name")?);
p.credential = credential;
p.scope = self.rest_scope.clone();
p.oauth2_server_uri = self.rest_oauth2_uri.clone();
p.token = token;
p.tls = s3.tls.clone();
p.s3 = s3;
p.vended_credentials = self.vended_credentials.unwrap_or(true);
p.read_only = read_only;
p.rest()
}
#[cfg(feature = "catalog-unity")]
CatalogKind::Unity => {
if credential.is_some() {
return Err("--catalog unity authenticates by token: set BIFROST_CATALOG_TOKEN, not BIFROST_CATALOG_CREDENTIAL".into());
}
let mut u = skade::connect::UnityCatalog::new(uri, need("the Unity catalog name")?);
u.token = token;
u.tls = s3.tls.clone();
u.s3 = s3;
u.vended_credentials = self.vended_credentials.unwrap_or(true);
u.read_only = read_only;
u.rest()
}
#[cfg(feature = "catalog-gravitino")]
CatalogKind::Gravitino => {
let mut g = skade::connect::Gravitino::new(uri);
g.warehouse = warehouse;
g.simple_user = self.gravitino_user.clone();
g.credential = credential;
g.scope = self.rest_scope.clone();
g.oauth2_server_uri = self.rest_oauth2_uri.clone();
g.token = token;
g.tls = s3.tls.clone();
g.s3 = s3;
g.vended_credentials = self.vended_credentials.unwrap_or(false);
g.read_only = read_only;
g.rest()
}
#[allow(unreachable_patterns)]
other => return Err(missing(other)),
};
rest.headers.extend(headers);
if let Some(service) = sigv4 {
rest.sigv4 = Some(skade::connect::SigV4 { service });
}
Ok(rest)
}
/// The `--grpc-uri` catalog. `s3` is where its tables' files are; `None`
/// reads them off the local filesystem.
#[cfg(feature = "skade-grpc")]
pub fn grpc(&self, uri: &str, s3: Option<skade::S3Config>, read_only: bool) -> skade::connect::SkadeGrpc {
let tls = (self.grpc_ca.is_some() || self.grpc_cert.is_some()).then(|| skade::tls::ClientTls {
ca_pem: self.grpc_ca.clone(),
client_cert_pem: self.grpc_cert.clone(),
client_key_pem: self.grpc_key.clone(),
});
skade::connect::SkadeGrpc { uri: uri.to_string(), tls, s3, read_only }
}
/// Fail on the flags a `--rest-uri`-less, `--grpc-uri`-less lake would ignore.
pub fn check_unused(&self, has_rest: bool) -> Result<(), String> {
if !has_rest
&& (self.catalog != CatalogKind::Rest
|| self.rest_scope.is_some()
|| self.rest_oauth2_uri.is_some()
|| !self.rest_headers.is_empty()
|| self.vended_credentials.is_some()
|| self.gravitino_user.is_some())
{
return Err("--catalog / --rest-scope / --rest-oauth2-uri / --rest-header / --vended-credentials / --gravitino-user need --rest-uri".into());
}
if self.grpc_uri.is_none() && (self.grpc_ca.is_some() || self.grpc_cert.is_some()) {
return Err("--grpc-ca / --grpc-cert / --grpc-key need --grpc-uri".into());
}
Ok(())
}
/// The flags that recreate these — no secrets among them. Paths are kept
/// as given (an installer absolutises them first).
pub fn to_args(&self) -> Vec<String> {
let mut a = Vec::new();
if self.catalog != CatalogKind::Rest {
a.extend(["--catalog".to_string(), self.catalog.as_str().to_string()]);
}
let mut push = |k: &str, v: Option<String>| {
if let Some(v) = v {
a.push(format!("--{k}"));
a.push(v);
}
};
push("rest-scope", self.rest_scope.clone());
push("rest-oauth2-uri", self.rest_oauth2_uri.clone());
for h in &self.rest_headers {
push("rest-header", Some(h.clone()));
}
push("vended-credentials", self.vended_credentials.map(|b| b.to_string()));
push("gravitino-user", self.gravitino_user.clone());
push("grpc-uri", self.grpc_uri.clone());
push("grpc-ca", self.grpc_ca.as_ref().map(|p| p.display().to_string()));
push("grpc-cert", self.grpc_cert.as_ref().map(|p| p.display().to_string()));
push("grpc-key", self.grpc_key.as_ref().map(|p| p.display().to_string()));
a
}
/// The same flags with every PEM path made absolute against `cwd`.
pub fn absolutized(&self, cwd: &std::path::Path) -> Self {
let abs = |p: &Option<PathBuf>| p.as_ref().map(|p| if p.is_absolute() { p.clone() } else { cwd.join(p) });
Self { grpc_ca: abs(&self.grpc_ca), grpc_cert: abs(&self.grpc_cert), grpc_key: abs(&self.grpc_key), ..self.clone() }
}
}
#[cfg(test)]
mod tests {
use super::*;
use clap::Parser;
#[derive(Parser)]
struct Cli {
#[command(flatten)]
c: CatalogArgs,
}
fn parse(args: &[&str]) -> CatalogArgs {
Cli::try_parse_from(std::iter::once("t").chain(args.iter().copied())).unwrap().c
}
#[cfg(feature = "skade-rest")]
fn s3() -> skade::S3Config {
skade::S3Config { endpoint: "https://minio:9000".into(), bucket: "lake".into(), ..Default::default() }
}
#[test]
fn to_args_round_trips_and_carries_no_secret() {
let c = parse(&[
"--catalog", "gravitino", "--rest-scope", "s", "--rest-oauth2-uri", "https://idp/token",
"--rest-header", "X-Tenant=a", "--vended-credentials", "true", "--gravitino-user", "ana",
"--grpc-uri", "https://k:50051", "--grpc-ca", "/ca.pem",
]);
let args = c.to_args();
let refs: Vec<&str> = args.iter().map(String::as_str).collect();
assert_eq!(parse(&refs), c);
let joined = args.join(" ");
assert!(!joined.contains(CREDENTIAL_ENV) && !joined.contains(TOKEN_ENV));
}
#[test]
fn the_default_is_plain_rest_and_emits_no_flag() {
assert_eq!(parse(&[]).to_args(), Vec::<String>::new());
}
#[test]
fn an_authorization_header_is_refused() {
let c = parse(&["--rest-header", "authorization=Bearer x"]);
assert!(c.headers().unwrap_err().contains("BIFROST_CATALOG_TOKEN"));
assert!(parse(&["--rest-header", "nope"]).headers().is_err());
}
#[cfg(feature = "skade-rest")]
#[test]
fn plain_rest_takes_secrets_from_the_caller_and_headers_from_flags() {
let c = parse(&["--rest-header", "X-Tenant=a", "--rest-scope", "catalog"]);
let r = c
.rest_with("https://cat", Some("wh".into()), Some("s3tables".into()), s3(), true, Some("id:sec".into()), None)
.unwrap();
assert_eq!(r.uri, "https://cat");
assert_eq!(r.credential.as_deref(), Some("id:sec"));
assert_eq!(r.headers.get("X-Tenant").map(String::as_str), Some("a"));
assert_eq!(r.sigv4.map(|s| s.service).as_deref(), Some("s3tables"));
assert!(r.read_only && !r.vended_credentials);
}
#[test]
fn rest_only_flags_without_a_rest_uri_are_refused() {
assert!(parse(&["--catalog", "polaris"]).check_unused(false).is_err());
assert!(parse(&["--catalog", "polaris"]).check_unused(true).is_ok());
assert!(parse(&["--grpc-ca", "/c"]).check_unused(false).is_err());
}
#[cfg(all(feature = "skade-rest", not(feature = "catalog-polaris")))]
#[test]
fn a_vendor_this_build_lacks_names_its_feature() {
let e = parse(&["--catalog", "polaris"]).rest_with("u", Some("c".into()), None, s3(), true, None, None);
assert!(e.unwrap_err().contains("--features catalog-polaris"));
}
#[cfg(feature = "catalog-polaris")]
#[test]
fn polaris_gets_its_path_scope_and_vended_credentials() {
let r = parse(&["--catalog", "polaris"])
.rest_with("https://polaris:8181", Some("cat".into()), None, s3(), true, Some("id:sec".into()), None)
.unwrap();
assert_eq!(r.uri, "https://polaris:8181/api/catalog");
assert_eq!(r.warehouse.as_deref(), Some("cat"));
assert_eq!(r.scope.as_deref(), Some("PRINCIPAL_ROLE:ALL"));
assert!(r.vended_credentials && r.read_only);
assert!(parse(&["--catalog", "polaris"]).rest_with("u", None, None, s3(), true, None, None).is_err());
}
#[cfg(feature = "catalog-aistor")]
#[test]
fn aistor_points_at_the_minio_iceberg_path() {
for uri in ["https://minio:9000", "https://minio:9000/_iceberg/"] {
let r = parse(&["--catalog", "aistor"])
.rest_with(uri, Some("analytics".into()), None, s3(), true, None, None)
.unwrap();
assert_eq!(r.uri, "https://minio:9000/_iceberg");
assert_eq!(r.sigv4.map(|s| s.service).as_deref(), Some("s3tables"));
}
}
#[cfg(feature = "skade-grpc")]
#[test]
fn grpc_tls_comes_from_the_pem_flags() {
let c = parse(&["--grpc-uri", "https://k:50051", "--grpc-ca", "/ca.pem"]);
let g = c.grpc("https://k:50051", None, true);
assert_eq!(g.tls.unwrap().ca_pem.as_deref(), Some(std::path::Path::new("/ca.pem")));
assert!(g.read_only && g.s3.is_none());
}
}