use crate::error::{CliError, CliResult};
use axum::http::Method;
use serde::{Deserialize, Serialize};
use std::path::Path;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Permission {
RunRead,
RunWrite,
SchemaRead,
Doctor,
TriggerFire,
DlqRead,
DlqManage,
CatalogRead,
TemplateRead,
TemplateWrite,
AuditRead,
Reload,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Role {
Viewer,
Operator,
Admin,
}
impl Role {
pub fn grants(self, perm: Permission) -> bool {
use Permission::*;
match self {
Role::Viewer => {
matches!(
perm,
RunRead | SchemaRead | DlqRead | CatalogRead | TemplateRead
)
}
Role::Operator => {
matches!(
perm,
RunRead
| SchemaRead
| DlqRead
| CatalogRead
| TemplateRead
| RunWrite
| Doctor
| TriggerFire
| DlqManage
| TemplateWrite
)
}
Role::Admin => true,
}
}
pub fn as_str(self) -> &'static str {
match self {
Role::Viewer => "viewer",
Role::Operator => "operator",
Role::Admin => "admin",
}
}
}
#[derive(Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PrincipalSpec {
pub name: String,
pub token: String,
pub role: Role,
}
impl std::fmt::Debug for PrincipalSpec {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PrincipalSpec")
.field("name", &self.name)
.field("token", &"***")
.field("role", &self.role)
.finish()
}
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct AuthConfigFile {
principals: Vec<PrincipalSpec>,
}
#[derive(Debug, Clone)]
pub struct RbacConfig {
principals: Vec<PrincipalSpec>,
}
#[derive(Debug, Clone)]
pub struct AuthContext {
pub principal: String,
pub role: Role,
pub source_ip: Option<String>,
}
impl AuthContext {
pub fn trigger(name: &str) -> Self {
Self {
principal: format!("trigger:{name}"),
role: Role::Operator,
source_ip: None,
}
}
}
impl RbacConfig {
pub fn from_file(path: &Path) -> CliResult<Self> {
let text = std::fs::read_to_string(path).map_err(|e| {
CliError::Serve(format!("reading --auth-config {}: {e}", path.display()))
})?;
let file: AuthConfigFile = serde_yaml::from_str(&text).map_err(|e| {
CliError::Serve(format!("parsing --auth-config {}: {e}", path.display()))
})?;
Self::new(file.principals)
}
pub fn new(principals: Vec<PrincipalSpec>) -> CliResult<Self> {
if principals.is_empty() {
return Err(CliError::Serve(
"--auth-config must define at least one principal".into(),
));
}
let mut seen_names = std::collections::HashSet::new();
let mut seen_tokens = std::collections::HashSet::new();
for p in &principals {
if p.name.trim().is_empty() {
return Err(CliError::Serve(
"--auth-config: every principal must have a non-empty name".into(),
));
}
if p.token.is_empty() {
return Err(CliError::Serve(format!(
"--auth-config: principal '{}' has an empty token",
p.name
)));
}
if !seen_names.insert(p.name.clone()) {
return Err(CliError::Serve(format!(
"--auth-config: duplicate principal name '{}'",
p.name
)));
}
if !seen_tokens.insert(p.token.clone()) {
return Err(CliError::Serve(format!(
"--auth-config: principal '{}' reuses a token already assigned to another \
principal",
p.name
)));
}
}
Ok(Self { principals })
}
pub fn authenticate(&self, token: &str) -> Option<AuthContext> {
let mut matched: Option<(&str, Role)> = None;
for p in &self.principals {
if crate::serve::auth::constant_time_eq(token.as_bytes(), p.token.as_bytes()) {
matched = Some((p.name.as_str(), p.role));
}
}
matched.map(|(name, role)| AuthContext {
principal: name.to_string(),
role,
source_ip: None,
})
}
pub fn tokens(&self) -> impl Iterator<Item = &str> {
self.principals.iter().map(|p| p.token.as_str())
}
}
pub fn required_permission(method: &Method, matched_path: &str) -> Option<Permission> {
use Permission::*;
match (method, matched_path) {
(&Method::POST, "/v1/runs") => Some(RunWrite),
(&Method::GET, "/v1/runs") => Some(RunRead),
(&Method::GET, "/v1/runs/{id}") => Some(RunRead),
(&Method::DELETE, "/v1/runs/{id}") => Some(RunWrite),
(&Method::POST, "/v1/runs/{id}/cancel") => Some(RunWrite),
(&Method::GET, "/v1/runs/{id}/logs") => Some(RunRead),
(&Method::GET, "/v1/schemas") => Some(SchemaRead),
(&Method::GET, "/v1/schemas/{kind}/{name}") => Some(SchemaRead),
(&Method::POST, "/v1/doctor") => Some(Doctor),
(&Method::POST, "/v1/backfill") => Some(RunWrite),
(&Method::POST, "/v1/dlq/inspect") => Some(DlqRead),
(&Method::POST, "/v1/dlq/replay") => Some(DlqManage),
(&Method::POST, "/v1/dlq/discard") => Some(DlqManage),
(&Method::GET, "/v1/audit") => Some(AuditRead),
(&Method::POST, "/v1/triggers/{name}") => Some(TriggerFire),
(&Method::PUT, "/v1/triggers/{name}") => Some(TriggerFire),
(&Method::GET, "/v1/catalog/datasets") => Some(CatalogRead),
(&Method::GET, "/v1/catalog/datasets/{id}") => Some(CatalogRead),
(&Method::GET, "/v1/catalog/lineage") => Some(CatalogRead),
(&Method::POST, "/v1/templates") => Some(TemplateWrite),
(&Method::GET, "/v1/templates") => Some(TemplateRead),
(&Method::GET, "/v1/templates/{id}") => Some(TemplateRead),
(&Method::DELETE, "/v1/templates/{id}") => Some(TemplateWrite),
(&Method::POST, "/v1/templates/{id}/runs") => Some(RunWrite),
(&Method::POST, "/v1/templates/{id}/tags") => Some(TemplateWrite),
(&Method::POST, "/v1/templates/{id}/launch") => Some(TemplateWrite),
(&Method::POST, "/v1/templates/{id}/rollback") => Some(TemplateWrite),
(&Method::POST, "/v1/templates/{id}/deprecate") => Some(TemplateWrite),
(&Method::POST, "/v1/reload") => Some(Reload),
(&Method::POST, "/mcp") => Some(SchemaRead),
_ => None,
}
}
pub fn audit_action(method: &Method, matched_path: &str) -> &'static str {
match (method, matched_path) {
(&Method::POST, "/v1/runs") => "run.submit",
(&Method::GET, "/v1/runs") => "run.list",
(&Method::GET, "/v1/runs/{id}") => "run.get",
(&Method::DELETE, "/v1/runs/{id}") => "run.delete",
(&Method::POST, "/v1/runs/{id}/cancel") => "run.cancel",
(&Method::GET, "/v1/runs/{id}/logs") => "run.logs",
(&Method::GET, "/v1/schemas") => "schema.list",
(&Method::GET, "/v1/schemas/{kind}/{name}") => "schema.get",
(&Method::POST, "/v1/doctor") => "doctor",
(&Method::POST, "/v1/backfill") => "backfill.submit",
(&Method::POST, "/v1/dlq/inspect") => "dlq.inspect",
(&Method::POST, "/v1/dlq/replay") => "dlq.replay",
(&Method::POST, "/v1/dlq/discard") => "dlq.discard",
(&Method::GET, "/v1/audit") => "audit.list",
(&Method::POST | &Method::PUT, "/v1/triggers/{name}") => "trigger.fire",
(&Method::GET, "/v1/catalog/datasets") => "catalog.list",
(&Method::GET, "/v1/catalog/datasets/{id}") => "catalog.get",
(&Method::GET, "/v1/catalog/lineage") => "catalog.lineage",
(&Method::POST, "/v1/templates") => "template.register",
(&Method::GET, "/v1/templates") => "template.list",
(&Method::GET, "/v1/templates/{id}") => "template.get",
(&Method::DELETE, "/v1/templates/{id}") => "template.delete",
(&Method::POST, "/v1/templates/{id}/runs") => "template.run",
(&Method::POST, "/v1/templates/{id}/tags") => "template.promote",
(&Method::POST, "/v1/templates/{id}/launch") => "template.launch",
(&Method::POST, "/v1/templates/{id}/rollback") => "template.rollback",
(&Method::POST, "/v1/templates/{id}/deprecate") => "template.deprecate",
(&Method::POST, "/v1/reload") => "config.reload",
(&Method::POST, "/mcp") => "mcp",
_ => "unknown",
}
}
#[cfg(test)]
mod tests {
use super::*;
fn spec(name: &str, token: &str, role: Role) -> PrincipalSpec {
PrincipalSpec {
name: name.into(),
token: token.into(),
role,
}
}
#[test]
fn role_permission_ladder() {
use Permission::*;
assert!(Role::Viewer.grants(RunRead));
assert!(Role::Viewer.grants(SchemaRead));
assert!(Role::Viewer.grants(DlqRead));
assert!(Role::Viewer.grants(TemplateRead));
assert!(!Role::Viewer.grants(RunWrite));
assert!(!Role::Viewer.grants(Doctor));
assert!(!Role::Viewer.grants(DlqManage));
assert!(!Role::Viewer.grants(AuditRead));
assert!(!Role::Viewer.grants(TemplateWrite));
assert!(Role::Operator.grants(RunWrite));
assert!(Role::Operator.grants(Doctor));
assert!(Role::Operator.grants(TriggerFire));
assert!(Role::Operator.grants(DlqRead));
assert!(Role::Operator.grants(DlqManage));
assert!(Role::Operator.grants(TemplateWrite));
assert!(!Role::Operator.grants(AuditRead));
for p in [
RunRead,
RunWrite,
SchemaRead,
Doctor,
TriggerFire,
DlqRead,
DlqManage,
AuditRead,
TemplateRead,
TemplateWrite,
] {
assert!(Role::Admin.grants(p));
}
}
#[test]
fn authenticate_resolves_token_to_principal() {
let cfg = RbacConfig::new(vec![
spec("alice", "tok-a", Role::Admin),
spec("bob", "tok-b", Role::Viewer),
])
.unwrap();
let a = cfg.authenticate("tok-a").unwrap();
assert_eq!(a.principal, "alice");
assert_eq!(a.role, Role::Admin);
let b = cfg.authenticate("tok-b").unwrap();
assert_eq!(b.role, Role::Viewer);
assert!(cfg.authenticate("nope").is_none());
}
#[test]
fn rejects_empty_duplicate_and_blank() {
assert!(RbacConfig::new(vec![]).is_err());
assert!(RbacConfig::new(vec![spec("", "t", Role::Admin)]).is_err());
assert!(RbacConfig::new(vec![spec("a", "", Role::Admin)]).is_err());
assert!(
RbacConfig::new(vec![
spec("a", "t1", Role::Admin),
spec("a", "t2", Role::Viewer),
])
.is_err()
);
assert!(
RbacConfig::new(vec![
spec("a", "dup", Role::Admin),
spec("b", "dup", Role::Viewer),
])
.is_err()
);
}
#[test]
fn debug_masks_token() {
let s = format!("{:?}", spec("alice", "supersecret", Role::Admin));
assert!(!s.contains("supersecret"), "token leaked: {s}");
assert!(s.contains("***"));
}
#[test]
fn trigger_actor_is_operator() {
let ctx = AuthContext::trigger("nightly");
assert_eq!(ctx.principal, "trigger:nightly");
assert_eq!(ctx.role, Role::Operator);
assert!(ctx.source_ip.is_none());
}
#[test]
fn tokens_iterates_all_principals() {
let cfg = RbacConfig::new(vec![
spec("a", "t1", Role::Admin),
spec("b", "t2", Role::Viewer),
])
.unwrap();
let toks: Vec<&str> = cfg.tokens().collect();
assert_eq!(toks, vec!["t1", "t2"]);
}
#[test]
fn required_permission_covers_all_routes() {
use Permission::*;
for (m, path, want) in [
(Method::GET, "/v1/runs/{id}", RunRead),
(Method::DELETE, "/v1/runs/{id}", RunWrite),
(Method::POST, "/v1/runs/{id}/cancel", RunWrite),
(Method::GET, "/v1/runs/{id}/logs", RunRead),
(Method::GET, "/v1/schemas", SchemaRead),
(Method::GET, "/v1/schemas/{kind}/{name}", SchemaRead),
(Method::POST, "/v1/doctor", Doctor),
(Method::POST, "/v1/triggers/{name}", TriggerFire),
(Method::PUT, "/v1/triggers/{name}", TriggerFire),
(Method::POST, "/v1/backfill", RunWrite),
(Method::POST, "/v1/dlq/inspect", DlqRead),
(Method::POST, "/v1/dlq/replay", DlqManage),
(Method::POST, "/v1/dlq/discard", DlqManage),
(Method::GET, "/v1/catalog/datasets", CatalogRead),
(Method::GET, "/v1/catalog/datasets/{id}", CatalogRead),
(Method::GET, "/v1/catalog/lineage", CatalogRead),
(Method::POST, "/v1/templates", TemplateWrite),
(Method::GET, "/v1/templates", TemplateRead),
(Method::GET, "/v1/templates/{id}", TemplateRead),
(Method::DELETE, "/v1/templates/{id}", TemplateWrite),
(Method::POST, "/v1/templates/{id}/runs", RunWrite),
(Method::POST, "/v1/templates/{id}/tags", TemplateWrite),
(Method::POST, "/v1/templates/{id}/launch", TemplateWrite),
(Method::POST, "/v1/templates/{id}/rollback", TemplateWrite),
(Method::POST, "/v1/templates/{id}/deprecate", TemplateWrite),
(Method::POST, "/v1/reload", Reload),
] {
assert_eq!(required_permission(&m, path), Some(want), "{m} {path}");
}
assert!(!Role::Viewer.grants(Permission::Reload));
assert!(!Role::Operator.grants(Permission::Reload));
assert!(Role::Admin.grants(Permission::Reload));
assert!(Role::Viewer.grants(Permission::CatalogRead));
assert!(Role::Operator.grants(Permission::CatalogRead));
assert!(Role::Admin.grants(Permission::CatalogRead));
}
#[test]
fn role_and_permission_serde_snake_case() {
assert_eq!(
serde_json::to_string(&Role::Operator).unwrap(),
"\"operator\""
);
assert_eq!(
serde_json::to_string(&Permission::AuditRead).unwrap(),
"\"audit_read\""
);
}
#[test]
fn required_permission_maps_routes() {
assert_eq!(
required_permission(&Method::POST, "/v1/runs"),
Some(Permission::RunWrite)
);
assert_eq!(
required_permission(&Method::GET, "/v1/runs"),
Some(Permission::RunRead)
);
assert_eq!(
required_permission(&Method::GET, "/v1/audit"),
Some(Permission::AuditRead)
);
assert_eq!(required_permission(&Method::GET, "/v1/unknown"), None);
}
#[test]
fn parses_yaml_and_json() {
let yaml = "principals:\n - name: alice\n token: tok-a\n role: admin\n";
let cfg: AuthConfigFile = serde_yaml::from_str(yaml).unwrap();
assert_eq!(cfg.principals.len(), 1);
let json = r#"{"principals":[{"name":"bob","token":"tok-b","role":"viewer"}]}"#;
let cfg: AuthConfigFile = serde_yaml::from_str(json).unwrap();
assert_eq!(cfg.principals[0].role, Role::Viewer);
}
#[test]
fn audit_action_labels() {
assert_eq!(audit_action(&Method::POST, "/v1/runs"), "run.submit");
assert_eq!(
audit_action(&Method::POST, "/v1/runs/{id}/cancel"),
"run.cancel"
);
assert_eq!(
audit_action(&Method::POST, "/v1/templates"),
"template.register"
);
assert_eq!(
audit_action(&Method::POST, "/v1/templates/{id}/runs"),
"template.run"
);
assert_eq!(audit_action(&Method::GET, "/v1/whatever"), "unknown");
}
}