1use crate::error::{CliError, CliResult};
16use axum::http::Method;
17use serde::{Deserialize, Serialize};
18use std::path::Path;
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
22#[serde(rename_all = "snake_case")]
23pub enum Permission {
24 RunRead,
26 RunWrite,
28 SchemaRead,
30 Doctor,
32 TriggerFire,
34 DlqRead,
36 DlqManage,
39 CatalogRead,
41 AuditRead,
43}
44
45#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
49#[serde(rename_all = "snake_case")]
50pub enum Role {
51 Viewer,
53 Operator,
56 Admin,
58}
59
60impl Role {
61 pub fn grants(self, perm: Permission) -> bool {
63 use Permission::*;
64 match self {
65 Role::Viewer => matches!(perm, RunRead | SchemaRead | DlqRead | CatalogRead),
66 Role::Operator => {
67 matches!(
68 perm,
69 RunRead
70 | SchemaRead
71 | DlqRead
72 | CatalogRead
73 | RunWrite
74 | Doctor
75 | TriggerFire
76 | DlqManage
77 )
78 }
79 Role::Admin => true,
80 }
81 }
82
83 pub fn as_str(self) -> &'static str {
84 match self {
85 Role::Viewer => "viewer",
86 Role::Operator => "operator",
87 Role::Admin => "admin",
88 }
89 }
90}
91
92#[derive(Clone, Deserialize)]
95#[serde(deny_unknown_fields)]
96pub struct PrincipalSpec {
97 pub name: String,
98 pub token: String,
99 pub role: Role,
100}
101
102impl std::fmt::Debug for PrincipalSpec {
105 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106 f.debug_struct("PrincipalSpec")
107 .field("name", &self.name)
108 .field("token", &"***")
109 .field("role", &self.role)
110 .finish()
111 }
112}
113
114#[derive(Debug, Clone, Deserialize)]
117#[serde(deny_unknown_fields)]
118struct AuthConfigFile {
119 principals: Vec<PrincipalSpec>,
120}
121
122#[derive(Debug, Clone)]
125pub struct RbacConfig {
126 principals: Vec<PrincipalSpec>,
127}
128
129#[derive(Debug, Clone)]
132pub struct AuthContext {
133 pub principal: String,
134 pub role: Role,
135 pub source_ip: Option<String>,
136}
137
138impl AuthContext {
139 pub fn trigger(name: &str) -> Self {
142 Self {
143 principal: format!("trigger:{name}"),
144 role: Role::Operator,
145 source_ip: None,
146 }
147 }
148}
149
150impl RbacConfig {
151 pub fn from_file(path: &Path) -> CliResult<Self> {
153 let text = std::fs::read_to_string(path).map_err(|e| {
154 CliError::Serve(format!("reading --auth-config {}: {e}", path.display()))
155 })?;
156 let file: AuthConfigFile = serde_yaml::from_str(&text).map_err(|e| {
157 CliError::Serve(format!("parsing --auth-config {}: {e}", path.display()))
158 })?;
159 Self::new(file.principals)
160 }
161
162 pub fn new(principals: Vec<PrincipalSpec>) -> CliResult<Self> {
164 if principals.is_empty() {
165 return Err(CliError::Serve(
166 "--auth-config must define at least one principal".into(),
167 ));
168 }
169 let mut seen_names = std::collections::HashSet::new();
170 let mut seen_tokens = std::collections::HashSet::new();
171 for p in &principals {
172 if p.name.trim().is_empty() {
173 return Err(CliError::Serve(
174 "--auth-config: every principal must have a non-empty name".into(),
175 ));
176 }
177 if p.token.is_empty() {
178 return Err(CliError::Serve(format!(
179 "--auth-config: principal '{}' has an empty token",
180 p.name
181 )));
182 }
183 if !seen_names.insert(p.name.clone()) {
184 return Err(CliError::Serve(format!(
185 "--auth-config: duplicate principal name '{}'",
186 p.name
187 )));
188 }
189 if !seen_tokens.insert(p.token.clone()) {
190 return Err(CliError::Serve(format!(
191 "--auth-config: principal '{}' reuses a token already assigned to another \
192 principal",
193 p.name
194 )));
195 }
196 }
197 Ok(Self { principals })
198 }
199
200 pub fn authenticate(&self, token: &str) -> Option<AuthContext> {
204 let mut matched: Option<(&str, Role)> = None;
205 for p in &self.principals {
206 if crate::serve::auth::constant_time_eq(token.as_bytes(), p.token.as_bytes()) {
207 matched = Some((p.name.as_str(), p.role));
208 }
209 }
210 matched.map(|(name, role)| AuthContext {
211 principal: name.to_string(),
212 role,
213 source_ip: None,
214 })
215 }
216
217 pub fn tokens(&self) -> impl Iterator<Item = &str> {
219 self.principals.iter().map(|p| p.token.as_str())
220 }
221}
222
223pub fn required_permission(method: &Method, matched_path: &str) -> Option<Permission> {
227 use Permission::*;
228 match (method, matched_path) {
229 (&Method::POST, "/v1/runs") => Some(RunWrite),
230 (&Method::GET, "/v1/runs") => Some(RunRead),
231 (&Method::GET, "/v1/runs/{id}") => Some(RunRead),
232 (&Method::DELETE, "/v1/runs/{id}") => Some(RunWrite),
233 (&Method::POST, "/v1/runs/{id}/cancel") => Some(RunWrite),
234 (&Method::GET, "/v1/runs/{id}/logs") => Some(RunRead),
235 (&Method::GET, "/v1/schemas") => Some(SchemaRead),
236 (&Method::GET, "/v1/schemas/{kind}/{name}") => Some(SchemaRead),
237 (&Method::POST, "/v1/doctor") => Some(Doctor),
238 (&Method::POST, "/v1/dlq/inspect") => Some(DlqRead),
239 (&Method::POST, "/v1/dlq/replay") => Some(DlqManage),
240 (&Method::POST, "/v1/dlq/discard") => Some(DlqManage),
241 (&Method::GET, "/v1/audit") => Some(AuditRead),
242 (&Method::POST, "/v1/triggers/{name}") => Some(TriggerFire),
243 (&Method::PUT, "/v1/triggers/{name}") => Some(TriggerFire),
244 (&Method::GET, "/v1/catalog/datasets") => Some(CatalogRead),
245 (&Method::GET, "/v1/catalog/datasets/{id}") => Some(CatalogRead),
246 (&Method::GET, "/v1/catalog/lineage") => Some(CatalogRead),
247 _ => None,
248 }
249}
250
251pub fn audit_action(method: &Method, matched_path: &str) -> &'static str {
253 match (method, matched_path) {
254 (&Method::POST, "/v1/runs") => "run.submit",
255 (&Method::GET, "/v1/runs") => "run.list",
256 (&Method::GET, "/v1/runs/{id}") => "run.get",
257 (&Method::DELETE, "/v1/runs/{id}") => "run.delete",
258 (&Method::POST, "/v1/runs/{id}/cancel") => "run.cancel",
259 (&Method::GET, "/v1/runs/{id}/logs") => "run.logs",
260 (&Method::GET, "/v1/schemas") => "schema.list",
261 (&Method::GET, "/v1/schemas/{kind}/{name}") => "schema.get",
262 (&Method::POST, "/v1/doctor") => "doctor",
263 (&Method::POST, "/v1/dlq/inspect") => "dlq.inspect",
264 (&Method::POST, "/v1/dlq/replay") => "dlq.replay",
265 (&Method::POST, "/v1/dlq/discard") => "dlq.discard",
266 (&Method::GET, "/v1/audit") => "audit.list",
267 (&Method::POST | &Method::PUT, "/v1/triggers/{name}") => "trigger.fire",
268 (&Method::GET, "/v1/catalog/datasets") => "catalog.list",
269 (&Method::GET, "/v1/catalog/datasets/{id}") => "catalog.get",
270 (&Method::GET, "/v1/catalog/lineage") => "catalog.lineage",
271 _ => "unknown",
272 }
273}
274
275#[cfg(test)]
276mod tests {
277 use super::*;
278
279 fn spec(name: &str, token: &str, role: Role) -> PrincipalSpec {
280 PrincipalSpec {
281 name: name.into(),
282 token: token.into(),
283 role,
284 }
285 }
286
287 #[test]
288 fn role_permission_ladder() {
289 use Permission::*;
290 assert!(Role::Viewer.grants(RunRead));
292 assert!(Role::Viewer.grants(SchemaRead));
293 assert!(Role::Viewer.grants(DlqRead));
294 assert!(!Role::Viewer.grants(RunWrite));
295 assert!(!Role::Viewer.grants(Doctor));
296 assert!(!Role::Viewer.grants(DlqManage));
297 assert!(!Role::Viewer.grants(AuditRead));
298 assert!(Role::Operator.grants(RunWrite));
300 assert!(Role::Operator.grants(Doctor));
301 assert!(Role::Operator.grants(TriggerFire));
302 assert!(Role::Operator.grants(DlqRead));
303 assert!(Role::Operator.grants(DlqManage));
304 assert!(!Role::Operator.grants(AuditRead));
305 for p in [
307 RunRead,
308 RunWrite,
309 SchemaRead,
310 Doctor,
311 TriggerFire,
312 DlqRead,
313 DlqManage,
314 AuditRead,
315 ] {
316 assert!(Role::Admin.grants(p));
317 }
318 }
319
320 #[test]
321 fn authenticate_resolves_token_to_principal() {
322 let cfg = RbacConfig::new(vec![
323 spec("alice", "tok-a", Role::Admin),
324 spec("bob", "tok-b", Role::Viewer),
325 ])
326 .unwrap();
327 let a = cfg.authenticate("tok-a").unwrap();
328 assert_eq!(a.principal, "alice");
329 assert_eq!(a.role, Role::Admin);
330 let b = cfg.authenticate("tok-b").unwrap();
331 assert_eq!(b.role, Role::Viewer);
332 assert!(cfg.authenticate("nope").is_none());
333 }
334
335 #[test]
336 fn rejects_empty_duplicate_and_blank() {
337 assert!(RbacConfig::new(vec![]).is_err());
338 assert!(RbacConfig::new(vec![spec("", "t", Role::Admin)]).is_err());
339 assert!(RbacConfig::new(vec![spec("a", "", Role::Admin)]).is_err());
340 assert!(
342 RbacConfig::new(vec![
343 spec("a", "t1", Role::Admin),
344 spec("a", "t2", Role::Viewer),
345 ])
346 .is_err()
347 );
348 assert!(
350 RbacConfig::new(vec![
351 spec("a", "dup", Role::Admin),
352 spec("b", "dup", Role::Viewer),
353 ])
354 .is_err()
355 );
356 }
357
358 #[test]
359 fn debug_masks_token() {
360 let s = format!("{:?}", spec("alice", "supersecret", Role::Admin));
361 assert!(!s.contains("supersecret"), "token leaked: {s}");
362 assert!(s.contains("***"));
363 }
364
365 #[test]
366 fn trigger_actor_is_operator() {
367 let ctx = AuthContext::trigger("nightly");
368 assert_eq!(ctx.principal, "trigger:nightly");
369 assert_eq!(ctx.role, Role::Operator);
370 assert!(ctx.source_ip.is_none());
371 }
372
373 #[test]
374 fn tokens_iterates_all_principals() {
375 let cfg = RbacConfig::new(vec![
376 spec("a", "t1", Role::Admin),
377 spec("b", "t2", Role::Viewer),
378 ])
379 .unwrap();
380 let toks: Vec<&str> = cfg.tokens().collect();
381 assert_eq!(toks, vec!["t1", "t2"]);
382 }
383
384 #[test]
385 fn required_permission_covers_all_routes() {
386 use Permission::*;
387 for (m, path, want) in [
388 (Method::GET, "/v1/runs/{id}", RunRead),
389 (Method::DELETE, "/v1/runs/{id}", RunWrite),
390 (Method::POST, "/v1/runs/{id}/cancel", RunWrite),
391 (Method::GET, "/v1/runs/{id}/logs", RunRead),
392 (Method::GET, "/v1/schemas", SchemaRead),
393 (Method::GET, "/v1/schemas/{kind}/{name}", SchemaRead),
394 (Method::POST, "/v1/doctor", Doctor),
395 (Method::POST, "/v1/triggers/{name}", TriggerFire),
396 (Method::PUT, "/v1/triggers/{name}", TriggerFire),
397 (Method::POST, "/v1/dlq/inspect", DlqRead),
398 (Method::POST, "/v1/dlq/replay", DlqManage),
399 (Method::POST, "/v1/dlq/discard", DlqManage),
400 (Method::GET, "/v1/catalog/datasets", CatalogRead),
401 (Method::GET, "/v1/catalog/datasets/{id}", CatalogRead),
402 (Method::GET, "/v1/catalog/lineage", CatalogRead),
403 ] {
404 assert_eq!(required_permission(&m, path), Some(want), "{m} {path}");
405 }
406 assert!(Role::Viewer.grants(Permission::CatalogRead));
408 assert!(Role::Operator.grants(Permission::CatalogRead));
409 assert!(Role::Admin.grants(Permission::CatalogRead));
410 }
411
412 #[test]
413 fn role_and_permission_serde_snake_case() {
414 assert_eq!(
415 serde_json::to_string(&Role::Operator).unwrap(),
416 "\"operator\""
417 );
418 assert_eq!(
419 serde_json::to_string(&Permission::AuditRead).unwrap(),
420 "\"audit_read\""
421 );
422 }
423
424 #[test]
425 fn required_permission_maps_routes() {
426 assert_eq!(
427 required_permission(&Method::POST, "/v1/runs"),
428 Some(Permission::RunWrite)
429 );
430 assert_eq!(
431 required_permission(&Method::GET, "/v1/runs"),
432 Some(Permission::RunRead)
433 );
434 assert_eq!(
435 required_permission(&Method::GET, "/v1/audit"),
436 Some(Permission::AuditRead)
437 );
438 assert_eq!(required_permission(&Method::GET, "/v1/unknown"), None);
440 }
441
442 #[test]
443 fn parses_yaml_and_json() {
444 let yaml = "principals:\n - name: alice\n token: tok-a\n role: admin\n";
445 let cfg: AuthConfigFile = serde_yaml::from_str(yaml).unwrap();
446 assert_eq!(cfg.principals.len(), 1);
447 let json = r#"{"principals":[{"name":"bob","token":"tok-b","role":"viewer"}]}"#;
448 let cfg: AuthConfigFile = serde_yaml::from_str(json).unwrap();
449 assert_eq!(cfg.principals[0].role, Role::Viewer);
450 }
451
452 #[test]
453 fn audit_action_labels() {
454 assert_eq!(audit_action(&Method::POST, "/v1/runs"), "run.submit");
455 assert_eq!(
456 audit_action(&Method::POST, "/v1/runs/{id}/cancel"),
457 "run.cancel"
458 );
459 assert_eq!(audit_action(&Method::GET, "/v1/whatever"), "unknown");
460 }
461}