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