what_core/policy/

mod.rs

//! Declarative collection authorization policies.
//!
//! Policies are declared per collection in `what.toml` `[collections.<name>]`
//! sections and govern who may create/read/update/delete records, plus record
//! ownership, tenant scoping, and field-level restrictions. Collections without
//! an explicit policy get an owner-protected default (create = "all",
//! update/delete = "owner", read = "all").
//!
//! Enforcement lives in the server handlers: [`CollectionPolicy::allows_create`]
//! / [`allows_mutation`](CollectionPolicy::allows_mutation) gate writes,
//! [`read_scope`](CollectionPolicy::read_scope) forces a WHERE clause on reads,
//! and [`stamp_owner`](CollectionPolicy::stamp_owner) / [`sanitize_input`] keep
//! server-managed fields trustworthy.

use crate::auth::UserContext;
use crate::config::CollectionPolicyConfig;
use crate::sessions::Session;
use crate::{Error, Result};
use serde_json::{Map, Value};
use std::collections::HashMap;

/// A single term in a rule (rules are an OR of terms).
#[derive(Debug, Clone, PartialEq)]
pub enum RuleTerm {
    /// Anyone, including anonymous visitors.
    All,
    /// Any authenticated user.
    User,
    /// The record's owner (mutations/reads only — invalid for create).
    Owner,
    /// No one.
    Nobody,
    /// A named role.
    Role(String),
}

impl std::fmt::Display for RuleTerm {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RuleTerm::All => write!(f, "all"),
            RuleTerm::User => write!(f, "user"),
            RuleTerm::Owner => write!(f, "owner"),
            RuleTerm::Nobody => write!(f, "none"),
            RuleTerm::Role(r) => write!(f, "{r}"),
        }
    }
}

/// An access rule — a disjunction (OR) of terms.
#[derive(Debug, Clone, PartialEq)]
pub struct Rule {
    terms: Vec<RuleTerm>,
}

impl Rule {
    /// Parse a rule expression like `"owner, admin"`. `ctx` is a dotted path
    /// (e.g. `collections.notes.update`) used in error messages. `allow_owner`
    /// is false for create rules (a record has no owner before it exists).
    pub fn parse(s: &str, ctx: &str, allow_owner: bool) -> Result<Rule> {
        let raw: Vec<String> = s
            .split(',')
            .map(|t| t.trim().to_string())
            .filter(|t| !t.is_empty())
            .collect();
        if raw.is_empty() {
            return Err(Error::Config(format!("{ctx}: empty rule")));
        }

        let mut terms = Vec::new();
        for word in &raw {
            let term = match word.as_str() {
                "all" => RuleTerm::All,
                "user" => RuleTerm::User,
                "none" => RuleTerm::Nobody,
                "owner" => {
                    if !allow_owner {
                        return Err(Error::Config(format!(
                            "{ctx}: \"owner\" is not valid for a create rule (a record has no owner until it is created)"
                        )));
                    }
                    RuleTerm::Owner
                }
                other => RuleTerm::Role(other.to_string()),
            };
            terms.push(term);
        }

        // Reserved words cannot be combined with anything else.
        if terms.contains(&RuleTerm::All) && terms.len() > 1 {
            return Err(Error::Config(format!(
                "{ctx}: \"all\" cannot be combined with other terms"
            )));
        }
        if terms.contains(&RuleTerm::Nobody) && terms.len() > 1 {
            return Err(Error::Config(format!(
                "{ctx}: \"none\" cannot be combined with other terms"
            )));
        }

        Ok(Rule { terms })
    }

    /// Rule that allows everyone.
    pub fn all() -> Rule {
        Rule { terms: vec![RuleTerm::All] }
    }

    /// Rule that allows only the record owner.
    pub fn owner() -> Rule {
        Rule { terms: vec![RuleTerm::Owner] }
    }

    /// True if this rule allows anyone unconditionally.
    pub fn is_all(&self) -> bool {
        self.terms == [RuleTerm::All]
    }

    /// True if any term references ownership.
    fn has_owner(&self) -> bool {
        self.terms.contains(&RuleTerm::Owner)
    }

    /// True if any non-owner term matches this actor (i.e. the actor may act on
    /// records regardless of ownership).
    fn non_owner_match(&self, actor: &Actor) -> bool {
        self.terms.iter().any(|t| match t {
            RuleTerm::All => true,
            RuleTerm::User => actor.authenticated,
            RuleTerm::Role(r) => actor.roles.iter().any(|ar| ar == r),
            RuleTerm::Owner | RuleTerm::Nobody => false,
        })
    }
}

impl std::fmt::Display for Rule {
    /// Render the rule as its config vocabulary (round-trips `Rule::parse`).
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let terms: Vec<String> = self.terms.iter().map(|t| t.to_string()).collect();
        write!(f, "{}", terms.join(", "))
    }
}

/// Ownership mode for a collection.
#[derive(Debug, Clone, PartialEq)]
pub enum OwnerMode {
    /// Stamp `_owner` on create.
    Auto,
    /// No ownership tracking.
    None,
}

impl std::fmt::Display for OwnerMode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            OwnerMode::Auto => write!(f, "auto"),
            OwnerMode::None => write!(f, "none"),
        }
    }
}

/// Result of a mutation authorization check.
#[derive(Debug, Clone, PartialEq)]
pub enum Decision {
    /// The mutation is allowed.
    Allow,
    /// Allowed only because the record predates ownership (implicit policy,
    /// unowned record) — callers should warn in dev mode.
    AllowLegacy,
    /// The mutation is denied.
    Deny,
}

/// The read scope to apply to a fetch for a collection.
#[derive(Debug, Clone, PartialEq)]
pub enum ReadScope {
    /// No restriction.
    All,
    /// No matching identity — return nothing.
    Deny,
    /// Filter expressions to AND into the query (each is the filter
    /// mini-language: comma = OR, `&` = AND).
    Filters(Vec<String>),
}

/// Which write a mutation performs.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum MutationKind {
    Update,
    Delete,
}

/// A compiled collection policy.
#[derive(Debug, Clone)]
pub struct CollectionPolicy {
    pub owner_mode: OwnerMode,
    pub create: Rule,
    pub update: Rule,
    pub delete: Rule,
    pub read: Rule,
    pub filter: Option<String>,
    pub readonly_fields: Vec<String>,
    pub private_fields: Vec<String>,
    /// Whether this policy was explicitly declared (vs the implicit default).
    pub explicit: bool,
}

impl Default for CollectionPolicy {
    /// The implicit owner-protected default for undeclared collections.
    fn default() -> Self {
        CollectionPolicy {
            owner_mode: OwnerMode::Auto,
            create: Rule::all(),
            update: Rule::owner(),
            delete: Rule::owner(),
            read: Rule::all(),
            filter: None,
            readonly_fields: Vec::new(),
            private_fields: Vec::new(),
            explicit: false,
        }
    }
}

impl CollectionPolicy {
    /// Compile a raw config entry for collection `name`.
    fn compile(name: &str, cfg: &CollectionPolicyConfig) -> Result<Self> {
        let owner_mode = match cfg.owner.as_deref() {
            None | Some("auto") => OwnerMode::Auto,
            Some("none") => OwnerMode::None,
            Some(other) => {
                return Err(Error::Config(format!(
                    "collections.{name}.owner: expected \"auto\" or \"none\", got \"{other}\""
                )));
            }
        };

        let create = match &cfg.create {
            Some(s) => Rule::parse(s, &format!("collections.{name}.create"), false)?,
            None => Rule::all(),
        };
        let update = match &cfg.update {
            Some(s) => Rule::parse(s, &format!("collections.{name}.update"), true)?,
            None => Rule::owner(),
        };
        let delete = match &cfg.delete {
            Some(s) => Rule::parse(s, &format!("collections.{name}.delete"), true)?,
            None => Rule::owner(),
        };
        let read = match &cfg.read {
            Some(s) => Rule::parse(s, &format!("collections.{name}.read"), true)?,
            None => Rule::all(),
        };

        Ok(CollectionPolicy {
            owner_mode,
            create,
            update,
            delete,
            read,
            filter: cfg.filter.clone().filter(|f| !f.trim().is_empty()),
            readonly_fields: cfg.fields.readonly.clone(),
            private_fields: cfg.fields.private.clone(),
            explicit: true,
        })
    }

    fn mutation_rule(&self, kind: MutationKind) -> &Rule {
        match kind {
            MutationKind::Update => &self.update,
            MutationKind::Delete => &self.delete,
        }
    }

    /// Whether reads of this collection are restricted in any way.
    pub fn is_read_scoped(&self) -> bool {
        !self.read.is_all() || self.filter.is_some()
    }

    /// Authorize a create.
    pub fn allows_create(&self, actor: &Actor) -> bool {
        self.create.non_owner_match(actor)
    }

    /// Authorize an update/delete against the existing record.
    ///
    /// `resolved_filter` is the tenant filter with `#var#` already resolved; if
    /// present, the record must also satisfy it (otherwise a tenant could mutate
    /// another tenant's row by guessing its id).
    pub fn allows_mutation(
        &self,
        actor: &Actor,
        record: Option<&Value>,
        kind: MutationKind,
        resolved_filter: Option<&str>,
    ) -> Decision {
        // A missing record makes the mutation a safe no-op.
        let Some(record) = record else {
            return Decision::Allow;
        };

        // Tenant scope: the record must be inside the actor's tenant.
        if let Some(filter) = resolved_filter {
            if !record_matches_filter(record, filter) {
                return Decision::Deny;
            }
        }

        let rule = self.mutation_rule(kind);

        // Non-owner terms (all/user/role) grant access regardless of ownership.
        if rule.non_owner_match(actor) {
            return Decision::Allow;
        }

        if rule.has_owner() {
            let record_owner = record.get("_owner").and_then(|v| v.as_str());
            match record_owner {
                Some(owner) if actor.owner_keys().iter().any(|k| k == owner) => {
                    return Decision::Allow;
                }
                None => {
                    // Unowned record: legacy-modifiable only under the implicit
                    // default; explicit policies stay strict.
                    if !self.explicit {
                        return Decision::AllowLegacy;
                    }
                }
                Some(_) => {}
            }
        }

        Decision::Deny
    }

    /// Compute the read scope for `actor`. `filter_ctx` supplies `#var#` values
    /// (user/session) for the tenant filter.
    pub fn read_scope(&self, actor: &Actor, filter_ctx: &HashMap<String, Value>) -> ReadScope {
        let mut filters: Vec<String> = Vec::new();

        // Tenant filter (fail closed if any #var# is unresolved).
        if let Some(raw) = &self.filter {
            let resolved = crate::parser::replace_variables(raw, filter_ctx);
            if resolved.contains('#') {
                return ReadScope::Deny;
            }
            filters.push(resolved);
        }

        if self.read.is_all() {
            return if filters.is_empty() {
                ReadScope::All
            } else {
                ReadScope::Filters(filters)
            };
        }

        // A non-owner term (user/role) grants the full row-set (subject to the
        // tenant filter above).
        if self.read.non_owner_match(actor) {
            return if filters.is_empty() {
                ReadScope::All
            } else {
                ReadScope::Filters(filters)
            };
        }

        // Owner scope: restrict to the actor's own records.
        if self.read.has_owner() {
            let keys = actor.owner_keys();
            if keys.is_empty() {
                return ReadScope::Deny;
            }
            // Comma = OR in the filter mini-language.
            let owner_or = keys
                .iter()
                .map(|k| format!("_owner={k}"))
                .collect::<Vec<_>>()
                .join(",");
            filters.push(owner_or);
            return ReadScope::Filters(filters);
        }

        // No term matched (e.g. read = "none", or role rule with no match).
        ReadScope::Deny
    }

    /// The resolved tenant filter for mutation checks, or None.
    pub fn resolved_filter(&self, filter_ctx: &HashMap<String, Value>) -> Option<String> {
        let raw = self.filter.as_ref()?;
        let resolved = crate::parser::replace_variables(raw, filter_ctx);
        if resolved.contains('#') {
            // Unresolved — treat as "matches nothing" by returning a filter no
            // record satisfies. Callers use this only for mutation gating.
            return Some("_owner=\u{0}__unresolved__".to_string());
        }
        Some(resolved)
    }

    /// Stamp `_owner` into a new record's field map (create only).
    pub fn stamp_owner(&self, map: &mut Map<String, Value>, actor: &Actor) {
        if self.owner_mode == OwnerMode::None {
            return;
        }
        if let Some(key) = actor.primary_owner_key() {
            map.insert("_owner".to_string(), Value::String(key));
        }
    }

    /// Stamp the tenant filter field(s) into a new record on create, forcing them
    /// to the actor's resolved values. The `filter` (e.g. `org=#user.org#`) gates
    /// reads/updates/deletes but NOT create, and the tenant field arrives from
    /// client form data — so without this a tenant-A user could POST `org=B` and
    /// inject a record into tenant B. Only the simple `field=value` form is
    /// stamped; an unresolved or multi-term (`,` OR) filter is left alone (the
    /// record then falls outside every tenant read scope — fail-safe).
    pub fn stamp_tenant(&self, map: &mut Map<String, Value>, filter_ctx: &HashMap<String, Value>) {
        let Some(resolved) = self.resolved_filter(filter_ctx) else {
            return;
        };
        if resolved.contains('#') || resolved.contains(',') || resolved.contains('\u{0}') {
            return;
        }
        if let Some((field, value)) = resolved.split_once('=') {
            let field = field.trim();
            if !field.is_empty() {
                map.insert(field.to_string(), Value::String(value.trim().to_string()));
            }
        }
    }

    /// Remove client-supplied values the server manages: the `_owner` field
    /// (never trust it from input) and any `fields.readonly`.
    pub fn sanitize_input(&self, map: &mut Map<String, Value>) {
        map.remove("_owner");
        for field in &self.readonly_fields {
            map.remove(field);
        }
    }
}

/// Registry of compiled collection policies, keyed by collection name.
#[derive(Debug, Clone)]
pub struct PolicyRegistry {
    map: HashMap<String, CollectionPolicy>,
    default_policy: CollectionPolicy,
}

impl PolicyRegistry {
    /// Compile all policies from config, failing loud on any semantic error.
    pub fn from_config(cfg: &HashMap<String, CollectionPolicyConfig>) -> Result<Self> {
        let mut map = HashMap::new();
        for (name, entry) in cfg {
            map.insert(name.clone(), CollectionPolicy::compile(name, entry)?);
        }
        Ok(PolicyRegistry {
            map,
            default_policy: CollectionPolicy::default(),
        })
    }

    /// An empty registry (all collections get the implicit default).
    pub fn empty() -> Self {
        PolicyRegistry {
            map: HashMap::new(),
            default_policy: CollectionPolicy::default(),
        }
    }

    /// Look up a collection's policy, falling back to the implicit default.
    pub fn get(&self, collection: &str) -> &CollectionPolicy {
        self.map.get(collection).unwrap_or(&self.default_policy)
    }

    /// Iterate over explicitly configured collections.
    pub fn configured(&self) -> impl Iterator<Item = (&str, &CollectionPolicy)> {
        self.map.iter().map(|(k, v)| (k.as_str(), v))
    }

    /// Whether reads of this collection are scoped (used to scrub base context).
    pub fn is_read_scoped(&self, collection: &str) -> bool {
        self.get(collection).is_read_scoped()
    }
}

/// The acting principal for a request.
#[derive(Debug, Clone)]
pub struct Actor {
    pub authenticated: bool,
    pub user_sub: Option<String>,
    pub roles: Vec<String>,
    /// `session:<hash>` owner key for anonymous ownership.
    pub session_key: Option<String>,
}

impl Actor {
    /// An anonymous actor (no session, no auth).
    pub fn anonymous() -> Actor {
        Actor {
            authenticated: false,
            user_sub: None,
            roles: Vec::new(),
            session_key: None,
        }
    }

    /// Build an actor from the request's user context and session.
    pub fn from_parts(user: &UserContext, session: Option<&Session>) -> Actor {
        let user_sub = if user.authenticated {
            match user.sub() {
                // Filter mini-language metacharacters would corrupt owner
                // filters — skip the user key and let session ownership stand.
                Some(s) if s.contains([',', '&', '=', '<', '>']) => {
                    tracing::warn!(
                        target: "what::policy",
                        "user sub contains filter metacharacters; skipping user ownership key"
                    );
                    None
                }
                other => other,
            }
        } else {
            None
        };
        let session_key = session.map(|s| format!("session:{}", &s.id[..s.id.len().min(32)]));
        Actor {
            authenticated: user.authenticated,
            user_sub,
            roles: if user.authenticated { user.roles() } else { Vec::new() },
            session_key,
        }
    }

    /// All owner keys this actor can claim (`user:<sub>` and/or `session:<hash>`).
    pub fn owner_keys(&self) -> Vec<String> {
        let mut keys = Vec::new();
        if let Some(sub) = &self.user_sub {
            keys.push(format!("user:{sub}"));
        }
        if let Some(sk) = &self.session_key {
            keys.push(sk.clone());
        }
        keys
    }

    /// The key used to stamp new records: the user key if authenticated, else
    /// the session key.
    pub fn primary_owner_key(&self) -> Option<String> {
        if let Some(sub) = &self.user_sub {
            return Some(format!("user:{sub}"));
        }
        self.session_key.clone()
    }
}

/// Strip `fields.private` from a JSON array of records (or a single record).
pub fn strip_private_fields(items: &mut Value, private: &[String]) {
    if private.is_empty() {
        return;
    }
    match items {
        Value::Array(arr) => {
            for item in arr.iter_mut() {
                if let Value::Object(map) = item {
                    for f in private {
                        map.remove(f);
                    }
                }
            }
        }
        Value::Object(map) => {
            for f in private {
                map.remove(f);
            }
        }
        _ => {}
    }
}

/// Scrub the base template context (built from the entire store): drop any
/// read-scoped collection entirely (fetch directives re-add it scoped) and
/// strip private fields from the rest. Collections with the implicit default
/// (read = "all", no private fields) are untouched.
pub fn scrub_base_context(reg: &PolicyRegistry, ctx: &mut HashMap<String, Value>) {
    for (name, policy) in reg.configured() {
        if policy.is_read_scoped() {
            ctx.remove(name);
        } else if !policy.private_fields.is_empty() {
            if let Some(v) = ctx.get_mut(name) {
                strip_private_fields(v, &policy.private_fields);
            }
        }
    }
}

/// Evaluate a record against a filter expression, mirroring the SQL builder's
/// semantics exactly (comma = OR groups, `&` = AND within a group, ops
/// `>= <= > < =`). Used to enforce tenant scoping on mutations by id.
pub fn record_matches_filter(record: &Value, filter_expr: &str) -> bool {
    let obj = match record {
        Value::Object(m) => m,
        _ => return false,
    };

    // OR across comma-separated groups.
    for group in filter_expr.split(',') {
        if group.trim().is_empty() {
            continue;
        }
        let mut group_ok = true;
        for cond in group.split('&') {
            let cond = cond.trim();
            if cond.is_empty() {
                continue;
            }
            if !eval_condition(obj, cond) {
                group_ok = false;
                break;
            }
        }
        if group_ok {
            return true;
        }
    }
    false
}

fn eval_condition(obj: &Map<String, Value>, cond: &str) -> bool {
    // Order matters: check two-char ops before single-char.
    for (op, is_ge, is_le, is_gt, is_lt, is_eq) in [
        (">=", true, false, false, false, false),
        ("<=", false, true, false, false, false),
        (">", false, false, true, false, false),
        ("<", false, false, false, true, false),
        ("=", false, false, false, false, true),
    ] {
        if let Some((field, val)) = cond.split_once(op) {
            let field = field.trim();
            let val = val.trim();
            let actual = obj.get(field);
            let actual_str = actual.map(value_to_string).unwrap_or_default();

            // Try numeric comparison first, fall back to string.
            let (an, vn) = (actual_str.parse::<f64>(), val.parse::<f64>());
            return if let (Ok(a), Ok(v)) = (an, vn) {
                if is_ge {
                    a >= v
                } else if is_le {
                    a <= v
                } else if is_gt {
                    a > v
                } else if is_lt {
                    a < v
                } else {
                    a == v
                }
            } else if is_eq {
                actual_str == val
            } else if is_ge {
                actual_str >= val.to_string()
            } else if is_le {
                actual_str <= val.to_string()
            } else if is_gt {
                actual_str > val.to_string()
            } else {
                actual_str < val.to_string()
            };
        }
    }
    false
}

fn value_to_string(v: &Value) -> String {
    match v {
        Value::String(s) => s.clone(),
        Value::Number(n) => n.to_string(),
        Value::Bool(b) => b.to_string(),
        Value::Null => String::new(),
        other => other.to_string(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn cfg(f: impl FnOnce(&mut CollectionPolicyConfig)) -> CollectionPolicyConfig {
        let mut c = CollectionPolicyConfig::default();
        f(&mut c);
        c
    }

    fn user_actor(sub: &str, roles: &[&str]) -> Actor {
        Actor {
            authenticated: true,
            user_sub: Some(sub.to_string()),
            roles: roles.iter().map(|s| s.to_string()).collect(),
            session_key: None,
        }
    }

    fn session_actor(id: &str) -> Actor {
        Actor {
            authenticated: false,
            user_sub: None,
            roles: Vec::new(),
            session_key: Some(format!("session:{id}")),
        }
    }

    #[test]
    fn rule_parse_basic() {
        assert!(Rule::parse("all", "x", true).unwrap().is_all());
        assert_eq!(
            Rule::parse("owner, admin", "x", true).unwrap().terms,
            vec![RuleTerm::Owner, RuleTerm::Role("admin".into())]
        );
        assert_eq!(
            Rule::parse("editor,admin", "x", false).unwrap().terms,
            vec![RuleTerm::Role("editor".into()), RuleTerm::Role("admin".into())]
        );
    }

    #[test]
    fn rule_parse_failures() {
        assert!(Rule::parse("owner", "collections.x.create", false).is_err());
        assert!(Rule::parse("all, admin", "x", true).is_err());
        assert!(Rule::parse("none, user", "x", true).is_err());
        assert!(Rule::parse("", "x", true).is_err());
    }

    #[test]
    fn default_policy_is_owner_protected() {
        let p = CollectionPolicy::default();
        assert!(p.create.is_all());
        assert_eq!(p.update, Rule::owner());
        assert_eq!(p.delete, Rule::owner());
        assert!(p.read.is_all());
        assert!(!p.explicit);
        assert!(!p.is_read_scoped());
    }

    #[test]
    fn create_authorization() {
        let p = CollectionPolicy::compile("notes", &cfg(|c| c.create = Some("user".into()))).unwrap();
        assert!(!p.allows_create(&Actor::anonymous()));
        assert!(p.allows_create(&user_actor("alice", &[])));

        let roles = CollectionPolicy::compile("a", &cfg(|c| c.create = Some("editor".into()))).unwrap();
        assert!(!roles.allows_create(&user_actor("bob", &[])));
        assert!(roles.allows_create(&user_actor("bob", &["editor"])));
    }

    #[test]
    fn update_owner_matching() {
        let p = CollectionPolicy::default(); // update = owner, implicit
        let rec = json!({"_owner": "user:alice", "title": "x"});
        assert_eq!(
            p.allows_mutation(&user_actor("alice", &[]), Some(&rec), MutationKind::Update, None),
            Decision::Allow
        );
        assert_eq!(
            p.allows_mutation(&user_actor("bob", &[]), Some(&rec), MutationKind::Update, None),
            Decision::Deny
        );
    }

    #[test]
    fn legacy_unowned_record() {
        let implicit = CollectionPolicy::default();
        let rec = json!({"title": "no owner"});
        assert_eq!(
            implicit.allows_mutation(&Actor::anonymous(), Some(&rec), MutationKind::Delete, None),
            Decision::AllowLegacy
        );

        // Explicit policy stays strict.
        let explicit = CollectionPolicy::compile("n", &cfg(|c| c.update = Some("owner".into()))).unwrap();
        assert_eq!(
            explicit.allows_mutation(&session_actor("abc"), Some(&rec), MutationKind::Update, None),
            Decision::Deny
        );
    }

    #[test]
    fn missing_record_is_safe_noop() {
        let p = CollectionPolicy::default();
        assert_eq!(
            p.allows_mutation(&Actor::anonymous(), None, MutationKind::Delete, None),
            Decision::Allow
        );
    }

    #[test]
    fn role_delete_rule() {
        let p = CollectionPolicy::compile("n", &cfg(|c| c.delete = Some("owner, admin".into()))).unwrap();
        let rec = json!({"_owner": "user:alice"});
        // Non-owner admin passes.
        assert_eq!(
            p.allows_mutation(&user_actor("carol", &["admin"]), Some(&rec), MutationKind::Delete, None),
            Decision::Allow
        );
        // Non-owner non-admin denied.
        assert_eq!(
            p.allows_mutation(&user_actor("carol", &[]), Some(&rec), MutationKind::Delete, None),
            Decision::Deny
        );
    }

    #[test]
    fn tenant_filter_gates_mutation() {
        let p = CollectionPolicy::compile("n", &cfg(|c| {
            c.filter = Some("org=acme".into());
            c.update = Some("user".into());
        })).unwrap();
        let mine = json!({"org": "acme"});
        let theirs = json!({"org": "other"});
        assert_eq!(
            p.allows_mutation(&user_actor("a", &[]), Some(&mine), MutationKind::Update, Some("org=acme")),
            Decision::Allow
        );
        assert_eq!(
            p.allows_mutation(&user_actor("a", &[]), Some(&theirs), MutationKind::Update, Some("org=acme")),
            Decision::Deny
        );
    }

    #[test]
    fn read_scope_permutations() {
        let empty = HashMap::new();

        // read = all
        assert_eq!(CollectionPolicy::default().read_scope(&Actor::anonymous(), &empty), ReadScope::All);

        // read = owner, has session
        let owner_read = CollectionPolicy::compile("n", &cfg(|c| c.read = Some("owner".into()))).unwrap();
        match owner_read.read_scope(&session_actor("abc"), &empty) {
            ReadScope::Filters(f) => assert_eq!(f, vec!["_owner=session:abc".to_string()]),
            other => panic!("expected filters, got {other:?}"),
        }

        // read = owner, no identity → deny
        assert_eq!(owner_read.read_scope(&Actor::anonymous(), &empty), ReadScope::Deny);

        // read = user, authenticated → all
        let user_read = CollectionPolicy::compile("n", &cfg(|c| c.read = Some("user".into()))).unwrap();
        assert_eq!(user_read.read_scope(&user_actor("a", &[]), &empty), ReadScope::All);
        assert_eq!(user_read.read_scope(&Actor::anonymous(), &empty), ReadScope::Deny);
    }

    #[test]
    fn read_scope_tenant_filter() {
        let mut ctx = HashMap::new();
        ctx.insert("user".to_string(), json!({"org_id": "acme"}));
        let p = CollectionPolicy::compile("n", &cfg(|c| c.filter = Some("org_id=#user.org_id#".into()))).unwrap();
        match p.read_scope(&user_actor("a", &[]), &ctx) {
            ReadScope::Filters(f) => assert_eq!(f, vec!["org_id=acme".to_string()]),
            other => panic!("expected filters, got {other:?}"),
        }
        // Unresolved var → deny.
        assert_eq!(p.read_scope(&user_actor("a", &[]), &HashMap::new()), ReadScope::Deny);
    }

    #[test]
    fn owner_and_tenant_combine() {
        let mut ctx = HashMap::new();
        ctx.insert("user".to_string(), json!({"org_id": "acme"}));
        let p = CollectionPolicy::compile("n", &cfg(|c| {
            c.read = Some("owner".into());
            c.filter = Some("org_id=#user.org_id#".into());
        })).unwrap();
        let actor = user_actor("alice", &[]);
        match p.read_scope(&actor, &ctx) {
            ReadScope::Filters(f) => {
                assert_eq!(f, vec!["org_id=acme".to_string(), "_owner=user:alice".to_string()]);
            }
            other => panic!("expected filters, got {other:?}"),
        }
    }

    #[test]
    fn stamp_and_sanitize() {
        let p = CollectionPolicy::compile("n", &cfg(|c| c.fields.readonly = vec!["price".into()])).unwrap();
        let mut map = Map::new();
        map.insert("title".into(), json!("hi"));
        map.insert("_owner".into(), json!("user:evil")); // spoof attempt
        map.insert("price".into(), json!("0")); // readonly
        p.sanitize_input(&mut map);
        assert!(!map.contains_key("_owner"));
        assert!(!map.contains_key("price"));

        p.stamp_owner(&mut map, &user_actor("alice", &[]));
        assert_eq!(map.get("_owner"), Some(&json!("user:alice")));
    }

    #[test]
    fn owner_mode_none_skips_stamp() {
        let p = CollectionPolicy::compile("n", &cfg(|c| c.owner = Some("none".into()))).unwrap();
        let mut map = Map::new();
        p.stamp_owner(&mut map, &user_actor("alice", &[]));
        assert!(!map.contains_key("_owner"));
    }

    #[test]
    fn actor_owner_keys() {
        let both = Actor {
            authenticated: true,
            user_sub: Some("alice".into()),
            roles: vec![],
            session_key: Some("session:abc".into()),
        };
        assert_eq!(both.owner_keys(), vec!["user:alice", "session:abc"]);
        assert_eq!(both.primary_owner_key(), Some("user:alice".into()));
        assert_eq!(session_actor("abc").primary_owner_key(), Some("session:abc".into()));
    }

    #[test]
    fn record_matches_filter_parity() {
        let rec = json!({"org": "acme", "count": 5});
        assert!(record_matches_filter(&rec, "org=acme"));
        assert!(!record_matches_filter(&rec, "org=other"));
        assert!(record_matches_filter(&rec, "count>=5"));
        assert!(record_matches_filter(&rec, "count>3"));
        assert!(!record_matches_filter(&rec, "count>5"));
        // OR groups
        assert!(record_matches_filter(&rec, "org=other,org=acme"));
        // AND within group
        assert!(record_matches_filter(&rec, "org=acme&count=5"));
        assert!(!record_matches_filter(&rec, "org=acme&count=9"));
    }

    #[test]
    fn strip_private() {
        let mut items = json!([{"email": "a@x.com", "name": "A"}, {"email": "b@x.com", "name": "B"}]);
        strip_private_fields(&mut items, &["email".to_string()]);
        assert_eq!(items, json!([{"name": "A"}, {"name": "B"}]));
    }

    #[test]
    fn registry_defaults_and_lookup() {
        let mut cfg_map = HashMap::new();
        cfg_map.insert("notes".to_string(), cfg(|c| c.read = Some("owner".into())));
        let reg = PolicyRegistry::from_config(&cfg_map).unwrap();
        assert!(reg.get("notes").explicit);
        assert!(reg.is_read_scoped("notes"));
        // Unknown collection → implicit default.
        assert!(!reg.get("other").explicit);
        assert!(!reg.is_read_scoped("other"));
    }

    #[test]
    fn registry_fails_loud() {
        let mut cfg_map = HashMap::new();
        cfg_map.insert("bad".to_string(), cfg(|c| c.create = Some("owner".into())));
        assert!(PolicyRegistry::from_config(&cfg_map).is_err());
    }

    #[test]
    fn scrub_base_context_drops_scoped() {
        let mut cfg_map = HashMap::new();
        cfg_map.insert("secret".to_string(), cfg(|c| c.read = Some("owner".into())));
        cfg_map.insert("public".to_string(), cfg(|c| c.fields.private = vec!["ssn".into()]));
        let reg = PolicyRegistry::from_config(&cfg_map).unwrap();

        let mut ctx = HashMap::new();
        ctx.insert("secret".to_string(), json!([{"x": 1}]));
        ctx.insert("public".to_string(), json!([{"name": "A", "ssn": "123"}]));
        ctx.insert("open".to_string(), json!([{"y": 2}]));

        scrub_base_context(&reg, &mut ctx);
        assert!(!ctx.contains_key("secret"));
        assert_eq!(ctx.get("public"), Some(&json!([{"name": "A"}])));
        assert_eq!(ctx.get("open"), Some(&json!([{"y": 2}])));
    }
}