diff --git a/CHANGELOG.md b/CHANGELOG.md index a528c67..08715cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,19 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Added + +- Declarative role-based authorization: `require_roles(name, roles)` on the + engine and builder declares the roles a caller must hold to invoke a + workflow. The check runs before the body on every execution path (direct, + queued, scheduled, child, recovery) against the run's `AuthContext`; the + first matching required role becomes the run's assumed role, and a denial + is a typed `Error::NotAuthorized` (`ErrorCode::NotAuthorized`) that + finalizes the run `ERROR` — terminal by construction, so an unauthorized + queued run cannot loop through the dispatcher. Portable-mode rows record + the denial under the cross-SDK `DBOSNotAuthorizedError` envelope name. A + declaration naming an unregistered workflow is rejected at launch/build. + Documented in the `security` guide's new Authorization section. ### Documentation - HTTP triggering recipe: a runnable axum example diff --git a/src/engine.rs b/src/engine.rs index 8588b81..ac8b33f 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -683,6 +683,9 @@ pub struct DurableEngine { /// Automatic history retention enforced by a [`launch`](Self::launch)-spawned /// sweeper, if configured. See [`EngineConfig::retention`]. retention: Option, + /// Required-roles declarations keyed by workflow name; see + /// [`require_roles`](Self::require_roles). + required_roles: HashMap>, /// Cancelled by [`shutdown`](Self::shutdown) to stop the background loops. /// [`launch`](Self::launch) installs a fresh token after a shutdown, since a /// cancelled token can't be reset. @@ -720,6 +723,7 @@ pub struct DurableEngineBuilder { queues: Vec, listen_filter: Option>, max_recovery_attempts: i32, + required_roles: HashMap>, } impl DurableEngineBuilder { @@ -760,6 +764,19 @@ impl DurableEngineBuilder { self } + /// Declare the roles required to invoke workflow `name` — see + /// [`DurableEngine::require_roles`]. Validated against the registrations + /// at [`build`](Self::build). + pub fn require_roles(&mut self, name: impl Into, roles: I) -> &mut Self + where + I: IntoIterator, + S: Into, + { + self.required_roles + .insert(name.into(), roles.into_iter().map(Into::into).collect()); + self + } + /// Restrict which registered queues this process dispatches. See /// [`DurableEngine::listen_queues`]. pub fn listen_queues(&mut self, names: I) -> &mut Self @@ -836,6 +853,16 @@ impl DurableEngineBuilder { } } + // A required-roles declaration must name a registered workflow. + for name in self.required_roles.keys() { + let configured_prefix = format!("{name}/"); + if !workflows.contains_key(name) + && !workflows.keys().any(|k| k.starts_with(&configured_prefix)) + { + return Err(Error::UnknownWorkflow(name.clone())); + } + } + // Reserve the internal queue first, then add user queues strictly: a // duplicate queue name — or a collision with the reserved internal // queue that resume/fork/debouncer route through — is rejected, not @@ -860,6 +887,7 @@ impl DurableEngineBuilder { max_recovery_attempts: self.max_recovery_attempts, recover_on_launch: self.config.resolve_recover_on_launch(), retention: self.config.retention, + required_roles: self.required_roles, shutdown_token: std::sync::Mutex::new(CancellationToken::new()), deactivated: Arc::new(AtomicBool::new(false)), tasks: TaskTracker::new(), @@ -935,6 +963,7 @@ impl DurableEngine { max_recovery_attempts: 100, recover_on_launch: config.resolve_recover_on_launch(), retention: config.retention, + required_roles: HashMap::new(), shutdown_token: std::sync::Mutex::new(CancellationToken::new()), deactivated: Arc::new(AtomicBool::new(false)), tasks: TaskTracker::new(), @@ -972,6 +1001,7 @@ impl DurableEngine { queues: Vec::new(), listen_filter: None, max_recovery_attempts: 100, + required_roles: HashMap::new(), } } @@ -1032,6 +1062,7 @@ impl DurableEngine { provider: self.provider.clone(), workflows: self.workflows.clone(), queues: self.queues.clone(), + required_roles: self.required_roles.clone(), executor_id: self.executor_id.clone(), app_version: self.app_version.clone(), tasks: self.tasks.clone(), @@ -1104,6 +1135,27 @@ impl DurableEngine { self.queues.insert(queue.name.clone(), Arc::new(queue)); } + /// Declare the roles required to invoke workflow `name`: before the body + /// runs — on **every** execution path (direct, queued, scheduled, child, + /// recovery) — the run's [`AuthContext`](crate::AuthContext) must hold at + /// least one of `roles`, and the first match becomes the run's assumed + /// role. A run that fails the check is finalized `ERROR` with a typed + /// [`Error::NotAuthorized`] — terminal by construction, since the + /// persisted auth context can never satisfy the check on a retry. + /// + /// Must be called before [`launch`](Self::launch), which validates that + /// `name` is a registered workflow. A workflow with no declaration is + /// unrestricted (the default). + pub fn require_roles(&mut self, name: impl Into, roles: I) -> &mut Self + where + I: IntoIterator, + S: Into, + { + self.required_roles + .insert(name.into(), roles.into_iter().map(Into::into).collect()); + self + } + /// Restrict which registered queues this process dispatches at /// [`launch`](Self::launch) to the named subset. By default every registered /// queue gets a dispatcher; call this (before `launch`) to have a process @@ -1379,6 +1431,19 @@ impl DurableEngine { )); } } + // A required-roles declaration for an unregistered workflow is a typo + // that would silently enforce nothing — reject it up front. + for name in self.required_roles.keys() { + let configured_prefix = format!("{name}/"); + if !self.workflows.contains_key(name) + && !self + .workflows + .keys() + .any(|k| k.starts_with(&configured_prefix)) + { + return Err(Error::UnknownWorkflow(name.clone())); + } + } // A prior `shutdown` cancelled the token and closed the tracker. Install a // fresh token (a cancelled one can't be reset) and reopen the tracker, so // this launch's loops and runs are live and the next `shutdown` stops and @@ -2475,6 +2540,7 @@ pub(crate) async fn dispatch_pending_workflows( let _ = run_to_completion( rt.clone(), handler, + record.name.clone(), record.id.clone(), record.input.clone(), record.deadline_ms, @@ -2507,6 +2573,9 @@ pub(crate) struct Runtime { provider: Arc, workflows: HashMap, queues: HashMap>, + /// Required-roles declarations keyed by workflow name; enforced before a + /// body runs, on every execution path. See [`DurableEngine::require_roles`]. + required_roles: HashMap>, executor_id: String, app_version: String, tasks: TaskTracker, @@ -2654,8 +2723,9 @@ impl Runtime { // parents under the workflow (or handler) span that started it. let span = self.workflow_span(&id, name, None, &auth); let rt = self.clone(); + let name = name.to_string(); self.tasks.spawn(async move { - run_to_completion(rt, handler, id, input, deadline_ms, auth, span).await + run_to_completion(rt, handler, name, id, input, deadline_ms, auth, span).await }) } @@ -2895,6 +2965,7 @@ async fn queue_dispatch_loop( let _ = run_to_completion( run_rt, handler, + wf.name, wf.id, wf.input, wf.deadline_ms, @@ -2937,9 +3008,32 @@ async fn queue_dispatch_loop( /// Free function (not a method) so it can run inside a spawned task without /// borrowing the engine. Carries the [`Runtime`] so the workflow can start child /// workflows through its [`DurableContext`]. +/// Check a run's identity against a workflow's required roles: the first +/// required role the caller holds becomes the run's assumed role; no +/// authentication information, or no matching role, is a typed denial. The +/// reference semantics (Python/TS `required_roles`), Rust-shaped. +fn check_required_roles(name: &str, required: &[String], auth: &AuthContext) -> Result { + if auth.authenticated_roles.is_empty() { + return Err(Error::NotAuthorized(format!( + "workflow `{name}` requires a role, but was invoked without authentication information" + ))); + } + required + .iter() + .find(|r| auth.authenticated_roles.contains(r)) + .cloned() + .ok_or_else(|| { + Error::NotAuthorized(format!( + "workflow `{name}` has required roles, but the caller is not authenticated for any of them" + )) + }) +} + +#[allow(clippy::too_many_arguments)] // one arg per run coordinate; grouping would obscure async fn run_to_completion( rt: Arc, handler: WorkflowFn, + name: String, id: String, input: Value, deadline_ms: Option, @@ -2951,6 +3045,23 @@ async fn run_to_completion( let recorder = span.clone(); async move { let provider = rt.provider().clone(); + // Required-roles enforcement, before the body runs — the single gate every + // execution path (direct, queued, scheduled, child, recovery) flows + // through. On a match the role is assumed for the run; a denial is routed + // through the ordinary returned-error path below, finalizing the row + // ERROR: the persisted auth context can never satisfy the check on a + // retry, so leaving the row PENDING would redequeue it forever. + let mut auth = auth; + let denial = match rt.required_roles.get(&name) { + Some(required) => match check_required_roles(&name, required, &auth) { + Ok(role) => { + auth.assumed_role = Some(role); + None + } + Err(e) => Some(e), + }, + None => None, + }; let ctx = DurableContext::new(id.clone(), rt, auth); // Catch a panic in the workflow body so it can't unwind past the status // write below — which would strand the row PENDING with observers waiting @@ -2962,7 +3073,11 @@ async fn run_to_completion( // future is dropped (cancelled at its next await) and the workflow is // marked CANCELLED. `caught` is `Ok(returned)` if the body finished (returned // Ok or Err) or `Err(panic)` if it panicked. - let caught = match deadline_ms { + let caught = if let Some(e) = denial { + drop(run); // the body never starts on a denial + Ok(Err(e)) + } else { + match deadline_ms { Some(dl) => { let remaining = (dl - chrono::Utc::now().timestamp_millis()).max(0) as u64; match tokio::time::timeout(Duration::from_millis(remaining), run).await { @@ -2977,7 +3092,8 @@ async fn run_to_completion( } } } - None => run.await, + None => run.await, + } }; // A panic in the workflow body is treated as a *recoverable* failure, like a diff --git a/src/error.rs b/src/error.rs index 2c94212..f93a596 100644 --- a/src/error.rs +++ b/src/error.rs @@ -23,6 +23,9 @@ pub enum ErrorCode { /// An enqueue was rejected because its deduplication key is already in use /// on the queue. QueueDeduplicated, + /// The caller's authentication context does not satisfy the workflow's + /// required roles. + NotAuthorized, /// The workflow was cancelled; execution was refused. WorkflowCancelled, /// The workflow exceeded its configured recovery-attempt cap and was parked @@ -77,6 +80,13 @@ pub enum Error { #[error("workflow `{0}` does not exist")] NonExistentWorkflow(String), + /// The caller's authentication context does not satisfy a workflow's + /// required roles (see + /// [`DurableEngine::require_roles`](crate::DurableEngine::require_roles)). + /// Terminal when raised at execution: the run is finalized `ERROR`. + #[error("not authorized: {0}")] + NotAuthorized(String), + /// An enqueue collided with an existing workflow holding the same /// deduplication key on the queue. #[error("deduplication id `{dedup_id}` already in use on queue `{queue_name}`")] @@ -243,6 +253,7 @@ impl Error { Error::UnknownQueue(_) => ErrorCode::QueueNotRegistered, Error::NonExistentWorkflow(_) => ErrorCode::NonExistentWorkflow, Error::QueueDeduplicated { .. } => ErrorCode::QueueDeduplicated, + Error::NotAuthorized(_) => ErrorCode::NotAuthorized, Error::Cancelled(_) => ErrorCode::WorkflowCancelled, Error::MaxRecoveryAttemptsExceeded(_) => ErrorCode::MaxRecoveryAttemptsExceeded, Error::ConflictingRegistration(_) => ErrorCode::ConflictingRegistration, diff --git a/src/security.rs b/src/security.rs index 95e2ea5..3923699 100644 --- a/src/security.rs +++ b/src/security.rs @@ -80,6 +80,48 @@ //! The conductor opens no listener at all — it dials out over TLS and serves //! management commands across that connection, so no inbound rule is needed. //! +//! # Authorization +//! +//! durare carries a run's identity end to end — the +//! [`AuthContext`](crate::AuthContext) set at start flows into the workflow, +//! its children, trace spans, and the persisted row. Enforcement is opt-in +//! per workflow: +//! [`require_roles`](crate::DurableEngine::require_roles) declares the roles +//! a caller must hold, and the engine checks the declaration **before the +//! body runs, on every execution path** — direct starts, queued and +//! scheduled runs, children, and recovery. The first required role the +//! caller holds becomes the run's assumed role +//! ([`assumed_role`](crate::DurableContext::assumed_role)); a caller holding +//! none is refused with [`Error::NotAuthorized`](crate::Error::NotAuthorized) +//! and the run is finalized `ERROR` — terminal by construction, because the +//! persisted identity can never satisfy the check on a retry. +//! +//! ``` +//! # use durare::{DurableContext, DurableEngine, Error, InMemoryProvider, Result, WorkflowOptions}; +//! # use std::sync::Arc; +//! # async fn run() -> Result<()> { +//! # let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; +//! engine.register("delete-tenant", |_ctx: DurableContext, tenant: String| async move { +//! Ok::<_, Error>(tenant) +//! }); +//! engine.require_roles("delete-tenant", ["admin"]); +//! # engine.launch().await?; +//! // Runs only when the caller's identity holds "admin": +//! let opts = WorkflowOptions::with_id("del-42") +//! .authenticated_user("alice") +//! .authenticated_roles(["admin"]); +//! # let _ = engine.start::("delete-tenant", "42".into(), opts).await?.await?; +//! # Ok(()) +//! # } +//! ``` +//! +//! Two boundaries to keep straight: the check trusts the `AuthContext` you +//! attach — durare is a library, so *authenticating* the caller (validating +//! a token, resolving their roles) happens in your API layer before the +//! start; and enqueueing is not gated (a `Client` has no registry to consult) +//! — the executor refuses at dequeue, which also protects against callers +//! that bypass your API and enqueue rows directly. +//! //! # Supply chain //! //! CI enforces two independent checks on every push and on a daily schedule: diff --git a/src/serialize.rs b/src/serialize.rs index 9901767..df32b16 100644 --- a/src/serialize.rs +++ b/src/serialize.rs @@ -301,6 +301,14 @@ pub fn encode_error(serializer: &Serializer, err: &Error) -> String { if matches!(serializer, Serializer::Portable) { let env = match err { Error::Portable(pe) => (**pe).clone(), + // The cross-SDK class name for a role denial, so foreign readers + // classify it the way Python raises it. + Error::NotAuthorized(msg) => PortableWorkflowError { + name: "DBOSNotAuthorizedError".to_string(), + message: msg.clone(), + code: None, + data: None, + }, other => PortableWorkflowError { name: PORTABLE_ERROR_NAME.to_string(), message: other.to_string(), diff --git a/tests/roles.rs b/tests/roles.rs new file mode 100644 index 0000000..1f98139 --- /dev/null +++ b/tests/roles.rs @@ -0,0 +1,183 @@ +//! Declarative role-based authorization: `require_roles` declarations are +//! enforced before a workflow body runs, on every execution path. The first +//! required role the caller holds becomes the run's assumed role; a denial is +//! terminal — the row is finalized `ERROR`, never left to be redequeued. + +use durare::{ + DurableContext, DurableEngine, Error, ErrorCode, InMemoryProvider, ListFilter, Result, + WorkflowOptions, WorkflowQueue, +}; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +async fn engine_with_admin_wf() -> Result { + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + engine.register("delete-tenant", |ctx: DurableContext, (): ()| async move { + Ok::<_, Error>(ctx.assumed_role().unwrap_or_default().to_string()) + }); + engine.require_roles("delete-tenant", ["admin", "operator"]); + Ok(engine) +} + +/// A caller holding one of the required roles runs, and the first matching +/// required role becomes the run's assumed role. +#[tokio::test] +async fn matching_role_runs_and_is_assumed() -> Result<()> { + let engine = engine_with_admin_wf().await?; + engine.launch().await?; + + let assumed: String = engine + .start::<(), String>( + "delete-tenant", + (), + WorkflowOptions::with_id("authz-ok") + .authenticated_user("alice") + .authenticated_roles(["viewer", "operator"]), + ) + .await? + .await?; + // "admin" is required first but alice doesn't hold it; "operator" matches. + assert_eq!(assumed, "operator"); + Ok(()) +} + +/// No authentication information at all: denied before the body, row ERROR. +#[tokio::test] +async fn missing_auth_is_denied_terminally() -> Result<()> { + let runs = Arc::new(AtomicU32::new(0)); + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + let counter = runs.clone(); + engine.register("delete-tenant", move |_ctx: DurableContext, (): ()| { + let counter = counter.clone(); + async move { + counter.fetch_add(1, Ordering::SeqCst); + Ok::<_, Error>(String::new()) + } + }); + engine.require_roles("delete-tenant", ["admin", "operator"]); + engine.launch().await?; + + let err = engine + .start::<(), String>("delete-tenant", (), WorkflowOptions::with_id("authz-none")) + .await? + .await + .expect_err("no auth info"); + assert_eq!(err.code(), ErrorCode::NotAuthorized, "{err}"); + assert_eq!(runs.load(Ordering::SeqCst), 0, "body never ran"); + + let row = &engine + .list_workflows(&ListFilter { + workflow_ids: vec!["authz-none".into()], + ..Default::default() + }) + .await?[0]; + assert_eq!(row.status, "ERROR", "denial is finalized, not left pending"); + assert!( + row.error + .as_deref() + .unwrap_or_default() + .contains("requires a role"), + "recorded: {:?}", + row.error + ); + Ok(()) +} + +/// Roles present but none match: denied with the has-roles message. +#[tokio::test] +async fn wrong_roles_are_denied() -> Result<()> { + let engine = engine_with_admin_wf().await?; + engine.launch().await?; + + let err = engine + .start::<(), String>( + "delete-tenant", + (), + WorkflowOptions::with_id("authz-wrong").authenticated_roles(["viewer"]), + ) + .await? + .await + .expect_err("no matching role"); + assert_eq!(err.code(), ErrorCode::NotAuthorized); + assert!( + err.to_string().contains("not authenticated for any"), + "{err}" + ); + Ok(()) +} + +/// The queued path: an unauthorized enqueue is dequeued once, denied, and +/// finalized ERROR — it does not loop through the dispatcher forever. +#[tokio::test] +async fn queued_denial_finalizes_instead_of_looping() -> Result<()> { + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + engine.register("guarded", |_ctx: DurableContext, (): ()| async move { + Ok::<_, Error>(()) + }); + engine.require_roles("guarded", ["admin"]); + engine.register_queue(WorkflowQueue::new("authz-q")); + engine.launch().await?; + + engine + .start::<(), ()>( + "guarded", + (), + WorkflowOptions { + workflow_id: Some("authz-queued".into()), + queue: Some("authz-q".into()), + ..Default::default() + }, + ) + .await?; + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + let row = engine + .list_workflows(&ListFilter { + workflow_ids: vec!["authz-queued".into()], + ..Default::default() + }) + .await? + .remove(0); + if row.status == "ERROR" { + break; + } + assert!( + std::time::Instant::now() < deadline, + "queued denial not finalized within 5s (status {})", + row.status + ); + tokio::time::sleep(Duration::from_millis(25)).await; + } + Ok(()) +} + +/// Workflows without a declaration are unrestricted, authenticated or not. +#[tokio::test] +async fn undeclared_workflows_are_unrestricted() -> Result<()> { + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + engine.register("open", |_ctx: DurableContext, (): ()| async move { + Ok::<_, Error>(()) + }); + engine.launch().await?; + engine + .start::<(), ()>("open", (), WorkflowOptions::with_id("authz-open")) + .await? + .await?; + Ok(()) +} + +/// A declaration for an unregistered workflow is a configuration typo, +/// rejected at launch. +#[tokio::test] +async fn declaration_for_unknown_workflow_is_rejected_at_launch() -> Result<()> { + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + engine.register("real", |_ctx: DurableContext, (): ()| async move { + Ok::<_, Error>(()) + }); + engine.require_roles("no-such-workflow", ["admin"]); + let err = engine.launch().await.expect_err("typo declaration"); + assert!(err.to_string().contains("no-such-workflow"), "{err}"); + Ok(()) +}