af-workflow 0.7.1

//! Narrow workflow deployment grants, supplied by the host's entitlement mapping.
use af_context::{SubjectId, WorkflowResourceId, WorkflowSourceId};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::BTreeSet;

/// Exact source/event authority; a binding must retain this payload restriction.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkflowSourceGrant {
    /// Host's stable source identity.
    pub source: WorkflowSourceId,
    /// Exact event type, without wildcard expansion.
    pub event_type: String,
    /// JSON containment restriction, using the same semantics as trigger bindings.
    pub predicate: Value,
}
/// Host-managed authority for ordinary workflow authors; absent grants deny access.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkflowAdmissionPolicy {
    /// Revoke new publication, activation, binding, resume and dispatch when false.
    pub enabled: bool,
    /// Maximum active/draining instances before a new authoring activation is accepted.
    /// Existing-instance retries do not consume a second slot.
    pub max_active_instances: u32,
    /// Exact source/event grants; at most 64.
    pub sources: Vec<WorkflowSourceGrant>,
    /// Stable connection and external resource identities the subject may use.
    pub resources: BTreeSet<WorkflowResourceId>,
    /// Allowed trigger/data/action provider names, without wildcards.
    pub providers: BTreeSet<String>,
    /// Allowed execution modes.
    pub modes: Vec<crate::ExecutionMode>,
}
impl WorkflowAdmissionPolicy {
    /// Bound the grant and reject malformed restrictions.
    pub fn validate(&self) -> Result<(), String> {
        // ponytail: one owned policy row serializes deployment admission; split only if measured contention warrants it.
        if self.sources.len() > 64
            || self.resources.len() > 256
            || self.providers.len() > 256
            || self.modes.len() > 4
        {
            return Err("workflow admission exceeds grant collection bounds".into());
        }
        if self
            .sources
            .iter()
            .any(|grant| grant.event_type.trim().is_empty() || !grant.predicate.is_object())
            || self
                .providers
                .iter()
                .any(|provider| provider.trim().is_empty())
        {
            return Err(
                "workflow admission requires exact providers and source/object restrictions".into(),
            );
        }
        if serde_json::to_vec(self)
            .map_err(|error| error.to_string())?
            .len()
            > 65_536
        {
            return Err("workflow admission exceeds 64 KiB".into());
        }
        Ok(())
    }
    /// Authorize the profile against current host grants and the caller's permissions.
    pub fn authorize_profile(
        &self,
        context: &af_context::RequestContext,
        profile: &crate::ExecutionProfileRevision,
    ) -> Result<(), String> {
        if !self.enabled
            || !self.modes.contains(&profile.mode)
            || [
                &profile.trigger_provider,
                &profile.data_provider,
                &profile.action_provider,
            ]
            .iter()
            .any(|provider| !self.providers.contains(*provider))
            || profile.connection_bindings.values().any(|resource| {
                !self
                    .resources
                    .iter()
                    .any(|allowed| allowed.as_str() == resource)
            })
        {
            return Err("workflow profile exceeds current mode/provider/resource admission".into());
        }
        if let Some(permissions) = profile.policy_bundle.get("permissions") {
            let permissions = permissions
                .as_array()
                .ok_or("workflow profile permissions must be an array")?;
            if permissions.iter().any(|permission| {
                permission.as_str().is_none_or(|permission| {
                    !context.roles.contains(permission)
                        && !context.entitlements.contains(permission)
                })
            }) {
                return Err("workflow profile exceeds current subject permissions".into());
            }
        }
        Ok(())
    }
}
/// Current durable grant and its CAS version. Hosts map their rights into this record.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkflowAdmissionRecord {
    /// Authenticated tenant-local grantee; no tenant override in the payload.
    pub subject_id: SubjectId,
    /// Monotonic policy version, starting at one.
    pub version: u64,
    /// Current deployment authority.
    pub policy: WorkflowAdmissionPolicy,
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn deployment_grants_bound_config_and_never_add_subject_permissions() {
        let policy = WorkflowAdmissionPolicy {
            enabled: true,
            max_active_instances: 1,
            sources: vec![WorkflowSourceGrant {
                source: "source".parse().unwrap(),
                event_type: "event".into(),
                predicate: serde_json::json!({"owner":"self"}),
            }],
            resources: ["resource".parse().unwrap()].into(),
            providers: ["provider".into()].into(),
            modes: vec![crate::ExecutionMode::Paper],
        };
        let context = af_context::RequestContext {
            tenant_id: "tenant".parse().unwrap(),
            subject_id: "owner".parse().unwrap(),
            request_id: "request".parse().unwrap(),
            locale: "en".into(),
            roles: ["allowed".into()].into(),
            entitlements: Default::default(),
        };
        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();
        assert!(policy.validate().is_ok());
        assert!(policy.authorize_profile(&context, &profile).is_ok());
        let mut changed = policy.clone();
        changed.enabled = false;
        assert!(changed.authorize_profile(&context, &profile).is_err());
        for field in [
            "mode",
            "action_provider",
            "connection_bindings",
            "policy_bundle",
        ] {
            let mut changed = serde_json::to_value(&profile).unwrap();
            changed[field] = match field {
                "mode" => "live".into(),
                "action_provider" => "other".into(),
                "connection_bindings" => serde_json::json!({"primary":"other"}),
                _ => serde_json::json!({"permissions":["admin"]}),
            };
            assert!(policy
                .authorize_profile(&context, &serde_json::from_value(changed).unwrap())
                .is_err());
        }
        for permissions in [serde_json::json!("allowed"), serde_json::json!([null])] {
            let mut changed = profile.clone();
            changed.policy_bundle = serde_json::json!({"permissions":permissions});
            assert!(policy.authorize_profile(&context, &changed).is_err());
        }
        let mut changed = profile;
        changed.policy_bundle = Value::Null;
        assert!(policy.authorize_profile(&context, &changed).is_ok());
        for variant in 0..8 {
            let mut changed = policy.clone();
            match variant {
                0 => changed.sources = vec![policy.sources[0].clone(); 65],
                1 => changed.resources = (0..257).map(|i| i.to_string().parse().unwrap()).collect(),
                2 => changed.providers = (0..257).map(|i| i.to_string()).collect(),
                3 => changed.modes = vec![crate::ExecutionMode::Paper; 5],
                4 => changed.sources[0].event_type = " ".into(),
                5 => changed.sources[0].predicate = Value::Null,
                6 => {
                    changed.providers.insert(" ".into());
                }
                _ => changed.sources[0].predicate = serde_json::json!({"large":"x".repeat(65536)}),
            }
            assert!(changed.validate().is_err(), "case {variant}");
        }
    }
}