af_workflow/
admission.rs

1//! Narrow workflow deployment grants, supplied by the host's entitlement mapping.
2use af_context::{SubjectId, WorkflowResourceId, WorkflowSourceId};
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5use std::collections::BTreeSet;
6
7/// Exact source/event authority; a binding must retain this payload restriction.
8#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
9#[serde(deny_unknown_fields)]
10pub struct WorkflowSourceGrant {
11    /// Host's stable source identity.
12    pub source: WorkflowSourceId,
13    /// Exact event type, without wildcard expansion.
14    pub event_type: String,
15    /// JSON containment restriction, using the same semantics as trigger bindings.
16    pub predicate: Value,
17}
18/// Host-managed authority for ordinary workflow authors; absent grants deny access.
19#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
20#[serde(deny_unknown_fields)]
21pub struct WorkflowAdmissionPolicy {
22    /// Revoke new publication, activation, binding, resume and dispatch when false.
23    pub enabled: bool,
24    /// Maximum active/draining instances before a new authoring activation is accepted.
25    /// Existing-instance retries do not consume a second slot.
26    pub max_active_instances: u32,
27    /// Exact source/event grants; at most 64.
28    pub sources: Vec<WorkflowSourceGrant>,
29    /// Stable connection and external resource identities the subject may use.
30    pub resources: BTreeSet<WorkflowResourceId>,
31    /// Allowed trigger/data/action provider names, without wildcards.
32    pub providers: BTreeSet<String>,
33    /// Allowed execution modes.
34    pub modes: Vec<crate::ExecutionMode>,
35}
36impl WorkflowAdmissionPolicy {
37    /// Bound the grant and reject malformed restrictions.
38    pub fn validate(&self) -> Result<(), String> {
39        // ponytail: one owned policy row serializes deployment admission; split only if measured contention warrants it.
40        if self.sources.len() > 64
41            || self.resources.len() > 256
42            || self.providers.len() > 256
43            || self.modes.len() > 4
44        {
45            return Err("workflow admission exceeds grant collection bounds".into());
46        }
47        if self
48            .sources
49            .iter()
50            .any(|grant| grant.event_type.trim().is_empty() || !grant.predicate.is_object())
51            || self
52                .providers
53                .iter()
54                .any(|provider| provider.trim().is_empty())
55        {
56            return Err(
57                "workflow admission requires exact providers and source/object restrictions".into(),
58            );
59        }
60        if serde_json::to_vec(self)
61            .map_err(|error| error.to_string())?
62            .len()
63            > 65_536
64        {
65            return Err("workflow admission exceeds 64 KiB".into());
66        }
67        Ok(())
68    }
69    /// Authorize the profile against current host grants and the caller's permissions.
70    pub fn authorize_profile(
71        &self,
72        context: &af_context::RequestContext,
73        profile: &crate::ExecutionProfileRevision,
74    ) -> Result<(), String> {
75        if !self.enabled
76            || !self.modes.contains(&profile.mode)
77            || [
78                &profile.trigger_provider,
79                &profile.data_provider,
80                &profile.action_provider,
81            ]
82            .iter()
83            .any(|provider| !self.providers.contains(*provider))
84            || profile.connection_bindings.values().any(|resource| {
85                !self
86                    .resources
87                    .iter()
88                    .any(|allowed| allowed.as_str() == resource)
89            })
90        {
91            return Err("workflow profile exceeds current mode/provider/resource admission".into());
92        }
93        if let Some(permissions) = profile.policy_bundle.get("permissions") {
94            let permissions = permissions
95                .as_array()
96                .ok_or("workflow profile permissions must be an array")?;
97            if permissions.iter().any(|permission| {
98                permission.as_str().is_none_or(|permission| {
99                    !context.roles.contains(permission)
100                        && !context.entitlements.contains(permission)
101                })
102            }) {
103                return Err("workflow profile exceeds current subject permissions".into());
104            }
105        }
106        Ok(())
107    }
108}
109/// Current durable grant and its CAS version. Hosts map their rights into this record.
110#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
111pub struct WorkflowAdmissionRecord {
112    /// Authenticated tenant-local grantee; no tenant override in the payload.
113    pub subject_id: SubjectId,
114    /// Monotonic policy version, starting at one.
115    pub version: u64,
116    /// Current deployment authority.
117    pub policy: WorkflowAdmissionPolicy,
118}
119
120#[cfg(test)]
121mod tests {
122    use super::*;
123    #[test]
124    fn deployment_grants_bound_config_and_never_add_subject_permissions() {
125        let policy = WorkflowAdmissionPolicy {
126            enabled: true,
127            max_active_instances: 1,
128            sources: vec![WorkflowSourceGrant {
129                source: "source".parse().unwrap(),
130                event_type: "event".into(),
131                predicate: serde_json::json!({"owner":"self"}),
132            }],
133            resources: ["resource".parse().unwrap()].into(),
134            providers: ["provider".into()].into(),
135            modes: vec![crate::ExecutionMode::Paper],
136        };
137        let context = af_context::RequestContext {
138            tenant_id: "tenant".parse().unwrap(),
139            subject_id: "owner".parse().unwrap(),
140            request_id: "request".parse().unwrap(),
141            locale: "en".into(),
142            roles: ["allowed".into()].into(),
143            entitlements: Default::default(),
144        };
145        let profile:crate::ExecutionProfileRevision=serde_json::from_value(serde_json::json!({"id":"profile","revision":1,"content_digest":"digest","mode":"paper","durability_grade":"standard","trigger_provider":"provider","data_provider":"provider","action_provider":"provider","clock_model":"database","connection_bindings":{"primary":"resource"},"policy_bundle":{"permissions":["allowed"]}})).unwrap();
146        assert!(policy.validate().is_ok());
147        assert!(policy.authorize_profile(&context, &profile).is_ok());
148        let mut changed = policy.clone();
149        changed.enabled = false;
150        assert!(changed.authorize_profile(&context, &profile).is_err());
151        for field in [
152            "mode",
153            "action_provider",
154            "connection_bindings",
155            "policy_bundle",
156        ] {
157            let mut changed = serde_json::to_value(&profile).unwrap();
158            changed[field] = match field {
159                "mode" => "live".into(),
160                "action_provider" => "other".into(),
161                "connection_bindings" => serde_json::json!({"primary":"other"}),
162                _ => serde_json::json!({"permissions":["admin"]}),
163            };
164            assert!(policy
165                .authorize_profile(&context, &serde_json::from_value(changed).unwrap())
166                .is_err());
167        }
168        for permissions in [serde_json::json!("allowed"), serde_json::json!([null])] {
169            let mut changed = profile.clone();
170            changed.policy_bundle = serde_json::json!({"permissions":permissions});
171            assert!(policy.authorize_profile(&context, &changed).is_err());
172        }
173        let mut changed = profile;
174        changed.policy_bundle = Value::Null;
175        assert!(policy.authorize_profile(&context, &changed).is_ok());
176        for variant in 0..8 {
177            let mut changed = policy.clone();
178            match variant {
179                0 => changed.sources = vec![policy.sources[0].clone(); 65],
180                1 => changed.resources = (0..257).map(|i| i.to_string().parse().unwrap()).collect(),
181                2 => changed.providers = (0..257).map(|i| i.to_string()).collect(),
182                3 => changed.modes = vec![crate::ExecutionMode::Paper; 5],
183                4 => changed.sources[0].event_type = " ".into(),
184                5 => changed.sources[0].predicate = Value::Null,
185                6 => {
186                    changed.providers.insert(" ".into());
187                }
188                _ => changed.sources[0].predicate = serde_json::json!({"large":"x".repeat(65536)}),
189            }
190            assert!(changed.validate().is_err(), "case {variant}");
191        }
192    }
193}
af_workflow/admission.rs

af_workflow/
admission.rs
