1use 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
22pub const NAMESPACE_ATTRIBUTE: &str = "aion.namespace";
25
26pub const TASK_QUEUE_ATTRIBUTE: &str = aion_core::START_TIME_TASK_QUEUE_ATTRIBUTE;
37
38pub const DISPLAY_NAME_ATTRIBUTE: &str = aion_core::DISPLAY_NAME_ATTRIBUTE;
48
49#[derive(Clone, Copy, Debug, Eq, PartialEq)]
53pub(crate) enum GrantSource {
54 NamespacesHeader,
57 TokenClaim,
59 Operator,
65}
66
67impl GrantSource {
68 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#[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 deploy: bool,
88 all_namespaces: bool,
93}
94
95impl CallerIdentity {
96 #[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 #[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 #[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 #[must_use]
156 pub fn with_deploy(mut self, deploy: bool) -> Self {
157 self.deploy = deploy;
158 self
159 }
160
161 #[must_use]
163 pub const fn deploy_granted(&self) -> bool {
164 self.deploy
165 }
166
167 #[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 #[must_use]
182 pub fn subject(&self) -> &str {
183 &self.subject
184 }
185
186 #[must_use]
191 pub fn namespaces(&self) -> Vec<String> {
192 self.namespaces.iter().cloned().collect()
193 }
194
195 #[must_use]
200 pub const fn all_namespaces(&self) -> bool {
201 self.all_namespaces
202 }
203
204 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 pub(crate) const fn grant_source(&self) -> GrantSource {
223 self.grant_source
224 }
225}
226
227#[derive(Clone)]
229pub struct ScopedEngine {
230 namespace: String,
231 engine: Option<Arc<Engine>>,
232}
233
234impl ScopedEngine {
235 #[must_use]
237 pub fn namespace(&self) -> &str {
238 &self.namespace
239 }
240
241 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#[derive(Clone, Debug, Eq, PartialEq)]
261pub struct WorkflowAttribution {
262 pub namespace: String,
264 pub workflow_type: Option<String>,
267}
268
269#[async_trait]
275pub trait WorkflowNamespaceSource: Send + Sync {
276 async fn workflow_attribution(
283 &self,
284 workflow_id: &WorkflowId,
285 ) -> Result<Option<WorkflowAttribution>, ServerError>;
286}
287
288struct 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 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#[derive(Clone, Default)]
334pub struct StaticWorkflowNamespaces {
335 inner: Arc<RwLock<HashMap<WorkflowId, WorkflowAttribution>>>,
336}
337
338impl StaticWorkflowNamespaces {
339 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 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#[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 #[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 #[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 #[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 #[must_use]
465 pub const fn mode(&self) -> &NamespaceMode {
466 &self.mode
467 }
468
469 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 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 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 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 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 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 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 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 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 #[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 #[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 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 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 #[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}