Declarative role-based authorization: required roles enforced

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

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<RetentionPolicy>,
+    /// Required-roles declarations keyed by workflow name; see
+    /// [`require_roles`](Self::require_roles).
+    required_roles: HashMap<String, Vec<String>>,
     /// 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<WorkflowQueue>,
     listen_filter: Option<std::collections::HashSet<String>>,
     max_recovery_attempts: i32,
+    required_roles: HashMap<String, Vec<String>>,
 }
 
 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<I, S>(&mut self, name: impl Into<String>, roles: I) -> &mut Self
+    where
+        I: IntoIterator<Item = S>,
+        S: Into<String>,
+    {
+        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<I, S>(&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<I, S>(&mut self, name: impl Into<String>, roles: I) -> &mut Self
+    where
+        I: IntoIterator<Item = S>,
+        S: Into<String>,
+    {
+        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<dyn StateProvider>,
     workflows: HashMap<String, WorkflowFn>,
     queues: HashMap<String, Arc<WorkflowQueue>>,
+    /// Required-roles declarations keyed by workflow name; enforced before a
+    /// body runs, on every execution path. See [`DurableEngine::require_roles`].
+    required_roles: HashMap<String, Vec<String>>,
     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<String> {
+    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<Runtime>,
     handler: WorkflowFn,
+    name: String,
     id: String,
     input: Value,
     deadline_ms: Option<i64>,
@@ -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

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,

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::<String, String>("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:

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(),