Skip to main content

aion_server/namespace/
resolver.rs

1//! Namespace resolver type wired into shared state.
2//!
3//! Workflow→namespace ownership is a projection of durable history: the server
4//! records the owning namespace as the `aion.namespace` search attribute when a
5//! workflow starts, and verification folds that attribute back out of the
6//! workflow's recorded events. Nothing about ownership lives only in memory, so
7//! a server restart can never orphan a workflow from its namespace.
8
9use std::collections::{BTreeSet, HashMap};
10use std::sync::{Arc, RwLock};
11
12use aion::Engine;
13use aion_core::{ScheduleId, SearchAttributeValue, WorkflowId, search_attributes_from_events};
14use aion_proto::WireError;
15use async_trait::async_trait;
16
17use crate::config::{NamespaceConfig, NamespaceMode};
18use crate::error::ServerError;
19
20use super::schedule_source::{HistoryScheduleNamespaceSource, ScheduleNamespaceSource};
21
22/// Search attribute name that records the owning namespace of every workflow
23/// started through this server — the same spelling the visibility projection
24/// places a run by ([`aion_core::NAMESPACE_ATTRIBUTE`]).
25pub const NAMESPACE_ATTRIBUTE: &str = aion_core::NAMESPACE_ATTRIBUTE;
26
27/// Search attribute name that records the default task queue a workflow was
28/// started against (the namespace × `task_queue` targeting story). Recorded
29/// durably in the same atomic append as `WorkflowStarted`, mirroring
30/// [`NAMESPACE_ATTRIBUTE`], so the start-time queue selection survives replay
31/// and failover. Absent when the start did not select a task queue (the
32/// workflow falls back to the namespace's default queue).
33///
34/// Aliases [`aion_core::START_TIME_TASK_QUEUE_ATTRIBUTE`] rather than
35/// redeclaring the literal, so the server's recorded attribute and the engine's
36/// history-derived start-time queue (#144) cannot drift.
37pub const TASK_QUEUE_ATTRIBUTE: &str = aion_core::START_TIME_TASK_QUEUE_ATTRIBUTE;
38
39/// Search attribute name that carries a workflow run's operator-facing display
40/// name (#211). Recorded durably in the same atomic append as `WorkflowStarted`
41/// when the start named one, and re-recorded by every rename, so history keeps
42/// every name the run has worn. A label over the UUID identity, never an
43/// address: nothing resolves a workflow by name.
44///
45/// Aliases [`aion_core::DISPLAY_NAME_ATTRIBUTE`] rather than redeclaring the
46/// literal, so the server's recorded attribute and the engine's history-derived
47/// display name cannot drift.
48pub const DISPLAY_NAME_ATTRIBUTE: &str = aion_core::DISPLAY_NAME_ATTRIBUTE;
49
50/// Where a caller's grants came from, so a denial message can point the
51/// operator at the knob that actually carries the grant (the development
52/// headers, or the validated token's claims).
53#[derive(Clone, Copy, Debug, Eq, PartialEq)]
54pub(crate) enum GrantSource {
55    /// Grants parsed from the development `x-aion-namespaces` /
56    /// `x-aion-deploy` headers.
57    NamespacesHeader,
58    /// Grants carried by a validated token's claims.
59    TokenClaim,
60    /// Full access granted server-side because no auth is configured
61    /// (single-tenant operator mode): the caller is the operator and holds
62    /// every namespace plus the deployment-wide deploy grant. This grant is
63    /// decided at request time from `auth.enabled == false`, never asserted by
64    /// the client.
65    Operator,
66}
67
68impl GrantSource {
69    /// Stable label for audit log fields.
70    pub(crate) const fn label(self) -> &'static str {
71        match self {
72            Self::NamespacesHeader => "header",
73            Self::TokenClaim => "token_claim",
74            Self::Operator => "operator",
75        }
76    }
77}
78
79/// Authenticated caller metadata supplied by an adapter boundary.
80#[derive(Clone, Debug, Eq, PartialEq)]
81pub struct CallerIdentity {
82    subject: String,
83    namespaces: BTreeSet<String>,
84    denial_reason: Option<String>,
85    grant_source: GrantSource,
86    /// Whether the caller holds the deployment-wide deploy grant (the
87    /// `deploy` token claim, or the `x-aion-deploy` development header).
88    deploy: bool,
89    /// Whether the caller holds the `assistant.sessions` grant (the
90    /// `assistant.sessions` token claim, or the `x-aion-assistant-sessions`
91    /// development header): permission to start and drive server-owned
92    /// assistant harness sessions.
93    assistant_sessions: bool,
94    /// Whether the caller holds access to every namespace without enumerating
95    /// them. Set only by [`CallerIdentity::operator`] (auth-off single-tenant
96    /// operator mode); always `false` for header- and token-sourced
97    /// identities, which must enumerate their grants.
98    all_namespaces: bool,
99}
100
101impl CallerIdentity {
102    /// Build a caller identity whose namespace grants came from the
103    /// development `x-aion-namespaces` header.
104    #[must_use]
105    pub fn new(subject: impl Into<String>, namespaces: impl IntoIterator<Item = String>) -> Self {
106        Self {
107            subject: subject.into(),
108            namespaces: namespaces.into_iter().collect(),
109            denial_reason: None,
110            grant_source: GrantSource::NamespacesHeader,
111            deploy: false,
112            assistant_sessions: false,
113            all_namespaces: false,
114        }
115    }
116
117    /// Build a caller identity whose namespace grants came from a validated
118    /// token's namespace claim (the real JWT path), so denial messages direct
119    /// the operator to the token grant instead of the development header.
120    #[must_use]
121    pub fn from_token_claims(
122        subject: impl Into<String>,
123        namespaces: impl IntoIterator<Item = String>,
124    ) -> Self {
125        Self {
126            subject: subject.into(),
127            namespaces: namespaces.into_iter().collect(),
128            denial_reason: None,
129            grant_source: GrantSource::TokenClaim,
130            deploy: false,
131            assistant_sessions: false,
132            all_namespaces: false,
133        }
134    }
135
136    /// Build the single-tenant operator identity: full access to every
137    /// namespace plus EVERY grant word in
138    /// [`crate::namespace::grants::GRANT_WORDS`], with no namespaces
139    /// enumerated.
140    ///
141    /// This is the server's request-time decision when no auth is configured
142    /// (`auth.enabled == false`): the caller IS the operator. It is constructed
143    /// only at an adapter boundary that has already established auth is off; it
144    /// must never be reachable on the auth-enabled path, where grants come from
145    /// validated token claims (or the development-token path).
146    #[must_use]
147    pub fn operator(subject: impl Into<String>) -> Self {
148        Self {
149            subject: subject.into(),
150            namespaces: BTreeSet::new(),
151            denial_reason: None,
152            grant_source: GrantSource::Operator,
153            deploy: true,
154            assistant_sessions: true,
155            all_namespaces: true,
156        }
157    }
158
159    /// Attach the deployment-wide deploy grant decision to this identity.
160    ///
161    /// The grant is engine-global, never namespace-scoped: loading a package
162    /// re-points routing for a workflow type that is startable from every
163    /// namespace, so a namespace-valued grant would promise an isolation the
164    /// engine does not provide.
165    #[must_use]
166    pub fn with_deploy(mut self, deploy: bool) -> Self {
167        self.deploy = deploy;
168        self
169    }
170
171    /// Whether the caller holds the deployment-wide deploy grant.
172    #[must_use]
173    pub const fn deploy_granted(&self) -> bool {
174        self.deploy
175    }
176
177    /// Attach the `assistant.sessions` grant decision to this identity.
178    ///
179    /// Like [`Self::with_deploy`], the grant is deployment-wide rather than
180    /// namespace-scoped: an assistant session is a server-owned harness
181    /// process, not a record inside one namespace, so a namespace-valued grant
182    /// would promise an isolation the session boundary does not provide.
183    #[must_use]
184    pub fn with_assistant_sessions(mut self, assistant_sessions: bool) -> Self {
185        self.assistant_sessions = assistant_sessions;
186        self
187    }
188
189    /// Whether the caller holds the `assistant.sessions` grant: permission to
190    /// start and drive server-owned assistant harness sessions.
191    #[must_use]
192    pub const fn assistant_sessions_granted(&self) -> bool {
193        self.assistant_sessions
194    }
195
196    /// Build a caller identity that must be denied with a transport-specific reason.
197    #[must_use]
198    pub fn denied(subject: impl Into<String>, reason: impl Into<String>) -> Self {
199        Self {
200            subject: subject.into(),
201            namespaces: BTreeSet::new(),
202            denial_reason: Some(reason.into()),
203            grant_source: GrantSource::NamespacesHeader,
204            deploy: false,
205            assistant_sessions: false,
206            all_namespaces: false,
207        }
208    }
209
210    /// Caller subject as authenticated by the transport.
211    #[must_use]
212    pub fn subject(&self) -> &str {
213        &self.subject
214    }
215
216    /// Namespaces this caller is authorized for, in sorted order.
217    ///
218    /// Backed by a [`BTreeSet`], so the returned vector is already
219    /// lexicographically ordered with no duplicates.
220    #[must_use]
221    pub fn namespaces(&self) -> Vec<String> {
222        self.namespaces.iter().cloned().collect()
223    }
224
225    /// Whether this caller holds access to every namespace without enumerating
226    /// them. True only for the single-tenant operator identity (auth-off
227    /// operator mode); the all-access grant is signaled here, not by
228    /// [`Self::namespaces`], which stays the explicit (empty) set.
229    #[must_use]
230    pub const fn all_namespaces(&self) -> bool {
231        self.all_namespaces
232    }
233
234    /// Whether this caller may access `namespace`: the all-namespaces operator
235    /// always can; an enumerated caller can only for a granted name.
236    ///
237    /// This is the existence-leak boundary for read enumeration (`GET
238    /// /namespaces`): a caller must never learn that a namespace it cannot
239    /// access exists, so the durable set is filtered through this predicate at
240    /// the read hop, exactly as [`Self::resolve`]'s grant check gates the access
241    /// hop.
242    pub(crate) fn can_access(&self, namespace: &str) -> bool {
243        self.all_namespaces || self.namespaces.contains(namespace)
244    }
245
246    pub(crate) fn denial_reason(&self) -> Option<&str> {
247        self.denial_reason.as_deref()
248    }
249
250    /// Where this caller's grants came from, for grant-source-aware denials
251    /// and audit fields.
252    pub(crate) const fn grant_source(&self) -> GrantSource {
253        self.grant_source
254    }
255}
256
257/// Namespace-scoped access to the embedded engine.
258#[derive(Clone)]
259pub struct ScopedEngine {
260    namespace: String,
261    engine: Option<Arc<Engine>>,
262}
263
264impl ScopedEngine {
265    /// Authorized namespace attached to this engine access.
266    #[must_use]
267    pub fn namespace(&self) -> &str {
268        &self.namespace
269    }
270
271    /// Borrow the authorized engine handle for adapter code after guard approval.
272    ///
273    /// # Errors
274    ///
275    /// Returns [`ServerError::Config`] only for resolver instances constructed
276    /// without an engine for unit tests.
277    pub fn engine(&self) -> Result<&Arc<Engine>, ServerError> {
278        self.engine.as_ref().ok_or_else(|| ServerError::Config {
279            message: "namespace resolver has no engine handle".to_owned(),
280        })
281    }
282}
283
284/// Durable per-workflow attribution facts projected from recorded history.
285///
286/// Namespace ownership and workflow type are both immutable projections of the
287/// same durable history (ownership is recorded atomically with the
288/// `WorkflowStarted` batch; the type is the most recent run's recorded
289/// `WorkflowStarted` type), so one read serves both consumers.
290#[derive(Clone, Debug, Eq, PartialEq)]
291pub struct WorkflowAttribution {
292    /// Namespace recorded as the workflow's owner.
293    pub namespace: String,
294    /// Workflow type recorded by the most recent `WorkflowStarted` event, or
295    /// [`None`] when the history records no started run.
296    pub workflow_type: Option<String>,
297}
298
299/// Durable source of workflow→namespace ownership and type attribution facts.
300///
301/// The production implementation projects attribution from recorded workflow
302/// history; tests substitute a static fixture to prove adapter-boundary
303/// denials without an engine.
304#[async_trait]
305pub trait WorkflowNamespaceSource: Send + Sync {
306    /// Returns the attribution recorded for a workflow, or [`None`] when the
307    /// workflow is unknown or recorded no namespace attribute.
308    ///
309    /// # Errors
310    ///
311    /// Returns [`ServerError`] when the underlying ownership data cannot be read.
312    async fn workflow_attribution(
313        &self,
314        workflow_id: &WorkflowId,
315    ) -> Result<Option<WorkflowAttribution>, ServerError>;
316}
317
318/// Production attribution source: folds the `aion.namespace` search attribute
319/// and the most recent `WorkflowStarted` type out of the workflow's durable
320/// event history in a single read.
321struct HistoryNamespaceSource {
322    engine: Arc<Engine>,
323}
324
325#[async_trait]
326impl WorkflowNamespaceSource for HistoryNamespaceSource {
327    async fn workflow_attribution(
328        &self,
329        workflow_id: &WorkflowId,
330    ) -> Result<Option<WorkflowAttribution>, ServerError> {
331        let history = self
332            .engine
333            .store()
334            .read_history(workflow_id)
335            .await
336            .map_err(ServerError::from)?;
337        let namespace = match search_attributes_from_events(&history).remove(NAMESPACE_ATTRIBUTE) {
338            Some(SearchAttributeValue::String(namespace)) => namespace,
339            Some(other) => {
340                return Err(ServerError::Config {
341                    message: format!(
342                        "workflow {workflow_id} recorded a non-string {NAMESPACE_ATTRIBUTE} search attribute: {other:?}"
343                    ),
344                });
345            }
346            None => return Ok(None),
347        };
348        // Continue-as-new runs share one history; the most recent
349        // `WorkflowStarted` carries the current run's workflow type.
350        let workflow_type = history.iter().rev().find_map(|event| match event {
351            aion_core::Event::WorkflowStarted { workflow_type, .. } => Some(workflow_type.clone()),
352            _ => None,
353        });
354        Ok(Some(WorkflowAttribution {
355            namespace,
356            workflow_type,
357        }))
358    }
359}
360
361/// Static workflow→namespace fixture for adapter-boundary tests and alternate
362/// wiring that must authorize without an engine handle.
363#[derive(Clone, Default)]
364pub struct StaticWorkflowNamespaces {
365    inner: Arc<RwLock<HashMap<WorkflowId, WorkflowAttribution>>>,
366}
367
368impl StaticWorkflowNamespaces {
369    /// Record that a workflow is owned by a namespace, with no recorded
370    /// workflow type (the fixture equivalent of a history without a
371    /// `WorkflowStarted` event).
372    ///
373    /// # Errors
374    ///
375    /// Returns [`ServerError::LockPoisoned`] if the fixture lock was poisoned.
376    pub fn record(&self, workflow_id: WorkflowId, namespace: &str) -> Result<(), ServerError> {
377        self.insert(
378            workflow_id,
379            WorkflowAttribution {
380                namespace: namespace.to_owned(),
381                workflow_type: None,
382            },
383        )
384    }
385
386    /// Record that a workflow is owned by a namespace and carries a recorded
387    /// workflow type.
388    ///
389    /// # Errors
390    ///
391    /// Returns [`ServerError::LockPoisoned`] if the fixture lock was poisoned.
392    pub fn record_with_type(
393        &self,
394        workflow_id: WorkflowId,
395        namespace: &str,
396        workflow_type: &str,
397    ) -> Result<(), ServerError> {
398        self.insert(
399            workflow_id,
400            WorkflowAttribution {
401                namespace: namespace.to_owned(),
402                workflow_type: Some(workflow_type.to_owned()),
403            },
404        )
405    }
406
407    fn insert(
408        &self,
409        workflow_id: WorkflowId,
410        attribution: WorkflowAttribution,
411    ) -> Result<(), ServerError> {
412        let mut ownership = self
413            .inner
414            .write()
415            .map_err(|_| ServerError::lock_poisoned("namespace workflow ownership"))?;
416        ownership.insert(workflow_id, attribution);
417        Ok(())
418    }
419}
420
421#[async_trait]
422impl WorkflowNamespaceSource for StaticWorkflowNamespaces {
423    async fn workflow_attribution(
424        &self,
425        workflow_id: &WorkflowId,
426    ) -> Result<Option<WorkflowAttribution>, ServerError> {
427        let ownership = self
428            .inner
429            .read()
430            .map_err(|_| ServerError::lock_poisoned("namespace workflow ownership"))?;
431        Ok(ownership.get(workflow_id).cloned())
432    }
433}
434
435/// Resolver that authorizes callers and yields namespace-scoped engine access.
436#[derive(Clone)]
437pub struct NamespaceResolver {
438    mode: NamespaceMode,
439    engine: Option<Arc<Engine>>,
440    ownership: Arc<dyn WorkflowNamespaceSource>,
441    schedule_ownership: Arc<dyn ScheduleNamespaceSource>,
442}
443
444impl NamespaceResolver {
445    /// Build a resolver from operator-supplied namespace configuration and the
446    /// engine selected for this deployment.
447    #[must_use]
448    pub fn from_config(config: NamespaceConfig, engine: Arc<Engine>) -> Self {
449        Self {
450            mode: config.mode,
451            ownership: Arc::new(HistoryNamespaceSource {
452                engine: Arc::clone(&engine),
453            }),
454            schedule_ownership: Arc::new(HistoryScheduleNamespaceSource::new(Arc::clone(&engine))),
455            engine: Some(engine),
456        }
457    }
458
459    /// Build a resolver from explicit parts for tests and alternate wiring.
460    #[must_use]
461    pub fn from_parts(
462        mode: NamespaceMode,
463        engine: Option<Arc<Engine>>,
464        ownership: Arc<dyn WorkflowNamespaceSource>,
465        schedule_ownership: Arc<dyn ScheduleNamespaceSource>,
466    ) -> Self {
467        Self {
468            mode,
469            engine,
470            ownership,
471            schedule_ownership,
472        }
473    }
474
475    /// Build a resolver that performs authorization and ownership checks only.
476    ///
477    /// This constructor is intended for adapter-boundary unit tests that must
478    /// prove denied operations do not reach any engine handle.
479    #[must_use]
480    pub fn authorization_only(
481        mode: NamespaceMode,
482        ownership: impl WorkflowNamespaceSource + 'static,
483        schedule_ownership: impl ScheduleNamespaceSource + 'static,
484    ) -> Self {
485        Self::from_parts(
486            mode,
487            None,
488            Arc::new(ownership),
489            Arc::new(schedule_ownership),
490        )
491    }
492
493    /// Inspect the configured namespace mode.
494    #[must_use]
495    pub const fn mode(&self) -> &NamespaceMode {
496        &self.mode
497    }
498
499    /// Borrow the engine handle for engine-global (non-namespace) operations.
500    ///
501    /// # Errors
502    ///
503    /// Returns [`ServerError::Config`] only for resolver instances constructed
504    /// without an engine for unit tests.
505    pub(crate) fn engine(&self) -> Result<&Arc<Engine>, ServerError> {
506        self.engine.as_ref().ok_or_else(|| ServerError::Config {
507            message: "namespace resolver has no engine handle".to_owned(),
508        })
509    }
510
511    /// Shut down the engine owned by this resolver.
512    ///
513    /// # Errors
514    ///
515    /// Returns [`ServerError::Config`] when no engine is attached, or [`ServerError::EngineCall`]
516    /// when the engine rejects shutdown.
517    pub fn shutdown_engine(&self) -> Result<(), ServerError> {
518        self.engine
519            .as_ref()
520            .ok_or_else(|| ServerError::Config {
521                message: "namespace resolver has no engine handle".to_owned(),
522            })?
523            .shutdown()
524            .map_err(ServerError::from)
525    }
526
527    /// Authorize a caller for a requested namespace and return scoped engine
528    /// access if allowed.
529    ///
530    /// # Errors
531    ///
532    /// Returns [`ServerError::Namespace`] when the caller is not authorized for
533    /// the namespace selected by the wire request.
534    pub(super) fn resolve(
535        &self,
536        caller: &CallerIdentity,
537        requested_namespace: &str,
538    ) -> Result<ScopedEngine, ServerError> {
539        if requested_namespace.is_empty() {
540            return Err(ServerError::namespace_denied(
541                "requested namespace must not be empty",
542            ));
543        }
544
545        if let Some(reason) = caller.denial_reason() {
546            return Err(ServerError::namespace_denied(reason));
547        }
548
549        match &self.mode {
550            NamespaceMode::SingleTenant { namespace } if namespace == requested_namespace => {
551                Ok(self.scoped(requested_namespace))
552            }
553            NamespaceMode::SharedEngine if caller.can_access(requested_namespace) => {
554                Ok(self.scoped(requested_namespace))
555            }
556            NamespaceMode::SingleTenant { .. } | NamespaceMode::SharedEngine => {
557                Err(namespace_denied(caller, requested_namespace))
558            }
559        }
560    }
561
562    /// Verify durable workflow ownership against the requested namespace.
563    ///
564    /// `NamespaceDenied` means exactly one thing: the caller has no grant for
565    /// the requested namespace, and that is decided by [`Self::resolve`] before
566    /// this check runs. Workflow-level visibility misses are `NotFound` to
567    /// prevent existence leaks: when the caller's requested namespace is
568    /// granted but the workflow's recorded owner namespace is absent (unknown
569    /// workflow, or no recorded attribute) or different (owned by another
570    /// tenant), both cases return the identical `not_found` wire error with
571    /// the identical message, so a cross-tenant probe is byte-for-byte
572    /// indistinguishable from querying a workflow that never existed.
573    ///
574    /// # Errors
575    ///
576    /// Returns a [`ServerError::Wire`] `not_found` error when the workflow is
577    /// not visible in the requested namespace; ownership-source read failures
578    /// surface as their own typed errors.
579    pub async fn verify_workflow_ownership(
580        &self,
581        namespace: &str,
582        workflow_id: &WorkflowId,
583    ) -> Result<(), ServerError> {
584        match self.workflow_attribution(namespace, workflow_id).await? {
585            Some(_) => Ok(()),
586            None => Err(ServerError::Wire {
587                wire: WireError::not_found(format!("workflow not found in namespace {namespace}")),
588            }),
589        }
590    }
591
592    /// Read a workflow's durable attribution scoped to one namespace.
593    ///
594    /// Returns the recorded attribution only when the workflow's recorded
595    /// owner namespace equals `namespace`. Foreign-owned and unknown workflows
596    /// both yield [`None`] (anti-existence-leak: callers must treat the two
597    /// cases identically and never disclose which one occurred).
598    ///
599    /// This is the single read that serves both the namespace verdict and the
600    /// workflow-type lookup at the streaming seam — one durable history read
601    /// per workflow answers both questions.
602    ///
603    /// # Errors
604    ///
605    /// Returns [`ServerError`] when the underlying ownership data cannot be
606    /// read; callers must fail loudly rather than guessing.
607    pub async fn workflow_attribution(
608        &self,
609        namespace: &str,
610        workflow_id: &WorkflowId,
611    ) -> Result<Option<WorkflowAttribution>, ServerError> {
612        Ok(self
613            .ownership
614            .workflow_attribution(workflow_id)
615            .await?
616            .filter(|attribution| attribution.namespace == namespace))
617    }
618
619    /// Read a workflow's recorded attribution WITHOUT scoping it to a namespace.
620    ///
621    /// [`Self::workflow_attribution`] is the right read whenever the caller
622    /// already named the namespace it is asking about — it answers the scoped
623    /// question and hides everything else behind the anti-existence-leak
624    /// `None`. A fleet-wide SWEEP cannot use it: the sweep starts from a set of
625    /// workflow ids and does not yet know which namespace each belongs to, so
626    /// scoping first would require guessing the answer it is trying to read.
627    ///
628    /// The grant filter is therefore the CALLER's obligation here, and it is not
629    /// optional. Every use must drop entries the caller cannot access — see
630    /// [`CallerIdentity::can_access`] — before anything reaches a response body,
631    /// exactly as the enumeration reads do. Returning the raw attribution keeps
632    /// that filter visible at the sweep, rather than a scoped read silently
633    /// reporting `None` for a workflow the caller could in fact see.
634    ///
635    /// `None` means the workflow recorded no owning namespace at all: an
636    /// unattributed run, not a denied one. The two are different facts and the
637    /// caller must not merge them.
638    ///
639    /// # Errors
640    ///
641    /// Returns [`ServerError`] when the underlying ownership data cannot be
642    /// read; callers must fail loudly rather than guessing.
643    pub async fn recorded_workflow_attribution(
644        &self,
645        workflow_id: &WorkflowId,
646    ) -> Result<Option<WorkflowAttribution>, ServerError> {
647        self.ownership.workflow_attribution(workflow_id).await
648    }
649
650    /// Verify durable schedule ownership against the requested namespace.
651    ///
652    /// `NamespaceDenied` means exactly one thing: the caller has no grant for
653    /// the requested namespace, and that is decided by [`Self::resolve`] before
654    /// this check runs. Schedule-level visibility misses are `NotFound` to
655    /// prevent existence leaks: when the caller's requested namespace is
656    /// granted but the schedule's creation-recorded owner namespace is absent
657    /// (unknown schedule, or no recorded attribute) or different (owned by
658    /// another tenant), both cases return the identical `not_found` wire error
659    /// with the identical message, so a cross-tenant probe is byte-for-byte
660    /// indistinguishable from targeting a schedule that never existed.
661    ///
662    /// # Errors
663    ///
664    /// Returns a [`ServerError::Wire`] `not_found` error when the schedule is
665    /// not visible in the requested namespace; ownership-source read failures
666    /// surface as their own typed errors.
667    pub async fn verify_schedule_ownership(
668        &self,
669        namespace: &str,
670        schedule_id: &ScheduleId,
671    ) -> Result<(), ServerError> {
672        match self
673            .schedule_ownership
674            .schedule_namespace(schedule_id)
675            .await?
676        {
677            Some(owner) if owner == namespace => Ok(()),
678            // Anti-existence-leak: absent and foreign ownership must be one
679            // identical NotFound, never a distinguishable denial.
680            Some(_) | None => Err(ServerError::Wire {
681                wire: WireError::not_found(format!("schedule not found in namespace {namespace}")),
682            }),
683        }
684    }
685
686    fn scoped(&self, namespace: &str) -> ScopedEngine {
687        ScopedEngine {
688            namespace: namespace.to_owned(),
689            engine: self.engine.clone(),
690        }
691    }
692}
693
694fn namespace_denied(caller: &CallerIdentity, requested_namespace: &str) -> ServerError {
695    let hint = match caller.grant_source {
696        GrantSource::NamespacesHeader => format!(
697            "add {requested_namespace} to x-aion-namespaces for subject `{}` or request a namespace listed in that header",
698            caller.subject()
699        ),
700        GrantSource::TokenClaim => format!(
701            "grant {requested_namespace} in the namespace claim of the token minted for subject `{}` or request a namespace the token grants",
702            caller.subject()
703        ),
704        // An operator holds every namespace (`can_access` always true), so this
705        // arm is never reached; keep the match exhaustive without inventing a
706        // misleading hint.
707        GrantSource::Operator => format!(
708            "subject `{}` is the operator and already holds every namespace",
709            caller.subject()
710        ),
711    };
712    ServerError::namespace_denied(format!(
713        "subject not authorized for namespace {requested_namespace}; {hint}"
714    ))
715}
716
717#[cfg(test)]
718mod tests {
719    use super::{
720        CallerIdentity, NamespaceResolver, StaticWorkflowNamespaces, WorkflowNamespaceSource,
721    };
722    use crate::config::NamespaceMode;
723    use crate::namespace::StaticScheduleNamespaces;
724    use aion_core::{ScheduleId, WorkflowId};
725
726    fn resolver(mode: NamespaceMode) -> NamespaceResolver {
727        NamespaceResolver::authorization_only(
728            mode,
729            StaticWorkflowNamespaces::default(),
730            StaticScheduleNamespaces::default(),
731        )
732    }
733
734    #[test]
735    fn shared_engine_authorizes_explicit_caller_grant() -> Result<(), Box<dyn std::error::Error>> {
736        let resolver = resolver(NamespaceMode::SharedEngine);
737        let caller = CallerIdentity::new("alice", [String::from("tenant-a")]);
738
739        let scoped = resolver.resolve(&caller, "tenant-a")?;
740
741        assert_eq!(scoped.namespace(), "tenant-a");
742        Ok(())
743    }
744
745    /// The operator identity (auth-off operator mode) is authorized for any
746    /// namespace without enumerating it, holds the deploy grant, and signals
747    /// all-access through `all_namespaces()` while its explicit namespace set
748    /// stays empty.
749    #[test]
750    fn operator_is_authorized_for_any_namespace() -> Result<(), Box<dyn std::error::Error>> {
751        let resolver = resolver(NamespaceMode::SharedEngine);
752        let operator = CallerIdentity::operator("operator");
753
754        assert!(operator.all_namespaces());
755        assert!(operator.deploy_granted());
756        assert!(operator.namespaces().is_empty());
757
758        assert_eq!(
759            resolver.resolve(&operator, "tenant-a")?.namespace(),
760            "tenant-a"
761        );
762        assert_eq!(
763            resolver.resolve(&operator, "tenant-z")?.namespace(),
764            "tenant-z"
765        );
766        Ok(())
767    }
768
769    #[test]
770    fn shared_engine_denies_missing_caller_grant() {
771        let resolver = resolver(NamespaceMode::SharedEngine);
772        let caller = CallerIdentity::new("alice", [String::from("tenant-a")]);
773
774        let denied = resolver.resolve(&caller, "tenant-b");
775
776        assert!(denied.is_err());
777    }
778
779    #[test]
780    fn single_tenant_authorizes_only_configured_namespace() -> Result<(), Box<dyn std::error::Error>>
781    {
782        let resolver = resolver(NamespaceMode::SingleTenant {
783            namespace: String::from("tenant-a"),
784        });
785        let caller = CallerIdentity::new("alice", [String::from("tenant-b")]);
786
787        let scoped = resolver.resolve(&caller, "tenant-a")?;
788        let denied = resolver.resolve(&caller, "tenant-b");
789
790        assert_eq!(scoped.namespace(), "tenant-a");
791        assert!(denied.is_err());
792        Ok(())
793    }
794
795    /// The denial hint must point at the knob that actually carries the
796    /// caller's grants: the development `x-aion-namespaces` header for
797    /// header-sourced identities, the token's namespace claim for identities
798    /// produced by the JWT path.
799    #[test]
800    fn denial_hint_names_the_grant_source() -> Result<(), Box<dyn std::error::Error>> {
801        let resolver = resolver(NamespaceMode::SharedEngine);
802
803        let header_caller = CallerIdentity::new("alice", [String::from("tenant-a")]);
804        let header_denial = resolver
805            .resolve(&header_caller, "tenant-b")
806            .err()
807            .map(|error| error.to_wire_error())
808            .ok_or("expected header-sourced caller to be denied")?;
809        assert!(
810            header_denial.message.contains("x-aion-namespaces"),
811            "header-path denial must hint the dev header: {}",
812            header_denial.message
813        );
814        assert!(
815            !header_denial.message.contains("namespace claim"),
816            "header-path denial must not hint the token claim: {}",
817            header_denial.message
818        );
819
820        let token_caller = CallerIdentity::from_token_claims("alice", [String::from("tenant-a")]);
821        let token_denial = resolver
822            .resolve(&token_caller, "tenant-b")
823            .err()
824            .map(|error| error.to_wire_error())
825            .ok_or("expected token-sourced caller to be denied")?;
826        assert!(
827            token_denial.message.contains("namespace claim"),
828            "JWT-path denial must hint the token's namespace claim: {}",
829            token_denial.message
830        );
831        assert!(
832            !token_denial.message.contains("x-aion-namespaces"),
833            "JWT-path denial must not hint the dev header: {}",
834            token_denial.message
835        );
836        Ok(())
837    }
838
839    #[test]
840    fn empty_namespace_is_denied_before_scoping() {
841        let resolver = resolver(NamespaceMode::SharedEngine);
842        let caller = CallerIdentity::new("alice", [String::new()]);
843
844        let denied = resolver.resolve(&caller, "");
845
846        assert!(denied.is_err());
847    }
848
849    #[tokio::test]
850    async fn ownership_misses_are_indistinguishable_not_found()
851    -> Result<(), Box<dyn std::error::Error>> {
852        let ownership = StaticWorkflowNamespaces::default();
853        let owned = WorkflowId::new(uuid::Uuid::from_u128(1));
854        let unknown = WorkflowId::new(uuid::Uuid::from_u128(2));
855        ownership.record(owned.clone(), "tenant-a")?;
856        let resolver = NamespaceResolver::authorization_only(
857            NamespaceMode::SharedEngine,
858            ownership,
859            StaticScheduleNamespaces::default(),
860        );
861
862        resolver
863            .verify_workflow_ownership("tenant-a", &owned)
864            .await?;
865
866        // Foreign-owned and nonexistent workflows must produce byte-for-byte
867        // identical NotFound wire errors (anti-existence-leak), never
868        // NamespaceDenied.
869        let foreign = resolver
870            .verify_workflow_ownership("tenant-b", &owned)
871            .await
872            .err()
873            .map(|error| error.to_wire_error())
874            .ok_or("expected foreign-owned workflow to be rejected")?;
875        let absent = resolver
876            .verify_workflow_ownership("tenant-b", &unknown)
877            .await
878            .err()
879            .map(|error| error.to_wire_error())
880            .ok_or("expected unknown workflow to be rejected")?;
881
882        assert_eq!(foreign.code, aion_proto::WireErrorCode::NotFound);
883        assert_eq!(foreign, absent);
884        assert_eq!(foreign.message, "workflow not found in namespace tenant-b");
885
886        let absent_in_granted = resolver
887            .verify_workflow_ownership("tenant-a", &unknown)
888            .await
889            .err()
890            .map(|error| error.to_wire_error())
891            .ok_or("expected unknown workflow to be rejected in granted namespace")?;
892        assert_eq!(absent_in_granted.code, aion_proto::WireErrorCode::NotFound);
893        assert_eq!(
894            absent_in_granted.message,
895            "workflow not found in namespace tenant-a"
896        );
897        Ok(())
898    }
899
900    #[tokio::test]
901    async fn schedule_ownership_misses_are_indistinguishable_not_found()
902    -> Result<(), Box<dyn std::error::Error>> {
903        let schedule_ownership = StaticScheduleNamespaces::default();
904        let owned = ScheduleId::new(uuid::Uuid::from_u128(1));
905        let unknown = ScheduleId::new(uuid::Uuid::from_u128(2));
906        schedule_ownership.record(owned.clone(), "tenant-a")?;
907        let resolver = NamespaceResolver::authorization_only(
908            NamespaceMode::SharedEngine,
909            StaticWorkflowNamespaces::default(),
910            schedule_ownership,
911        );
912
913        resolver
914            .verify_schedule_ownership("tenant-a", &owned)
915            .await?;
916
917        // Foreign-owned and nonexistent schedules must produce byte-for-byte
918        // identical NotFound wire errors (anti-existence-leak), never
919        // NamespaceDenied.
920        let foreign = resolver
921            .verify_schedule_ownership("tenant-b", &owned)
922            .await
923            .err()
924            .map(|error| error.to_wire_error())
925            .ok_or("expected foreign-owned schedule to be rejected")?;
926        let absent = resolver
927            .verify_schedule_ownership("tenant-b", &unknown)
928            .await
929            .err()
930            .map(|error| error.to_wire_error())
931            .ok_or("expected unknown schedule to be rejected")?;
932
933        assert_eq!(foreign.code, aion_proto::WireErrorCode::NotFound);
934        assert_eq!(foreign, absent);
935        assert_eq!(foreign.message, "schedule not found in namespace tenant-b");
936
937        let absent_in_granted = resolver
938            .verify_schedule_ownership("tenant-a", &unknown)
939            .await
940            .err()
941            .map(|error| error.to_wire_error())
942            .ok_or("expected unknown schedule to be rejected in granted namespace")?;
943        assert_eq!(absent_in_granted.code, aion_proto::WireErrorCode::NotFound);
944        assert_eq!(
945            absent_in_granted.message,
946            "schedule not found in namespace tenant-a"
947        );
948        Ok(())
949    }
950
951    #[tokio::test]
952    async fn static_source_reports_recorded_namespace() -> Result<(), Box<dyn std::error::Error>> {
953        let ownership = StaticWorkflowNamespaces::default();
954        let workflow_id = WorkflowId::new(uuid::Uuid::from_u128(3));
955        ownership.record(workflow_id.clone(), "tenant-a")?;
956
957        assert_eq!(
958            ownership.workflow_attribution(&workflow_id).await?,
959            Some(super::WorkflowAttribution {
960                namespace: String::from("tenant-a"),
961                workflow_type: None,
962            })
963        );
964        Ok(())
965    }
966
967    #[tokio::test]
968    async fn static_source_reports_recorded_workflow_type() -> Result<(), Box<dyn std::error::Error>>
969    {
970        let ownership = StaticWorkflowNamespaces::default();
971        let workflow_id = WorkflowId::new(uuid::Uuid::from_u128(4));
972        ownership.record_with_type(workflow_id.clone(), "tenant-a", "checkout")?;
973
974        assert_eq!(
975            ownership.workflow_attribution(&workflow_id).await?,
976            Some(super::WorkflowAttribution {
977                namespace: String::from("tenant-a"),
978                workflow_type: Some(String::from("checkout")),
979            })
980        );
981        Ok(())
982    }
983
984    /// The namespace-scoped attribution read must hide foreign and unknown
985    /// workflows identically (anti-existence-leak) while exposing the recorded
986    /// type for owned workflows.
987    #[tokio::test]
988    async fn scoped_attribution_hides_foreign_and_unknown_identically()
989    -> Result<(), Box<dyn std::error::Error>> {
990        let ownership = StaticWorkflowNamespaces::default();
991        let owned = WorkflowId::new(uuid::Uuid::from_u128(5));
992        let foreign = WorkflowId::new(uuid::Uuid::from_u128(6));
993        let unknown = WorkflowId::new(uuid::Uuid::from_u128(7));
994        ownership.record_with_type(owned.clone(), "tenant-a", "checkout")?;
995        ownership.record_with_type(foreign.clone(), "tenant-b", "checkout")?;
996        let resolver = NamespaceResolver::authorization_only(
997            NamespaceMode::SharedEngine,
998            ownership,
999            StaticScheduleNamespaces::default(),
1000        );
1001
1002        let visible = resolver
1003            .workflow_attribution("tenant-a", &owned)
1004            .await?
1005            .ok_or("owned workflow attribution must be visible")?;
1006        assert_eq!(visible.workflow_type.as_deref(), Some("checkout"));
1007        assert_eq!(
1008            resolver.workflow_attribution("tenant-a", &foreign).await?,
1009            None
1010        );
1011        assert_eq!(
1012            resolver.workflow_attribution("tenant-a", &unknown).await?,
1013            None
1014        );
1015        Ok(())
1016    }
1017}