Skip to main content

reddb_server/auth/
store.rs

1//! AuthStore -- manages users, sessions, and API keys in memory.
2//!
3//! Password hashing delegates to the existing Argon2id implementation in
4//! `crate::storage::encryption::argon2id`.  Token generation uses the
5//! OS CSPRNG (`crate::crypto::os_random`) plus SHA-256.
6#![deny(clippy::disallowed_methods)]
7
8use std::borrow::Cow;
9use std::collections::{HashMap, HashSet};
10use std::sync::atomic::{AtomicBool, Ordering};
11use std::sync::{Arc, OnceLock, RwLock};
12
13use crate::crypto::os_random;
14use crate::crypto::sha256::sha256;
15use crate::storage::encryption::argon2id::{derive_key, Argon2Params};
16use crate::storage::engine::pager::Pager;
17
18use super::column_policy_gate::{ColumnAccessRequest, ColumnPolicyGate, ColumnPolicyOutcome};
19use super::enforcement_mode::{legacy_rbac_decision, PolicyEnforcementMode};
20use super::managed_policy::{ManagedPolicyDecision, ManagedPolicyGate, PolicyOp};
21use super::policies::{self as iam_policies, EvalContext, Policy, ResourceRef, SimulationOutcome};
22use super::privileges::{
23    check_grant, Action, AuthzContext, AuthzError, Grant, GrantPrincipal, GrantsView,
24    PermissionCache, Resource, UserAttributes,
25};
26use super::registry::ConfigRegistry;
27use super::vault::{KeyPair, Vault, VaultState};
28use super::{now_ms, ApiKey, AuthConfig, AuthError, Role, Session, User, UserId};
29use crate::runtime::control_events::{
30    ControlEvent, ControlEventConfig, ControlEventCtx, ControlEventLedger, EventKind, Outcome,
31    Sensitivity,
32};
33
34// ---------------------------------------------------------------------------
35// PrincipalRef + SimCtx — IAM policy attachments
36// ---------------------------------------------------------------------------
37
38/// Principal targeted by `attach_policy` / `detach_policy`.
39#[derive(Debug, Clone, PartialEq, Eq, Hash)]
40pub enum PrincipalRef {
41    User(UserId),
42    Group(String),
43}
44
45/// Reserved IAM group that every principal belongs to. Used by the
46/// GRANT-to-PUBLIC compatibility layer.
47pub const PUBLIC_IAM_GROUP: &str = "__public__";
48
49/// Optional context overrides for `simulate` — anything not set falls back
50/// to a default value when the kernel evaluates the request.
51#[derive(Debug, Clone, Default)]
52pub struct SimCtx {
53    pub current_tenant: Option<String>,
54    pub peer_ip: Option<std::net::IpAddr>,
55    pub mfa_present: bool,
56    pub now_ms: Option<u128>,
57}
58
59/// Control-event dependencies for audited IAM policy mutations.
60///
61/// The existing `put_policy` / `delete_policy` / attach / detach
62/// methods remain unaudited for bootstrap and synthetic GRANT internals.
63/// Public mutation surfaces that need evidence pass this context to the
64/// `*_with_control_events` siblings.
65pub struct PolicyMutationControl<'a> {
66    pub ctx: &'a ControlEventCtx<'a>,
67    pub ledger: &'a dyn ControlEventLedger,
68    pub config: ControlEventConfig,
69    pub registry: Option<&'a ConfigRegistry>,
70    pub actor: &'a UserId,
71    pub eval_ctx: &'a EvalContext,
72}
73
74/// Control-event dependencies for audited user lifecycle mutations.
75pub struct UserLifecycleControl<'a> {
76    pub ctx: &'a ControlEventCtx<'a>,
77    pub ledger: &'a dyn ControlEventLedger,
78    pub config: ControlEventConfig,
79}
80
81#[derive(Clone)]
82struct AuthStoreControlEvents {
83    ledger: Arc<dyn ControlEventLedger>,
84    config: ControlEventConfig,
85}
86
87// ---------------------------------------------------------------------------
88// BootstrapResult
89// ---------------------------------------------------------------------------
90
91/// Result of a successful bootstrap operation.
92///
93/// The `certificate` is the hex-encoded string the admin must save --
94/// it is the ONLY way to unseal the vault after a restart.
95pub struct BootstrapResult {
96    pub user: User,
97    pub api_key: ApiKey,
98    /// Certificate hex string.  `None` when vault is not configured.
99    pub certificate: Option<String>,
100}
101
102impl std::fmt::Debug for BootstrapResult {
103    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
104        f.debug_struct("BootstrapResult")
105            .field("user", &self.user)
106            .field("api_key", &"<redacted>")
107            .field(
108                "certificate",
109                &self.certificate.as_ref().map(|_| "<redacted>"),
110            )
111            .finish()
112    }
113}
114
115// ---------------------------------------------------------------------------
116// AuthStore
117// ---------------------------------------------------------------------------
118
119/// Central in-process authority for auth state.
120///
121/// All mutations are guarded by `RwLock`s so the store is `Send + Sync`.
122pub struct AuthStore {
123    /// `(tenant_id, username) -> User`. Tenant scoping is built into the
124    /// key so `alice@acme` and `alice@globex` are distinct identities.
125    users: RwLock<HashMap<UserId, User>>,
126    sessions: RwLock<HashMap<String, Session>>,
127    counted_session_tokens: RwLock<HashSet<String>>,
128    /// key-string -> (owner UserId, role)
129    api_key_index: RwLock<HashMap<String, (UserId, Role)>>,
130    /// Once true, bootstrap() is permanently sealed.
131    bootstrapped: AtomicBool,
132    config: AuthConfig,
133    /// Optional encrypted vault for persisting auth state to pager pages.
134    vault: RwLock<Option<Vault>>,
135    /// Reference to the pager for vault page I/O.
136    pager: Option<Arc<Pager>>,
137    /// Existing operator credential; bootstrap must preserve this seal.
138    provided_certificate: Option<String>,
139    /// Certificate-based keypair for token signing and vault seal.
140    /// Populated after bootstrap or after restoring from a sealed vault.
141    keypair: RwLock<Option<KeyPair>>,
142    /// Encrypted key-value store for arbitrary secrets.
143    /// Persisted to vault alongside users/api_keys.
144    vault_kv: RwLock<HashMap<String, String>>,
145    secret_key_init: std::sync::Mutex<()>,
146    /// Plain (non-encrypted) user key-value store for `$kv.*` / `SET KV`
147    /// (#1602). Deliberately isolated from `vault_kv` (encrypted secrets)
148    /// and from the runtime config store — three independent flat-maps,
149    /// no mixing. Backs the `kv:read` / `kv:write` IAM-gated resolver.
150    plain_kv: RwLock<HashMap<String, String>>,
151    /// Per-user GRANT rows. Persisted via `vault_kv` under the
152    /// `red.acl.grants.<tenant>/<user>` key prefix so existing snapshot
153    /// logic keeps working without modification. See `privileges` module
154    /// for the resolution algorithm.
155    grants: RwLock<HashMap<UserId, Vec<Grant>>>,
156    /// PUBLIC grants — apply to every authenticated principal.
157    public_grants: RwLock<Vec<Grant>>,
158    /// PG-style account attributes (`VALID UNTIL`, `CONNECTION LIMIT`,
159    /// `search_path`). Keyed by `UserId`. Persisted under
160    /// `red.acl.attrs.<tenant>/<user>`.
161    user_attributes: RwLock<HashMap<UserId, UserAttributes>>,
162    /// Live session count per user — used by `CONNECTION LIMIT`
163    /// enforcement on login. Bumped at authenticate, decremented at
164    /// session revoke / expiry.
165    session_count_by_user: RwLock<HashMap<UserId, u32>>,
166    /// Pre-resolved (resource, action) cache built per-user so the
167    /// hot path skips a linear scan of the user's grants on every
168    /// statement. Invalidated on GRANT / REVOKE / ALTER USER.
169    permission_cache: RwLock<HashMap<UserId, PermissionCache>>,
170    /// IAM-style policies, keyed by id. Persisted under
171    /// `red.iam.policies`. The kernel in `super::policies` owns the
172    /// Policy type — this map just deduplicates and shares.
173    policies: RwLock<HashMap<String, Arc<Policy>>>,
174    /// Per-user policy attachments — ordered list of policy ids.
175    /// Persisted under `red.iam.attachments.users`.
176    user_attachments: RwLock<HashMap<UserId, Vec<String>>>,
177    /// Per-group policy attachments. Users join groups through
178    /// `UserAttributes::groups`; effective policies resolve group
179    /// attachments before user-direct attachments.
180    group_attachments: RwLock<HashMap<String, Vec<String>>>,
181    /// Cached effective `Vec<Arc<Policy>>` per user. Invalidated on
182    /// any policy mutation that affects the user's attachments.
183    iam_effective_cache: RwLock<HashMap<UserId, Vec<Arc<Policy>>>>,
184    /// Once any IAM policy is installed, authorization switches to the
185    /// IAM evaluator and stays deny-by-default even if policies are
186    /// later deleted. Persisted under `red.iam.enabled`.
187    iam_authorization_enabled: AtomicBool,
188    /// Selects the fallback behaviour when the IAM evaluator returns
189    /// `DefaultDeny`. See [`PolicyEnforcementMode`] and #712 / S5A.
190    /// Default is [`PolicyEnforcementMode::default_existing_install`]
191    /// (`LegacyRbac`) so an existing install upgrading past #712
192    /// keeps its current authorization posture until the operator
193    /// flips it explicitly. The bootstrap path sets this to
194    /// `PolicyOnly` for fresh installs; the boot-time config loader
195    /// supersedes both when a `red.config.policy.enforcement_mode`
196    /// value is present in the configured value store.
197    enforcement_mode: RwLock<PolicyEnforcementMode>,
198    /// Once-per-boot flag used by [`AuthStore::take_legacy_rbac_warn_once`]
199    /// to deliver exactly one `warn`-level log line when the store is
200    /// running under [`PolicyEnforcementMode::LegacyRbac`]. The boot
201    /// path checks the flag and, if it can claim it, emits the line.
202    legacy_rbac_boot_warn_emitted: AtomicBool,
203    /// `(tenant, role) → HashSet<CollectionId>` cache used by the AI
204    /// pipeline (issue #119). Distinct from `permission_cache`, which
205    /// is keyed by `UserId` and answers `(resource, action)` lookups —
206    /// this cache answers the inverse "what collections is this scope
207    /// allowed to read?" query that `AuthorizedSearch` uses to
208    /// pre-filter SEARCH SIMILAR / SEARCH CONTEXT candidates before any
209    /// similarity score is computed. Entries TTL out at 60s and are
210    /// invalidated explicitly on GRANT/REVOKE/CREATE POLICY/DROP
211    /// POLICY/DROP COLLECTION.
212    visible_collections_cache: super::scope_cache::AuthCache,
213    control_events: RwLock<Option<AuthStoreControlEvents>>,
214}
215
216/// RAII reservation for a long-lived wire connection. Dropping the lease
217/// releases exactly one CONNECTION LIMIT slot.
218pub struct ConnectionLease {
219    store: Arc<AuthStore>,
220    principal: UserId,
221}
222
223impl Drop for ConnectionLease {
224    fn drop(&mut self) {
225        self.store.decrement_session_count(&self.principal);
226    }
227}
228
229// Use fast-but-safe Argon2id params for auth hashing (smaller than the
230// default 64 MB so that user-management RPCs respond quickly).
231fn auth_argon2_params() -> Argon2Params {
232    Argon2Params {
233        m_cost: 4 * 1024, // 4 MB
234        t_cost: 3,
235        p: 1,
236        tag_len: 32,
237    }
238}
239
240fn policy_control_fields(
241    policy_id: &str,
242    policy: Option<&Policy>,
243    principal: Option<&PrincipalRef>,
244) -> HashMap<String, Sensitivity> {
245    let mut fields = HashMap::new();
246    fields.insert("policy_id".to_string(), Sensitivity::raw(policy_id));
247    if let Some(policy) = policy {
248        let effects = policy
249            .statements
250            .iter()
251            .map(|s| match s.effect {
252                iam_policies::Effect::Allow => "allow",
253                iam_policies::Effect::Deny => "deny",
254            })
255            .collect::<Vec<_>>()
256            .join(",");
257        let actions = policy
258            .statements
259            .iter()
260            .flat_map(|s| s.actions.iter())
261            .map(policy_action_pattern)
262            .collect::<Vec<_>>()
263            .join(",");
264        let resources = policy
265            .statements
266            .iter()
267            .flat_map(|s| s.resources.iter())
268            .map(policy_resource_pattern)
269            .collect::<Vec<_>>()
270            .join(",");
271        fields.insert("effect".to_string(), Sensitivity::raw(effects));
272        fields.insert("action".to_string(), Sensitivity::raw(actions));
273        fields.insert("resource".to_string(), Sensitivity::raw(resources));
274        if let Some(attrs) = policy_principal_attrs(policy) {
275            fields.insert("principal_attrs".to_string(), Sensitivity::raw(attrs));
276        }
277    }
278    if let Some(principal) = principal {
279        fields.insert(
280            "principal".to_string(),
281            Sensitivity::raw(policy_principal_label(principal)),
282        );
283    }
284    fields
285}
286
287fn policy_action_pattern(pattern: &iam_policies::ActionPattern) -> String {
288    match pattern {
289        iam_policies::ActionPattern::Exact(s) => s.clone(),
290        iam_policies::ActionPattern::Wildcard => "*".to_string(),
291        iam_policies::ActionPattern::Prefix(s) => format!("{s}:*"),
292    }
293}
294
295fn policy_resource_pattern(pattern: &iam_policies::ResourcePattern) -> String {
296    match pattern {
297        iam_policies::ResourcePattern::Exact { kind, name } => format!("{kind}:{name}"),
298        iam_policies::ResourcePattern::Glob(s) => s.clone(),
299        iam_policies::ResourcePattern::Wildcard => "*".to_string(),
300    }
301}
302
303fn policy_principal_attrs(policy: &Policy) -> Option<String> {
304    let mut attrs = Vec::new();
305    for statement in &policy.statements {
306        let Some(condition) = &statement.condition else {
307            continue;
308        };
309        if let Some(value) = condition.platform_scoped {
310            attrs.push(format!("platform_scoped={value}"));
311        }
312        if let Some(value) = condition.mfa {
313            attrs.push(format!("mfa={value}"));
314        }
315        if let Some(value) = condition.tenant_match {
316            attrs.push(format!("tenant_match={value}"));
317        }
318    }
319    if attrs.is_empty() {
320        None
321    } else {
322        Some(attrs.join(","))
323    }
324}
325
326fn policy_principal_label(principal: &PrincipalRef) -> String {
327    match principal {
328        PrincipalRef::User(uid) => format!("user:{uid}"),
329        PrincipalRef::Group(group) => format!("group:{group}"),
330    }
331}
332
333fn default_user_lifecycle_ctx<'a>() -> ControlEventCtx<'a> {
334    ControlEventCtx {
335        actor: crate::runtime::control_events::ActorRef::Anonymous,
336        scope: None,
337        request_id: None,
338        trace_id: None,
339    }
340}
341
342fn bootstrap_user_lifecycle_ctx<'a>() -> ControlEventCtx<'a> {
343    ControlEventCtx {
344        actor: crate::runtime::control_events::ActorRef::System("bootstrap"),
345        scope: None,
346        request_id: None,
347        trace_id: None,
348    }
349}
350
351fn user_resource(id: &UserId) -> String {
352    format!("user:{id}")
353}
354
355fn api_key_resource(api_key_id: &str) -> String {
356    format!("apikey:{api_key_id}")
357}
358
359/// Hex digest that identifies an API key in control events and, with the
360/// `rkh_` prefix, at rest. Accepts either the secret or the stored id, so
361/// every entry point can take whichever the caller holds.
362fn api_key_id(key: &str) -> String {
363    match key.strip_prefix("rkh_") {
364        Some(hex) => hex.to_string(),
365        None => hex::encode(sha256(key.as_bytes())),
366    }
367}
368
369/// Stored form of an API key: `rkh_<sha256(secret)>`. The secret is only
370/// ever compared by hashing it and looking the result up.
371pub(crate) fn api_key_credential_id(token: &str) -> String {
372    format!("rkh_{}", api_key_id(token))
373}
374
375fn password_evidence() -> Sensitivity {
376    Sensitivity::redacted()
377}
378
379fn user_control_fields(
380    id: &UserId,
381    role: Option<Role>,
382    enabled: Option<bool>,
383    include_password: bool,
384) -> HashMap<String, Sensitivity> {
385    let mut fields = HashMap::new();
386    fields.insert(
387        "username".to_string(),
388        Sensitivity::raw(id.username.clone()),
389    );
390    fields.insert(
391        "tenant_id".to_string(),
392        Sensitivity::raw(id.tenant.clone().unwrap_or_default()),
393    );
394    if let Some(role) = role {
395        fields.insert("role".to_string(), Sensitivity::raw(role.as_str()));
396    }
397    if let Some(enabled) = enabled {
398        fields.insert("enabled".to_string(), Sensitivity::raw(enabled.to_string()));
399    }
400    if include_password {
401        fields.insert("password".to_string(), password_evidence());
402    }
403    fields
404}
405
406fn api_key_control_fields(
407    id: &UserId,
408    role: Role,
409    api_key_id: &str,
410) -> HashMap<String, Sensitivity> {
411    let mut fields = user_control_fields(id, Some(role), None, false);
412    fields.insert("api_key_id".to_string(), Sensitivity::raw(api_key_id));
413    fields.insert("api_key".to_string(), Sensitivity::redacted());
414    fields
415}
416
417fn user_error_is_denied(err: &AuthError) -> bool {
418    !matches!(err, AuthError::Internal(_))
419}
420
421impl AuthStore {
422    // -----------------------------------------------------------------
423    // Construction
424    // -----------------------------------------------------------------
425
426    pub fn new(config: AuthConfig) -> Self {
427        Self {
428            users: RwLock::new(HashMap::new()),
429            sessions: RwLock::new(HashMap::new()),
430            counted_session_tokens: RwLock::new(HashSet::new()),
431            api_key_index: RwLock::new(HashMap::new()),
432            bootstrapped: AtomicBool::new(false),
433            config,
434            vault: RwLock::new(None),
435            pager: None,
436            provided_certificate: None,
437            keypair: RwLock::new(None),
438            vault_kv: RwLock::new(HashMap::new()),
439            secret_key_init: std::sync::Mutex::new(()),
440            plain_kv: RwLock::new(HashMap::new()),
441            grants: RwLock::new(HashMap::new()),
442            public_grants: RwLock::new(Vec::new()),
443            user_attributes: RwLock::new(HashMap::new()),
444            session_count_by_user: RwLock::new(HashMap::new()),
445            permission_cache: RwLock::new(HashMap::new()),
446            policies: RwLock::new(HashMap::new()),
447            user_attachments: RwLock::new(HashMap::new()),
448            group_attachments: RwLock::new(HashMap::new()),
449            iam_effective_cache: RwLock::new(HashMap::new()),
450            iam_authorization_enabled: AtomicBool::new(false),
451            enforcement_mode: RwLock::new(PolicyEnforcementMode::default_existing_install()),
452            legacy_rbac_boot_warn_emitted: AtomicBool::new(false),
453            visible_collections_cache: super::scope_cache::AuthCache::new(
454                super::scope_cache::DEFAULT_TTL,
455            ),
456            control_events: RwLock::new(None),
457        }
458    }
459
460    pub fn configure_control_events(
461        &self,
462        ledger: Arc<dyn ControlEventLedger>,
463        config: ControlEventConfig,
464    ) {
465        *self
466            .control_events
467            .write()
468            .unwrap_or_else(|e| e.into_inner()) = Some(AuthStoreControlEvents { ledger, config });
469    }
470
471    fn configured_control_events(&self) -> Option<AuthStoreControlEvents> {
472        self.control_events
473            .read()
474            .unwrap_or_else(|e| e.into_inner())
475            .clone()
476    }
477
478    /// Create an `AuthStore` backed by encrypted vault pages inside the
479    /// main `.rdb` database file.
480    ///
481    /// If vault pages already exist, their contents are loaded and
482    /// restored into the in-memory store.  All subsequent mutations are
483    /// automatically persisted to the vault pages via the pager.
484    pub fn with_vault(config: AuthConfig, pager: Arc<Pager>) -> Result<Self, AuthError> {
485        let certificate = crate::utils::env_with_file_fallback("REDDB_CERTIFICATE");
486        let passphrase = crate::utils::env_with_file_fallback("REDDB_VAULT_PASSPHRASE");
487        match (certificate, passphrase) {
488            (Some(_), Some(_)) => Err(AuthError::Internal(
489                "set either REDDB_CERTIFICATE or REDDB_VAULT_PASSPHRASE, not both".into(),
490            )),
491            (None, Some(passphrase)) => Self::with_vault_passphrase(config, pager, &passphrase),
492            (certificate, None) => {
493                Self::with_vault_optional_certificate(config, pager, certificate.as_deref())
494            }
495        }
496    }
497
498    pub fn with_vault_certificate(
499        config: AuthConfig,
500        pager: Arc<Pager>,
501        certificate_hex: &str,
502    ) -> Result<Self, AuthError> {
503        Self::with_vault_optional_certificate(config, pager, Some(certificate_hex))
504    }
505
506    pub fn with_vault_passphrase(
507        config: AuthConfig,
508        pager: Arc<Pager>,
509        passphrase: &str,
510    ) -> Result<Self, AuthError> {
511        let mut store = Self::new(config);
512        let vault = Vault::with_passphrase(&pager, passphrase)?;
513        if let Some(state) = vault.load(&pager)? {
514            store.restore_from_vault(state);
515        }
516        *store.vault.write().map_err(lock_err)? = Some(vault);
517        store.pager = Some(pager);
518        Ok(store)
519    }
520
521    fn with_vault_optional_certificate(
522        config: AuthConfig,
523        pager: Arc<Pager>,
524        certificate_hex: Option<&str>,
525    ) -> Result<Self, AuthError> {
526        let mut store = Self::new(config);
527
528        // Restore persisted state if vault pages exist. A fresh database does
529        // not need a temporary seal: bootstrap generates the certificate and
530        // installs the first real vault atomically.
531        if Vault::has_saved_state(&pager) {
532            let vault = match certificate_hex {
533                Some(certificate) => Vault::with_certificate(&pager, certificate)?,
534                None => Vault::open(&pager)?,
535            };
536            if let Some(state) = vault.load(&pager)? {
537                store.restore_from_vault(state);
538            }
539            *store.vault.write().unwrap_or_else(|e| e.into_inner()) = Some(vault);
540        } else if let Some(certificate) = certificate_hex {
541            let vault = Vault::with_certificate(&pager, certificate)?;
542            *store.vault.write().unwrap_or_else(|e| e.into_inner()) = Some(vault);
543        }
544
545        store.pager = Some(pager);
546        store.provided_certificate =
547            certificate_hex.map(|certificate| certificate.trim().to_ascii_lowercase());
548        Ok(store)
549    }
550
551    pub fn config(&self) -> &AuthConfig {
552        &self.config
553    }
554
555    pub fn is_enabled(&self) -> bool {
556        self.config.enabled
557    }
558
559    /// Returns true when no users exist yet and bootstrap hasn't been sealed.
560    pub fn needs_bootstrap(&self) -> bool {
561        !self.bootstrapped.load(Ordering::Acquire)
562            && self.users.read().map(|u| u.is_empty()).unwrap_or(true)
563    }
564
565    /// Internal: read-locked lookup by `UserId`.
566    fn get_user_cloned(&self, id: &UserId) -> Option<User> {
567        self.users.read().ok().and_then(|m| m.get(id).cloned())
568    }
569
570    /// Whether bootstrap has already been performed (sealed).
571    pub fn is_bootstrapped(&self) -> bool {
572        self.bootstrapped.load(Ordering::Acquire)
573    }
574
575    /// Bootstrap the first admin user. One-shot, irreversible.
576    ///
577    /// Uses an atomic compare-exchange to guarantee that even under
578    /// concurrent calls, only the first one succeeds. Once sealed,
579    /// all subsequent calls fail immediately -- there is no undo.
580    ///
581    /// When a vault/pager is configured, a certificate-based keypair is
582    /// generated and the vault is re-encrypted with the certificate-derived
583    /// key.  The certificate hex string is returned in `BootstrapResult`
584    /// so the admin can save it.
585    pub fn bootstrap(&self, username: &str, password: &str) -> Result<BootstrapResult, AuthError> {
586        if let Some(configured) = self.configured_control_events() {
587            let ctx = bootstrap_user_lifecycle_ctx();
588            let control = UserLifecycleControl {
589                ctx: &ctx,
590                ledger: configured.ledger.as_ref(),
591                config: configured.config,
592            };
593            self.bootstrap_with_control_events_inner(username, password, &control)
594        } else {
595            self.bootstrap_unaudited(username, password)
596        }
597    }
598
599    /// Bootstrap while durably delivering a newly generated vault
600    /// certificate before the sealed vault is installed or persisted.
601    pub fn bootstrap_with_certificate_delivery(
602        &self,
603        username: &str,
604        password: &str,
605        mut deliver: impl FnMut(&str) -> Result<(), AuthError>,
606    ) -> Result<BootstrapResult, AuthError> {
607        let delivery: &mut dyn FnMut(&str) -> Result<(), AuthError> = &mut deliver;
608        if let Some(configured) = self.configured_control_events() {
609            let ctx = bootstrap_user_lifecycle_ctx();
610            let control = UserLifecycleControl {
611                ctx: &ctx,
612                ledger: configured.ledger.as_ref(),
613                config: configured.config,
614            };
615            self.bootstrap_with_control_events_inner_and_delivery(
616                username,
617                password,
618                &control,
619                Some(delivery),
620            )
621        } else {
622            self.bootstrap_unaudited_with_delivery(username, password, Some(delivery))
623        }
624    }
625
626    pub fn bootstrap_with_control_events(
627        &self,
628        username: &str,
629        password: &str,
630        ctx: &ControlEventCtx<'_>,
631        ledger: &dyn ControlEventLedger,
632        config: ControlEventConfig,
633    ) -> Result<BootstrapResult, AuthError> {
634        let system_ctx = ControlEventCtx {
635            actor: crate::runtime::control_events::ActorRef::System("bootstrap"),
636            scope: ctx.scope.clone(),
637            request_id: ctx.request_id.clone(),
638            trace_id: ctx.trace_id.clone(),
639        };
640        let control = UserLifecycleControl {
641            ctx: &system_ctx,
642            ledger,
643            config,
644        };
645        self.bootstrap_with_control_events_inner(username, password, &control)
646    }
647
648    /// Backwards-compatible bootstrap entry point for older callers
649    /// that used to ask for an operator-owned first admin. Authorization
650    /// is now policy-first, so this creates the same ordinary platform
651    /// admin as [`Self::bootstrap`].
652    pub fn bootstrap_system_admin(
653        &self,
654        username: &str,
655        password: &str,
656    ) -> Result<BootstrapResult, AuthError> {
657        self.bootstrap(username, password)
658    }
659
660    fn bootstrap_unaudited(
661        &self,
662        username: &str,
663        password: &str,
664    ) -> Result<BootstrapResult, AuthError> {
665        self.bootstrap_unaudited_with_delivery(username, password, None)
666    }
667
668    fn bootstrap_unaudited_with_delivery(
669        &self,
670        username: &str,
671        password: &str,
672        mut certificate_delivery: Option<&mut dyn FnMut(&str) -> Result<(), AuthError>>,
673    ) -> Result<BootstrapResult, AuthError> {
674        // Atomic seal: only the first caller wins.
675        if self
676            .bootstrapped
677            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
678            .is_err()
679        {
680            return Err(AuthError::Forbidden(
681                "bootstrap already completed — sealed permanently".to_string(),
682            ));
683        }
684
685        // Double-check users are actually empty (belt and suspenders).
686        {
687            let users = self.users.read().map_err(lock_err)?;
688            if !users.is_empty() {
689                return Err(AuthError::Forbidden(
690                    "bootstrap already completed — users exist".to_string(),
691                ));
692            }
693        }
694
695        let user = self.create_user_in_tenant_unaudited(None, username, password, Role::Admin)?;
696        let key = self.create_api_key_in_tenant_unaudited(
697            None,
698            username,
699            "bootstrap",
700            Role::Admin,
701            None,
702        )?;
703
704        // Token signing gets a fresh master secret. Preserve an existing
705        // certificate/passphrase seal instead of replacing the operator's key.
706        let credential_result = (|| -> Result<Option<String>, AuthError> {
707            let certificate = if let Some(ref pager) = self.pager {
708                let keypair = KeyPair::generate();
709                let mut vault = self.vault.write().map_err(lock_err)?;
710                let certificate = if vault.is_none() {
711                    let certificate = keypair.certificate_hex();
712                    if let Some(deliver) = certificate_delivery.as_mut() {
713                        deliver(&certificate)?;
714                    }
715                    *vault = Some(Vault::with_certificate_bytes(pager, &keypair.certificate)?);
716                    Some(certificate)
717                } else {
718                    self.provided_certificate.clone()
719                };
720                *self.keypair.write().map_err(lock_err)? = Some(keypair);
721                drop(vault);
722                self.ensure_vault_secret_key_result()?;
723                certificate
724            } else {
725                None
726            };
727            *self.enforcement_mode.write().map_err(lock_err)? =
728                PolicyEnforcementMode::default_fresh_bootstrap();
729            self.vault_kv.write().map_err(lock_err)?.insert(
730                "red.config.policy.enforcement_mode".into(),
731                PolicyEnforcementMode::default_fresh_bootstrap()
732                    .as_str()
733                    .into(),
734            );
735            self.persist_to_vault_result()?;
736            Ok(certificate)
737        })();
738        let certificate = match credential_result {
739            Ok(certificate) => certificate,
740            Err(err) => {
741                self.rollback_bootstrap(&UserId::from_parts(None, username));
742                return Err(err);
743            }
744        };
745
746        Ok(BootstrapResult {
747            user,
748            api_key: key,
749            certificate,
750        })
751    }
752
753    fn bootstrap_with_control_events_inner(
754        &self,
755        username: &str,
756        password: &str,
757        control: &UserLifecycleControl<'_>,
758    ) -> Result<BootstrapResult, AuthError> {
759        self.bootstrap_with_control_events_inner_and_delivery(username, password, control, None)
760    }
761
762    fn bootstrap_with_control_events_inner_and_delivery(
763        &self,
764        username: &str,
765        password: &str,
766        control: &UserLifecycleControl<'_>,
767        certificate_delivery: Option<&mut dyn FnMut(&str) -> Result<(), AuthError>>,
768    ) -> Result<BootstrapResult, AuthError> {
769        let id = UserId::from_parts(None, username);
770        match self.bootstrap_unaudited_with_delivery(username, password, certificate_delivery) {
771            Ok(result) => {
772                let event_result = self.emit_user_lifecycle_allowed(
773                    control,
774                    EventKind::UserCreate,
775                    "user.create",
776                    &id,
777                    user_control_fields(&id, Some(Role::Admin), Some(true), true),
778                );
779                if let Err(err) = event_result {
780                    self.rollback_bootstrap(&id);
781                    return Err(err);
782                }
783                Ok(result)
784            }
785            Err(err) => {
786                self.emit_user_lifecycle_outcome(
787                    control,
788                    if user_error_is_denied(&err) {
789                        Outcome::Denied
790                    } else {
791                        Outcome::Error
792                    },
793                    EventKind::UserCreate,
794                    "user.create",
795                    &id,
796                    Some(err.to_string()),
797                    user_control_fields(&id, Some(Role::Admin), Some(true), true),
798                );
799                Err(err)
800            }
801        }
802    }
803
804    /// Auto-bootstrap from environment variables if no users exist.
805    ///
806    /// Checks `REDDB_USERNAME` and `REDDB_PASSWORD`. If both are set and
807    /// the user store is empty, creates the first admin user automatically.
808    /// This mirrors the Docker pattern (`MYSQL_ROOT_PASSWORD`, etc.).
809    ///
810    /// Returns `Some(BootstrapResult)` if bootstrapped, `None` if skipped.
811    pub fn bootstrap_from_env(&self) -> Option<BootstrapResult> {
812        if !self.needs_bootstrap() {
813            return None;
814        }
815
816        let username = crate::utils::env_with_file_fallback("REDDB_USERNAME")?;
817        let password = crate::utils::env_with_file_fallback("REDDB_PASSWORD")?;
818
819        if username.is_empty() || password.is_empty() {
820            return None;
821        }
822        let mut certificate_output = if self.bootstrap_requires_certificate_delivery() {
823            let Some(path) = crate::utils::env_with_file_fallback("REDDB_BOOTSTRAP_CERT_OUT")
824            else {
825                tracing::error!(
826                    target: "reddb::security",
827                    "env bootstrap requires REDDB_BOOTSTRAP_CERT_OUT when the vault is enabled; refusing to emit the certificate to logs"
828                );
829                return None;
830            };
831            let mut options = std::fs::OpenOptions::new();
832            options.write(true).create_new(true);
833            #[cfg(unix)]
834            {
835                use std::os::unix::fs::OpenOptionsExt;
836                options.mode(0o600);
837            }
838            let file = match options.open(&path) {
839                Ok(file) => file,
840                Err(error) => {
841                    tracing::error!(
842                        target: "reddb::security",
843                        path = %path,
844                        error = %error,
845                        "env bootstrap certificate output is not writable"
846                    );
847                    return None;
848                }
849            };
850            Some((path, file))
851        } else {
852            None
853        };
854
855        let mut certificate_staged = false;
856        let bootstrap_result = match certificate_output.as_mut() {
857            Some((path, file)) => {
858                self.bootstrap_with_certificate_delivery(&username, &password, |certificate| {
859                    use std::io::Write;
860                    writeln!(file, "{certificate}").map_err(|error| {
861                        AuthError::Internal(format!(
862                            "write bootstrap certificate {}: {error}",
863                            path
864                        ))
865                    })?;
866                    file.sync_all().map_err(|error| {
867                        AuthError::Internal(format!("sync bootstrap certificate {}: {error}", path))
868                    })?;
869                    certificate_staged = true;
870                    Ok(())
871                })
872            }
873            None => self.bootstrap(&username, &password),
874        };
875
876        match bootstrap_result {
877            Ok(result) => {
878                // F-04: `username` is REDDB_USERNAME — operator-supplied
879                // (env), but still routed through the LogField escaper
880                // because env strings cross trust boundaries in some
881                // deployment models (k8s downward API, Vault sidecar,
882                // external secret operator). See ADR 0010.
883                tracing::info!(
884                    username = %reddb_wire::audit_safe_log_field(&username),
885                    "bootstrapped admin user from REDDB_USERNAME/REDDB_PASSWORD"
886                );
887                if !certificate_staged {
888                    if let (Some(cert), Some((path, file))) =
889                        (result.certificate.as_deref(), certificate_output.as_mut())
890                    {
891                        use std::io::Write;
892                        if let Err(error) = writeln!(file, "{cert}").and_then(|()| file.sync_all())
893                        {
894                            tracing::error!(
895                                target: "reddb::security",
896                                path = %path,
897                                error = %error,
898                                "could not write the vault certificate"
899                            );
900                            return None;
901                        }
902                    }
903                }
904                if let Some((path, _)) = certificate_output {
905                    tracing::warn!(
906                        target: "reddb::security",
907                        path = %path,
908                        "vault certificate written — save it: the ONLY way to unseal after a restart"
909                    );
910                }
911                Some(result)
912            }
913            Err(e) => {
914                let output_path = certificate_output.as_ref().map(|(path, _)| path.clone());
915                drop(certificate_output);
916                if let Some(path) = output_path {
917                    let _ = std::fs::remove_file(path);
918                }
919                tracing::error!(err = %e, "env bootstrap failed");
920                None
921            }
922        }
923    }
924
925    // -----------------------------------------------------------------
926    // Vault persistence
927    // -----------------------------------------------------------------
928
929    /// Persist the current auth state to the vault pages (if configured).
930    pub(crate) fn persist_to_vault_result(&self) -> Result<(), AuthError> {
931        let vault_guard = self.vault.read().unwrap_or_else(|e| e.into_inner());
932        if let (Some(ref vault), Some(ref pager)) = (&*vault_guard, &self.pager) {
933            let state = self.snapshot();
934            vault.save(pager, &state)?;
935        }
936        Ok(())
937    }
938
939    /// Persist the current auth state to the vault pages (if configured).
940    ///
941    /// Legacy auth mutations still treat in-memory state as authoritative.
942    /// New secret-management paths use the `try_` methods below so callers
943    /// get a hard error if the vault write fails.
944    fn persist_to_vault(&self) {
945        if let Err(e) = self.persist_to_vault_result() {
946            tracing::error!(err = %e, "vault persist failed");
947            // Issue #205 — vault persist is the secret-rotation
948            // serialization point: when this fails, freshly rotated
949            // credentials live only in memory and a process restart
950            // loses them. Operator-grade event so the operator can
951            // intervene before the next restart.
952            crate::telemetry::operator_event::OperatorEvent::SecretRotationFailed {
953                secret_ref: "auth_vault".to_string(),
954                error: e.to_string(),
955            }
956            .emit_global();
957        }
958    }
959
960    /// True when this store has an encrypted vault and pager wired in.
961    pub fn is_vault_backed(&self) -> bool {
962        self.pager.is_some()
963            && self
964                .vault
965                .read()
966                .map(|guard| guard.is_some())
967                .unwrap_or(false)
968    }
969
970    /// Whether first-admin bootstrap must mint and deliver a new certificate
971    /// before the vault can be activated. Passphrase-backed and already
972    /// credentialed vaults preserve their existing seal and need no output.
973    pub fn bootstrap_requires_certificate_delivery(&self) -> bool {
974        self.pager.is_some()
975            && self
976                .vault
977                .read()
978                .map(|vault| vault.is_none())
979                .unwrap_or(true)
980    }
981
982    // -----------------------------------------------------------------
983    // Vault KV — encrypted key-value store for arbitrary secrets
984    // -----------------------------------------------------------------
985
986    /// Read a value from the vault KV store. Returns `None` if not set.
987    pub fn vault_kv_get(&self, key: &str) -> Option<String> {
988        self.vault_kv
989            .read()
990            .ok()
991            .and_then(|kv| kv.get(key).cloned())
992    }
993
994    /// Snapshot vault KV values for statement-local secret resolution.
995    pub fn vault_kv_snapshot(&self) -> HashMap<String, String> {
996        self.vault_kv
997            .read()
998            .map(|kv| kv.clone())
999            .unwrap_or_default()
1000    }
1001
1002    /// Export vault KV as an encrypted logical blob for JSONL dump/restore.
1003    /// Returns `None` when the vault has no KV entries.
1004    pub fn vault_kv_export_encrypted(&self) -> Result<Option<String>, AuthError> {
1005        if !self.is_vault_backed() {
1006            return Err(AuthError::Forbidden(
1007                "vault KV export requires an enabled, unsealed vault".to_string(),
1008            ));
1009        }
1010        let kv = self.vault_kv_snapshot();
1011        if kv.is_empty() {
1012            return Ok(None);
1013        }
1014
1015        let vault_guard = self.vault.read().map_err(lock_err)?;
1016        let vault = vault_guard.as_ref().ok_or_else(|| {
1017            AuthError::Forbidden("vault KV export requires an enabled, unsealed vault".to_string())
1018        })?;
1019        let state = VaultState {
1020            users: Vec::new(),
1021            api_keys: Vec::new(),
1022            bootstrapped: false,
1023            master_secret: None,
1024            kv,
1025        };
1026        Ok(Some(vault.seal_logical_export(&state)?))
1027    }
1028
1029    /// Merge imported vault KV entries and fail if the encrypted vault
1030    /// write cannot be made durable.
1031    pub fn vault_kv_try_import(
1032        &self,
1033        entries: HashMap<String, String>,
1034    ) -> Result<usize, AuthError> {
1035        if !self.is_vault_backed() {
1036            return Err(AuthError::Forbidden(
1037                "vault KV import requires an enabled, unsealed vault".to_string(),
1038            ));
1039        }
1040        if entries.is_empty() {
1041            return Ok(0);
1042        }
1043
1044        let mut previous = HashMap::new();
1045        {
1046            let mut kv = self.vault_kv.write().map_err(lock_err)?;
1047            for (key, value) in &entries {
1048                previous.insert(key.clone(), kv.insert(key.clone(), value.clone()));
1049            }
1050        }
1051
1052        if let Err(err) = self.persist_to_vault_result() {
1053            if let Ok(mut kv) = self.vault_kv.write() {
1054                for (key, old) in previous {
1055                    match old {
1056                        Some(value) => {
1057                            kv.insert(key, value);
1058                        }
1059                        None => {
1060                            kv.remove(&key);
1061                        }
1062                    }
1063                }
1064            }
1065            return Err(err);
1066        }
1067
1068        Ok(entries.len())
1069    }
1070
1071    /// Import false placeholders for secrets whose encrypted payload
1072    /// could not be decrypted during logical restore.
1073    pub fn vault_kv_try_import_placeholders(&self, keys: &[String]) -> Result<usize, AuthError> {
1074        let entries = keys
1075            .iter()
1076            .map(|key| (key.clone(), "false".to_string()))
1077            .collect();
1078        self.vault_kv_try_import(entries)
1079    }
1080
1081    /// Write a value to the vault KV store, persisting to disk.
1082    pub fn vault_kv_set(&self, key: String, value: String) {
1083        if let Ok(mut kv) = self.vault_kv.write() {
1084            kv.insert(key, value);
1085        }
1086        self.persist_to_vault();
1087    }
1088
1089    /// Write a value to the vault KV store and fail if the vault write
1090    /// cannot be made durable.
1091    pub fn vault_kv_try_set(&self, key: String, value: String) -> Result<(), AuthError> {
1092        if !self.is_vault_backed() {
1093            return Err(AuthError::Forbidden(
1094                "SET SECRET requires an enabled, unsealed vault".to_string(),
1095            ));
1096        }
1097
1098        let previous = {
1099            let mut kv = self.vault_kv.write().map_err(lock_err)?;
1100            kv.insert(key.clone(), value)
1101        };
1102
1103        if let Err(err) = self.persist_to_vault_result() {
1104            if let Ok(mut kv) = self.vault_kv.write() {
1105                match previous {
1106                    Some(value) => {
1107                        kv.insert(key, value);
1108                    }
1109                    None => {
1110                        kv.remove(&key);
1111                    }
1112                }
1113            }
1114            return Err(err);
1115        }
1116
1117        Ok(())
1118    }
1119
1120    /// Delete a value from the vault KV store. Returns true if it existed.
1121    pub fn vault_kv_delete(&self, key: &str) -> bool {
1122        let existed = self
1123            .vault_kv
1124            .write()
1125            .map(|mut kv| kv.remove(key).is_some())
1126            .unwrap_or(false);
1127        if existed {
1128            self.persist_to_vault();
1129        }
1130        existed
1131    }
1132
1133    /// Delete a value from the vault KV store and fail if the vault write
1134    /// cannot be made durable.
1135    pub fn vault_kv_try_delete(&self, key: &str) -> Result<bool, AuthError> {
1136        if !self.is_vault_backed() {
1137            return Err(AuthError::Forbidden(
1138                "DELETE SECRET requires an enabled, unsealed vault".to_string(),
1139            ));
1140        }
1141
1142        let removed = {
1143            let mut kv = self.vault_kv.write().map_err(lock_err)?;
1144            kv.remove(key)
1145        };
1146
1147        if removed.is_none() {
1148            return Ok(false);
1149        }
1150
1151        if let Err(err) = self.persist_to_vault_result() {
1152            if let Ok(mut kv) = self.vault_kv.write() {
1153                if let Some(value) = removed {
1154                    kv.insert(key.to_string(), value);
1155                }
1156            }
1157            return Err(err);
1158        }
1159
1160        Ok(true)
1161    }
1162
1163    /// List all keys in the vault KV store.
1164    pub fn vault_kv_keys(&self) -> Vec<String> {
1165        self.vault_kv
1166            .read()
1167            .map(|kv| kv.keys().cloned().collect())
1168            .unwrap_or_default()
1169    }
1170
1171    // -----------------------------------------------------------------
1172    // Plain KV — non-encrypted user key-value store (#1602)
1173    //
1174    // A sibling to the vault KV above, but stored in its own flat-map
1175    // with no encryption. Backs `$kv.*` / `SET KV` / `DELETE KV` and is
1176    // gated by `kv:read` / `kv:write` at the runtime layer. Kept fully
1177    // separate from `vault_kv` (secrets) and the config store.
1178    // -----------------------------------------------------------------
1179
1180    /// Read a value from the plain KV store. Returns `None` if not set.
1181    pub fn plain_kv_get(&self, key: &str) -> Option<String> {
1182        self.plain_kv
1183            .read()
1184            .ok()
1185            .and_then(|kv| kv.get(key).cloned())
1186    }
1187
1188    /// Snapshot plain KV values for statement-local resolution.
1189    pub fn plain_kv_snapshot(&self) -> HashMap<String, String> {
1190        self.plain_kv
1191            .read()
1192            .map(|kv| kv.clone())
1193            .unwrap_or_default()
1194    }
1195
1196    /// Write a value to the plain KV store.
1197    pub fn plain_kv_set(&self, key: String, value: String) {
1198        if let Ok(mut kv) = self.plain_kv.write() {
1199            kv.insert(key, value);
1200        }
1201    }
1202
1203    /// Delete a value from the plain KV store. Returns true if it existed.
1204    pub fn plain_kv_delete(&self, key: &str) -> bool {
1205        self.plain_kv
1206            .write()
1207            .map(|mut kv| kv.remove(key).is_some())
1208            .unwrap_or(false)
1209    }
1210
1211    /// List all keys in the plain KV store.
1212    pub fn plain_kv_keys(&self) -> Vec<String> {
1213        self.plain_kv
1214            .read()
1215            .map(|kv| kv.keys().cloned().collect())
1216            .unwrap_or_default()
1217    }
1218
1219    /// Convenience: get the 32-byte secret key for Value::Secret encryption.
1220    /// Generated on first boot and stored at `red.secret.aes_key`.
1221    pub fn vault_secret_key(&self) -> Option<[u8; 32]> {
1222        let hex_str = self.vault_kv_get("red.secret.aes_key")?;
1223        let bytes = hex::decode(hex_str).ok()?;
1224        if bytes.len() == 32 {
1225            let mut key = [0u8; 32];
1226            key.copy_from_slice(&bytes);
1227            Some(key)
1228        } else {
1229            None
1230        }
1231    }
1232
1233    /// Generate and store the AES-256 secret key on first boot if not present.
1234    pub fn ensure_vault_secret_key(&self) {
1235        if let Err(err) = self.ensure_vault_secret_key_result() {
1236            tracing::error!(error = %err, "vault secret key initialization failed");
1237        }
1238    }
1239
1240    pub(crate) fn ensure_vault_secret_key_result(&self) -> Result<(), AuthError> {
1241        let _guard = self.secret_key_init.lock().map_err(lock_err)?;
1242        if self.vault_secret_key().is_some() {
1243            return Ok(());
1244        }
1245        if self.vault_kv_get("red.secret.aes_key").is_some() {
1246            return Err(AuthError::Internal(
1247                "stored vault secret key is invalid".into(),
1248            ));
1249        }
1250        let value = hex::encode(random_bytes(32));
1251        if self.is_vault_backed() {
1252            self.vault_kv_try_set("red.secret.aes_key".into(), value)
1253        } else {
1254            self.vault_kv
1255                .write()
1256                .map_err(lock_err)?
1257                .insert("red.secret.aes_key".into(), value);
1258            Ok(())
1259        }
1260    }
1261
1262    /// Take a snapshot of the current auth state for vault serialization.
1263    fn snapshot(&self) -> VaultState {
1264        let users_guard = self.users.read().unwrap_or_else(|e| e.into_inner());
1265        let users: Vec<User> = users_guard.values().cloned().collect();
1266
1267        // Collect (owner UserId, api_key) pairs from all users so a
1268        // tenant-scoped owner can be reattached on restore.
1269        let mut api_keys = Vec::new();
1270        for user in &users {
1271            let owner = UserId::from_parts(user.tenant_id.as_deref(), &user.username);
1272            for key in &user.api_keys {
1273                api_keys.push((owner.clone(), key.clone()));
1274            }
1275        }
1276
1277        // Include the master secret if a keypair is loaded.
1278        let master_secret = self
1279            .keypair
1280            .read()
1281            .ok()
1282            .and_then(|guard| guard.as_ref().map(|kp| kp.master_secret.clone()));
1283
1284        let kv = self.vault_kv.read().map(|m| m.clone()).unwrap_or_default();
1285
1286        VaultState {
1287            users,
1288            api_keys,
1289            bootstrapped: self.bootstrapped.load(Ordering::Acquire),
1290            master_secret,
1291            kv,
1292        }
1293    }
1294
1295    /// Restore the in-memory auth state from a vault snapshot.
1296    fn restore_from_vault(&mut self, state: VaultState) {
1297        // Restore bootstrap seal.
1298        if state.bootstrapped {
1299            self.bootstrapped.store(true, Ordering::Release);
1300        }
1301
1302        // Restore keypair from master secret (if present).
1303        if let Some(secret) = state.master_secret {
1304            let kp = KeyPair::from_master_secret(secret);
1305            if let Ok(mut guard) = self.keypair.write() {
1306                *guard = Some(kp);
1307            }
1308        }
1309
1310        // Restore KV store.
1311        if let Ok(mut kv) = self.vault_kv.write() {
1312            *kv = state.kv;
1313        }
1314
1315        // Restore users.
1316        let mut users = self.users.write().unwrap_or_else(|e| e.into_inner());
1317        let mut idx = self
1318            .api_key_index
1319            .write()
1320            .unwrap_or_else(|e| e.into_inner());
1321
1322        for mut user in state.users {
1323            let id = UserId::from_parts(user.tenant_id.as_deref(), &user.username);
1324            // A vault written before keys were hashed at rest holds the
1325            // secret itself. Hash it on the way in; the next persist
1326            // rewrites the line and the secret is gone from disk.
1327            for key in &mut user.api_keys {
1328                if !key.key.starts_with("rkh_") {
1329                    key.key = api_key_credential_id(&key.key);
1330                }
1331            }
1332            // Register API keys in the index.
1333            for key in &user.api_keys {
1334                idx.insert(key.key.clone(), (id.clone(), key.role));
1335            }
1336            users.insert(id, user);
1337        }
1338        drop(idx);
1339        drop(users);
1340
1341        self.rehydrate_acl();
1342        self.rehydrate_iam();
1343    }
1344
1345    // -----------------------------------------------------------------
1346    // User management
1347    // -----------------------------------------------------------------
1348
1349    /// Create a new platform-scoped user (`tenant_id = None`).
1350    ///
1351    /// For tenant-scoped creation, use [`Self::create_user_in_tenant`].
1352    pub fn create_user(
1353        &self,
1354        username: &str,
1355        password: &str,
1356        role: Role,
1357    ) -> Result<User, AuthError> {
1358        self.create_user_in_tenant(None, username, password, role)
1359    }
1360
1361    pub fn create_user_with_control_events(
1362        &self,
1363        username: &str,
1364        password: &str,
1365        role: Role,
1366        ctx: &ControlEventCtx<'_>,
1367        ledger: &dyn ControlEventLedger,
1368        config: ControlEventConfig,
1369    ) -> Result<User, AuthError> {
1370        self.create_user_in_tenant_with_control_events(
1371            None, username, password, role, ctx, ledger, config,
1372        )
1373    }
1374
1375    /// Create a user under the given tenant scope. `tenant_id == None`
1376    /// produces a platform-wide user. `(tenant, username)` is the
1377    /// uniqueness key — the same `username` may exist independently
1378    /// under multiple tenants.
1379    pub fn create_user_in_tenant(
1380        &self,
1381        tenant_id: Option<&str>,
1382        username: &str,
1383        password: &str,
1384        role: Role,
1385    ) -> Result<User, AuthError> {
1386        if let Some(configured) = self.configured_control_events() {
1387            let ctx = default_user_lifecycle_ctx();
1388            let control = UserLifecycleControl {
1389                ctx: &ctx,
1390                ledger: configured.ledger.as_ref(),
1391                config: configured.config,
1392            };
1393            self.create_user_in_tenant_controlled(tenant_id, username, password, role, &control)
1394        } else {
1395            self.create_user_in_tenant_unaudited(tenant_id, username, password, role)
1396        }
1397    }
1398
1399    #[allow(clippy::too_many_arguments)]
1400    pub fn create_user_in_tenant_with_control_events(
1401        &self,
1402        tenant_id: Option<&str>,
1403        username: &str,
1404        password: &str,
1405        role: Role,
1406        ctx: &ControlEventCtx<'_>,
1407        ledger: &dyn ControlEventLedger,
1408        config: ControlEventConfig,
1409    ) -> Result<User, AuthError> {
1410        let control = UserLifecycleControl {
1411            ctx,
1412            ledger,
1413            config,
1414        };
1415        self.create_user_in_tenant_controlled(tenant_id, username, password, role, &control)
1416    }
1417
1418    pub fn create_admin_user(
1419        &self,
1420        username: &str,
1421        password: &str,
1422        role: Role,
1423        tenant_id: Option<&str>,
1424    ) -> Result<User, AuthError> {
1425        if let Some(configured) = self.configured_control_events() {
1426            let ctx = default_user_lifecycle_ctx();
1427            let control = UserLifecycleControl {
1428                ctx: &ctx,
1429                ledger: configured.ledger.as_ref(),
1430                config: configured.config,
1431            };
1432            self.create_user_in_tenant_controlled(tenant_id, username, password, role, &control)
1433        } else {
1434            self.create_user_in_tenant_unaudited(tenant_id, username, password, role)
1435        }
1436    }
1437
1438    fn create_user_in_tenant_unaudited(
1439        &self,
1440        tenant_id: Option<&str>,
1441        username: &str,
1442        password: &str,
1443        role: Role,
1444    ) -> Result<User, AuthError> {
1445        let id = UserId::from_parts(tenant_id, username);
1446        let mut users = self.users.write().map_err(lock_err)?;
1447        if users.contains_key(&id) {
1448            return Err(AuthError::UserExists(id.to_string()));
1449        }
1450
1451        let now = now_ms();
1452        let user = User {
1453            username: username.to_string(),
1454            tenant_id: tenant_id.map(|s| s.to_string()),
1455            password_hash: hash_password(password),
1456            scram_verifier: Some(make_scram_verifier(password)),
1457            role,
1458            api_keys: Vec::new(),
1459            created_at: now,
1460            updated_at: now,
1461            enabled: true,
1462        };
1463        users.insert(id, user.clone());
1464        drop(users); // release lock before vault I/O
1465        self.persist_to_vault();
1466        Ok(user)
1467    }
1468
1469    fn create_user_in_tenant_controlled(
1470        &self,
1471        tenant_id: Option<&str>,
1472        username: &str,
1473        password: &str,
1474        role: Role,
1475        control: &UserLifecycleControl<'_>,
1476    ) -> Result<User, AuthError> {
1477        let id = UserId::from_parts(tenant_id, username);
1478        match self.create_user_in_tenant_unaudited(tenant_id, username, password, role) {
1479            Ok(user) => {
1480                if let Err(err) = self.emit_user_lifecycle_allowed(
1481                    control,
1482                    EventKind::UserCreate,
1483                    "user.create",
1484                    &id,
1485                    user_control_fields(&id, Some(role), Some(true), true),
1486                ) {
1487                    self.rollback_create_user(&id);
1488                    return Err(err);
1489                }
1490                Ok(user)
1491            }
1492            Err(err) => {
1493                self.emit_user_lifecycle_outcome(
1494                    control,
1495                    if user_error_is_denied(&err) {
1496                        Outcome::Denied
1497                    } else {
1498                        Outcome::Error
1499                    },
1500                    EventKind::UserCreate,
1501                    "user.create",
1502                    &id,
1503                    Some(err.to_string()),
1504                    user_control_fields(&id, Some(role), Some(true), true),
1505                );
1506                Err(err)
1507            }
1508        }
1509    }
1510
1511    /// Look up a user's SCRAM verifier by full `UserId`.
1512    ///
1513    /// The wire handshake passes the tenant resolved from the session
1514    /// (or `None` for the bootstrap admin) so cross-tenant collisions
1515    /// never authenticate the wrong identity.
1516    pub fn lookup_scram_verifier(&self, id: &UserId) -> Option<crate::auth::scram::ScramVerifier> {
1517        // Synthetic platform-owner principal must never authenticate.
1518        if crate::auth::self_lock_guard::is_synthetic_principal(&id.username) {
1519            return None;
1520        }
1521        let users = self.users.read().ok()?;
1522        // A disabled account keeps its verifier (re-enable restores it)
1523        // but must not authenticate; returning `None` here makes the
1524        // handshake use the dummy verifier, so timing stays uniform.
1525        users
1526            .get(id)
1527            .filter(|u| u.enabled)
1528            .and_then(|u| u.scram_verifier.clone())
1529    }
1530
1531    /// Backwards-compatible shim for the v2 wire bootstrap path: looks
1532    /// up a user by username assuming the platform (`tenant=None`)
1533    /// scope. Use this only where the handshake hasn't yet learned the
1534    /// caller's tenant.
1535    pub fn lookup_scram_verifier_global(
1536        &self,
1537        username: &str,
1538    ) -> Option<crate::auth::scram::ScramVerifier> {
1539        self.lookup_scram_verifier(&UserId::platform(username))
1540    }
1541
1542    /// Return all users (password hashes redacted).
1543    ///
1544    /// The synthetic [`crate::auth::self_lock_guard::PLATFORM_OWNER_USERNAME`]
1545    /// principal is filtered out — it is a policy-graph anchor, not an
1546    /// operator-visible account.
1547    pub fn list_users(&self) -> Vec<User> {
1548        let users = match self.users.read() {
1549            Ok(g) => g,
1550            Err(_) => return Vec::new(),
1551        };
1552        users
1553            .values()
1554            .filter(|u| !crate::auth::self_lock_guard::is_synthetic_principal(&u.username))
1555            .map(|u| User {
1556                password_hash: String::new(), // redacted
1557                ..u.clone()
1558            })
1559            .collect()
1560    }
1561
1562    /// Return users restricted to a tenant scope.
1563    ///
1564    /// `tenant_filter`:
1565    ///   - `None` listing in `Some(None)` — only platform users
1566    ///   - `Some(Some("acme"))` — only users in tenant `acme`
1567    ///   - `None` argument — all users (admin-only callers)
1568    pub fn list_users_scoped(&self, tenant_filter: Option<Option<&str>>) -> Vec<User> {
1569        let users = match self.users.read() {
1570            Ok(g) => g,
1571            Err(_) => return Vec::new(),
1572        };
1573        users
1574            .values()
1575            .filter(|u| match tenant_filter {
1576                None => true,
1577                Some(t) => u.tenant_id.as_deref() == t,
1578            })
1579            .filter(|u| !crate::auth::self_lock_guard::is_synthetic_principal(&u.username))
1580            .map(|u| User {
1581                password_hash: String::new(), // redacted
1582                ..u.clone()
1583            })
1584            .collect()
1585    }
1586
1587    /// Resource shape used by IAM policies that govern user lifecycle
1588    /// mutations. Platform users are addressed as `user:<username>`;
1589    /// tenant users are addressed as `user:tenant/<tenant>/<username>`
1590    /// by the existing policy resource qualifier.
1591    pub fn user_lifecycle_resource(uid: &UserId) -> ResourceRef {
1592        let resource = ResourceRef::new("user", uid.username.clone());
1593        match &uid.tenant {
1594            Some(tenant) => resource.with_tenant(tenant.clone()),
1595            None => resource,
1596        }
1597    }
1598
1599    pub fn eval_context_for_principal(
1600        &self,
1601        principal: &UserId,
1602        role: Role,
1603        current_tenant: Option<String>,
1604    ) -> EvalContext {
1605        EvalContext {
1606            principal_tenant: principal.tenant.clone(),
1607            current_tenant,
1608            peer_ip: None,
1609            mfa_present: false,
1610            now_ms: now_ms(),
1611            principal_is_admin_role: role == Role::Admin,
1612            principal_is_platform_scoped: principal.tenant.is_none(),
1613        }
1614    }
1615
1616    /// Policy gate for user lifecycle mutations.
1617    ///
1618    /// This is intentionally separate from the store mutation methods:
1619    /// bootstrap and manifest loading can still seed users directly,
1620    /// while public surfaces must authorize `user:*` before calling the
1621    /// mutator. Explicit Deny wins even in `LegacyRbac` mode.
1622    pub fn check_user_lifecycle_authz(
1623        &self,
1624        actor: &UserId,
1625        actor_role: Role,
1626        action: &str,
1627        target: &UserId,
1628    ) -> bool {
1629        let resource = Self::user_lifecycle_resource(target);
1630        let ctx = self.eval_context_for_principal(actor, actor_role, target.tenant.clone());
1631        if actor.tenant.is_some() && actor.tenant != target.tenant {
1632            return false;
1633        }
1634        if self.has_explicit_user_lifecycle_deny(actor, actor_role, action, target) {
1635            return false;
1636        }
1637        actor_role.can_admin()
1638            || self.check_policy_authz_with_role(actor, action, &resource, &ctx, actor_role)
1639    }
1640
1641    /// Returns `true` only when an IAM policy explicitly denies the
1642    /// action on the target. `DefaultDeny` (no matching policy) is
1643    /// treated as permitted, so this is safe to call after the caller
1644    /// has already verified the actor's role — it only blocks protected
1645    /// principals (e.g. the cloud-preset `red_admin` guardrail).
1646    pub fn has_explicit_user_lifecycle_deny(
1647        &self,
1648        actor: &UserId,
1649        actor_role: Role,
1650        action: &str,
1651        target: &UserId,
1652    ) -> bool {
1653        let resource = Self::user_lifecycle_resource(target);
1654        let ctx = self.eval_context_for_principal(actor, actor_role, target.tenant.clone());
1655        let pols = self.effective_policies(actor);
1656        let refs: Vec<&Policy> = pols.iter().map(|p| p.as_ref()).collect();
1657        matches!(
1658            iam_policies::evaluate(&refs, action, &resource, &ctx),
1659            iam_policies::Decision::Deny { .. }
1660        )
1661    }
1662
1663    pub fn get_user(&self, tenant_id: Option<&str>, username: &str) -> Option<User> {
1664        let id = UserId::from_parts(tenant_id, username);
1665        self.get_user_cloned(&id).map(|u| User {
1666            password_hash: String::new(),
1667            ..u
1668        })
1669    }
1670
1671    /// Delete a platform-scoped user (`tenant_id = None`) and revoke
1672    /// all of their API keys + sessions.
1673    ///
1674    /// For tenant-scoped deletes, use [`Self::delete_user_in_tenant`].
1675    pub fn delete_user(&self, username: &str) -> Result<(), AuthError> {
1676        self.delete_user_in_tenant(None, username)
1677    }
1678
1679    pub fn delete_user_with_control_events(
1680        &self,
1681        username: &str,
1682        ctx: &ControlEventCtx<'_>,
1683        ledger: &dyn ControlEventLedger,
1684        config: ControlEventConfig,
1685    ) -> Result<(), AuthError> {
1686        self.delete_user_in_tenant_with_control_events(None, username, ctx, ledger, config)
1687    }
1688
1689    /// Delete a user identified by `(tenant_id, username)` and revoke
1690    /// all of their API keys + sessions.
1691    pub fn delete_user_in_tenant(
1692        &self,
1693        tenant_id: Option<&str>,
1694        username: &str,
1695    ) -> Result<(), AuthError> {
1696        if let Some(configured) = self.configured_control_events() {
1697            let ctx = default_user_lifecycle_ctx();
1698            let control = UserLifecycleControl {
1699                ctx: &ctx,
1700                ledger: configured.ledger.as_ref(),
1701                config: configured.config,
1702            };
1703            self.delete_user_in_tenant_controlled(tenant_id, username, &control)
1704        } else {
1705            self.delete_user_in_tenant_unaudited(tenant_id, username)
1706        }
1707    }
1708
1709    pub fn delete_user_in_tenant_with_control_events(
1710        &self,
1711        tenant_id: Option<&str>,
1712        username: &str,
1713        ctx: &ControlEventCtx<'_>,
1714        ledger: &dyn ControlEventLedger,
1715        config: ControlEventConfig,
1716    ) -> Result<(), AuthError> {
1717        let control = UserLifecycleControl {
1718            ctx,
1719            ledger,
1720            config,
1721        };
1722        self.delete_user_in_tenant_controlled(tenant_id, username, &control)
1723    }
1724
1725    fn delete_user_in_tenant_unaudited(
1726        &self,
1727        tenant_id: Option<&str>,
1728        username: &str,
1729    ) -> Result<(), AuthError> {
1730        let id = UserId::from_parts(tenant_id, username);
1731        let mut users = self.users.write().map_err(lock_err)?;
1732        let user = users
1733            .remove(&id)
1734            .ok_or_else(|| AuthError::UserNotFound(id.to_string()))?;
1735
1736        // Remove API key index entries.
1737        if let Ok(mut idx) = self.api_key_index.write() {
1738            for api_key in &user.api_keys {
1739                idx.remove(&api_key.key);
1740            }
1741        }
1742
1743        self.purge_sessions_for(&id);
1744
1745        self.persist_to_vault();
1746        Ok(())
1747    }
1748
1749    fn delete_user_in_tenant_controlled(
1750        &self,
1751        tenant_id: Option<&str>,
1752        username: &str,
1753        control: &UserLifecycleControl<'_>,
1754    ) -> Result<(), AuthError> {
1755        let id = UserId::from_parts(tenant_id, username);
1756        let rollback = self.user_delete_rollback_snapshot(&id, tenant_id, username);
1757        match self.delete_user_in_tenant_unaudited(tenant_id, username) {
1758            Ok(()) => {
1759                if let Err(err) = self.emit_user_lifecycle_allowed(
1760                    control,
1761                    EventKind::UserDelete,
1762                    "user.delete",
1763                    &id,
1764                    user_control_fields(&id, rollback.as_ref().map(|r| r.0.role), None, false),
1765                ) {
1766                    if let Some((user, sessions)) = rollback {
1767                        self.restore_deleted_user(&id, user, sessions);
1768                    }
1769                    return Err(err);
1770                }
1771                Ok(())
1772            }
1773            Err(err) => {
1774                self.emit_user_lifecycle_outcome(
1775                    control,
1776                    if user_error_is_denied(&err) {
1777                        Outcome::Denied
1778                    } else {
1779                        Outcome::Error
1780                    },
1781                    EventKind::UserDelete,
1782                    "user.delete",
1783                    &id,
1784                    Some(err.to_string()),
1785                    user_control_fields(&id, rollback.as_ref().map(|r| r.0.role), None, false),
1786                );
1787                Err(err)
1788            }
1789        }
1790    }
1791
1792    /// Change password (requires the old password). Defaults to
1793    /// platform tenant; use [`Self::change_password_in_tenant`] for
1794    /// scoped users.
1795    pub fn change_password(
1796        &self,
1797        username: &str,
1798        old_password: &str,
1799        new_password: &str,
1800    ) -> Result<(), AuthError> {
1801        self.change_password_in_tenant(None, username, old_password, new_password)
1802    }
1803
1804    pub fn change_password_with_control_events(
1805        &self,
1806        username: &str,
1807        old_password: &str,
1808        new_password: &str,
1809        ctx: &ControlEventCtx<'_>,
1810        ledger: &dyn ControlEventLedger,
1811        config: ControlEventConfig,
1812    ) -> Result<(), AuthError> {
1813        self.change_password_in_tenant_with_control_events(
1814            None,
1815            username,
1816            old_password,
1817            new_password,
1818            ctx,
1819            ledger,
1820            config,
1821        )
1822    }
1823
1824    #[allow(clippy::too_many_arguments)]
1825    pub fn change_password_in_tenant(
1826        &self,
1827        tenant_id: Option<&str>,
1828        username: &str,
1829        old_password: &str,
1830        new_password: &str,
1831    ) -> Result<(), AuthError> {
1832        if let Some(configured) = self.configured_control_events() {
1833            let ctx = default_user_lifecycle_ctx();
1834            let control = UserLifecycleControl {
1835                ctx: &ctx,
1836                ledger: configured.ledger.as_ref(),
1837                config: configured.config,
1838            };
1839            self.change_password_in_tenant_controlled(
1840                tenant_id,
1841                username,
1842                old_password,
1843                new_password,
1844                &control,
1845            )
1846        } else {
1847            self.change_password_in_tenant_unaudited(
1848                tenant_id,
1849                username,
1850                old_password,
1851                new_password,
1852            )
1853        }
1854    }
1855
1856    #[allow(clippy::too_many_arguments)]
1857    pub fn change_password_in_tenant_with_control_events(
1858        &self,
1859        tenant_id: Option<&str>,
1860        username: &str,
1861        old_password: &str,
1862        new_password: &str,
1863        ctx: &ControlEventCtx<'_>,
1864        ledger: &dyn ControlEventLedger,
1865        config: ControlEventConfig,
1866    ) -> Result<(), AuthError> {
1867        let control = UserLifecycleControl {
1868            ctx,
1869            ledger,
1870            config,
1871        };
1872        self.change_password_in_tenant_controlled(
1873            tenant_id,
1874            username,
1875            old_password,
1876            new_password,
1877            &control,
1878        )
1879    }
1880
1881    fn change_password_in_tenant_unaudited(
1882        &self,
1883        tenant_id: Option<&str>,
1884        username: &str,
1885        old_password: &str,
1886        new_password: &str,
1887    ) -> Result<(), AuthError> {
1888        let id = UserId::from_parts(tenant_id, username);
1889        let mut users = self.users.write().map_err(lock_err)?;
1890        let user = users
1891            .get_mut(&id)
1892            .ok_or_else(|| AuthError::UserNotFound(id.to_string()))?;
1893
1894        if !verify_password(old_password, &user.password_hash) {
1895            return Err(AuthError::InvalidCredentials);
1896        }
1897
1898        user.password_hash = hash_password(new_password);
1899        user.scram_verifier = Some(make_scram_verifier(new_password));
1900        user.updated_at = now_ms();
1901        drop(users); // release lock before vault I/O
1902        self.persist_to_vault();
1903        Ok(())
1904    }
1905
1906    /// Administrative password reset (`ALTER USER … PASSWORD '…'`): no
1907    /// old-password check. Rehashes, regenerates the SCRAM verifier and
1908    /// revokes every live session of the account so a rotated credential
1909    /// cannot keep an existing session alive. Callers gate authorization.
1910    pub fn set_password_by_admin(&self, id: &UserId, new_password: &str) -> Result<(), AuthError> {
1911        if new_password.is_empty() {
1912            return Err(AuthError::Forbidden("password cannot be empty".to_string()));
1913        }
1914        {
1915            let mut users = self.users.write().map_err(lock_err)?;
1916            let user = users
1917                .get_mut(id)
1918                .ok_or_else(|| AuthError::UserNotFound(id.to_string()))?;
1919            user.password_hash = hash_password(new_password);
1920            user.scram_verifier = Some(make_scram_verifier(new_password));
1921            user.updated_at = now_ms();
1922        }
1923        self.purge_sessions_for(id);
1924        self.persist_to_vault();
1925        Ok(())
1926    }
1927
1928    fn change_password_in_tenant_controlled(
1929        &self,
1930        tenant_id: Option<&str>,
1931        username: &str,
1932        old_password: &str,
1933        new_password: &str,
1934        control: &UserLifecycleControl<'_>,
1935    ) -> Result<(), AuthError> {
1936        let id = UserId::from_parts(tenant_id, username);
1937        let previous = self.get_user_cloned(&id);
1938        match self.change_password_in_tenant_unaudited(
1939            tenant_id,
1940            username,
1941            old_password,
1942            new_password,
1943        ) {
1944            Ok(()) => {
1945                if let Err(err) = self.emit_user_lifecycle_allowed(
1946                    control,
1947                    EventKind::UserUpdate,
1948                    "user.update",
1949                    &id,
1950                    user_control_fields(
1951                        &id,
1952                        previous.as_ref().map(|u| u.role),
1953                        previous.as_ref().map(|u| u.enabled),
1954                        true,
1955                    ),
1956                ) {
1957                    self.restore_user_snapshot(&id, previous);
1958                    return Err(err);
1959                }
1960                Ok(())
1961            }
1962            Err(err) => {
1963                self.emit_user_lifecycle_outcome(
1964                    control,
1965                    if user_error_is_denied(&err) {
1966                        Outcome::Denied
1967                    } else {
1968                        Outcome::Error
1969                    },
1970                    EventKind::UserUpdate,
1971                    "user.update",
1972                    &id,
1973                    Some(err.to_string()),
1974                    user_control_fields(
1975                        &id,
1976                        previous.as_ref().map(|u| u.role),
1977                        previous.as_ref().map(|u| u.enabled),
1978                        true,
1979                    ),
1980                );
1981                Err(err)
1982            }
1983        }
1984    }
1985
1986    /// Change a user's role (admin-only operation). Defaults to platform
1987    /// tenant; use [`Self::change_role_in_tenant`] for scoped users.
1988    pub fn change_role(&self, username: &str, new_role: Role) -> Result<(), AuthError> {
1989        self.change_role_in_tenant(None, username, new_role)
1990    }
1991
1992    pub fn change_role_in_tenant(
1993        &self,
1994        tenant_id: Option<&str>,
1995        username: &str,
1996        new_role: Role,
1997    ) -> Result<(), AuthError> {
1998        if let Some(configured) = self.configured_control_events() {
1999            let ctx = default_user_lifecycle_ctx();
2000            let control = UserLifecycleControl {
2001                ctx: &ctx,
2002                ledger: configured.ledger.as_ref(),
2003                config: configured.config,
2004            };
2005            self.change_role_in_tenant_controlled(tenant_id, username, new_role, &control)
2006        } else {
2007            self.change_role_in_tenant_unaudited(tenant_id, username, new_role)
2008        }
2009    }
2010
2011    pub fn change_role_in_tenant_with_control_events(
2012        &self,
2013        tenant_id: Option<&str>,
2014        username: &str,
2015        new_role: Role,
2016        ctx: &ControlEventCtx<'_>,
2017        ledger: &dyn ControlEventLedger,
2018        config: ControlEventConfig,
2019    ) -> Result<(), AuthError> {
2020        let control = UserLifecycleControl {
2021            ctx,
2022            ledger,
2023            config,
2024        };
2025        self.change_role_in_tenant_controlled(tenant_id, username, new_role, &control)
2026    }
2027
2028    fn change_role_in_tenant_unaudited(
2029        &self,
2030        tenant_id: Option<&str>,
2031        username: &str,
2032        new_role: Role,
2033    ) -> Result<(), AuthError> {
2034        let id = UserId::from_parts(tenant_id, username);
2035        let mut users = self.users.write().map_err(lock_err)?;
2036        let user = users
2037            .get_mut(&id)
2038            .ok_or_else(|| AuthError::UserNotFound(id.to_string()))?;
2039
2040        let prior_role = user.role;
2041        user.role = new_role;
2042        user.updated_at = now_ms();
2043
2044        // Issue #205 — promotion to Admin is an operator-grade event:
2045        // the new role grants destructive capabilities (DROP, ALTER,
2046        // GRANT) that an operator must observe out-of-band even when
2047        // the auth path itself is healthy.
2048        if new_role == Role::Admin && prior_role != Role::Admin {
2049            crate::telemetry::operator_event::OperatorEvent::AdminCapabilityGranted {
2050                granted_to: id.to_string(),
2051                capability: "Role::Admin".to_string(),
2052                granted_by: "auth_store::change_role".to_string(),
2053            }
2054            .emit_global();
2055        }
2056
2057        // Downgrade any API keys that now exceed the user's role.
2058        for key in &mut user.api_keys {
2059            if key.role > new_role {
2060                key.role = new_role;
2061            }
2062        }
2063
2064        // Update the api_key_index as well.
2065        if let Ok(mut idx) = self.api_key_index.write() {
2066            for key in &user.api_keys {
2067                if let Some(entry) = idx.get_mut(&key.key) {
2068                    entry.1 = key.role;
2069                }
2070            }
2071        }
2072
2073        // Sessions snapshot the role at login. A change in either direction
2074        // must not keep riding on a token minted under the old one — a
2075        // demotion would otherwise stay ineffective for the session TTL.
2076        if prior_role != new_role {
2077            self.purge_sessions_for(&id);
2078        }
2079
2080        self.persist_to_vault();
2081        Ok(())
2082    }
2083
2084    fn change_role_in_tenant_controlled(
2085        &self,
2086        tenant_id: Option<&str>,
2087        username: &str,
2088        new_role: Role,
2089        control: &UserLifecycleControl<'_>,
2090    ) -> Result<(), AuthError> {
2091        let id = UserId::from_parts(tenant_id, username);
2092        let previous = self.get_user_cloned(&id);
2093        match self.change_role_in_tenant_unaudited(tenant_id, username, new_role) {
2094            Ok(()) => {
2095                if let Err(err) = self.emit_user_lifecycle_allowed(
2096                    control,
2097                    EventKind::UserUpdate,
2098                    "user.update",
2099                    &id,
2100                    user_control_fields(
2101                        &id,
2102                        Some(new_role),
2103                        previous.as_ref().map(|u| u.enabled),
2104                        false,
2105                    ),
2106                ) {
2107                    self.restore_user_snapshot(&id, previous);
2108                    return Err(err);
2109                }
2110                Ok(())
2111            }
2112            Err(err) => {
2113                self.emit_user_lifecycle_outcome(
2114                    control,
2115                    if user_error_is_denied(&err) {
2116                        Outcome::Denied
2117                    } else {
2118                        Outcome::Error
2119                    },
2120                    EventKind::UserUpdate,
2121                    "user.update",
2122                    &id,
2123                    Some(err.to_string()),
2124                    user_control_fields(
2125                        &id,
2126                        Some(new_role),
2127                        previous.as_ref().map(|u| u.enabled),
2128                        false,
2129                    ),
2130                );
2131                Err(err)
2132            }
2133        }
2134    }
2135
2136    // -----------------------------------------------------------------
2137    // Authentication (login)
2138    // -----------------------------------------------------------------
2139
2140    /// Verify credentials for a platform-tenant user (`tenant_id = None`)
2141    /// and create a session. For tenant-scoped login use
2142    /// [`Self::authenticate_in_tenant`].
2143    ///
2144    /// When a keypair is available (certificate-based seal), session tokens
2145    /// are signed with the master secret so the server can verify they were
2146    /// genuinely issued by this vault instance.
2147    pub fn authenticate(&self, username: &str, password: &str) -> Result<Session, AuthError> {
2148        self.authenticate_in_tenant(None, username, password)
2149    }
2150
2151    /// Verify credentials for `(tenant_id, username, password)` and
2152    /// create a session. Tenant-aware: `alice@acme` and `alice@globex`
2153    /// authenticate independently.
2154    ///
2155    /// Every outcome is recorded in the control-event ledger when one is
2156    /// configured: login was the one credential decision with no chained
2157    /// trace, so a brute-force run or a login from a freshly re-enabled
2158    /// account was invisible to an audit of the store. Under a
2159    /// compliance-mode ledger a login that cannot be recorded is refused.
2160    pub fn authenticate_in_tenant(
2161        &self,
2162        tenant_id: Option<&str>,
2163        username: &str,
2164        password: &str,
2165    ) -> Result<Session, AuthError> {
2166        let id = UserId::from_parts(tenant_id, username);
2167        match self.authenticate_in_tenant_unaudited(tenant_id, username, password) {
2168            Ok(session) => {
2169                if let Err(err) = self.emit_auth_event(
2170                    EventKind::AuthLogin,
2171                    Outcome::Allowed,
2172                    &id,
2173                    Some(session.role),
2174                    None,
2175                ) {
2176                    if let Ok(mut sessions) = self.sessions.write() {
2177                        sessions.remove(&session.token);
2178                    }
2179                    return Err(err);
2180                }
2181                Ok(session)
2182            }
2183            Err(err) => {
2184                let outcome = if matches!(err, AuthError::InvalidCredentials) {
2185                    Outcome::Denied
2186                } else {
2187                    Outcome::Error
2188                };
2189                // Already refusing; a ledger failure cannot make it worse.
2190                let _ = self.emit_auth_event(
2191                    EventKind::AuthLoginFailed,
2192                    outcome,
2193                    &id,
2194                    None,
2195                    Some(&err.to_string()),
2196                );
2197                Err(err)
2198            }
2199        }
2200    }
2201
2202    /// Record one authentication decision. `Err` only when the ledger is
2203    /// configured to require persistence and refused the event.
2204    fn emit_auth_event(
2205        &self,
2206        kind: EventKind,
2207        outcome: Outcome,
2208        id: &UserId,
2209        role: Option<Role>,
2210        reason: Option<&str>,
2211    ) -> Result<(), AuthError> {
2212        let Some(configured) = self.configured_control_events() else {
2213            return Ok(());
2214        };
2215        let ctx = ControlEventCtx {
2216            actor: crate::runtime::control_events::ActorRef::User(id),
2217            scope: id.tenant.as_deref().map(Cow::Borrowed),
2218            request_id: None,
2219            trace_id: None,
2220        };
2221        let mut fields: HashMap<String, Sensitivity> = HashMap::new();
2222        fields.insert("principal".to_string(), Sensitivity::raw(id.to_string()));
2223        if let Some(role) = role {
2224            fields.insert("role".to_string(), Sensitivity::raw(role.as_str()));
2225        }
2226        let event = ControlEvent {
2227            kind,
2228            outcome,
2229            action: Cow::Borrowed(kind.as_str()),
2230            resource: Some(format!("principal:{id}")),
2231            reason: reason.map(str::to_string),
2232            matched_policy_id: None,
2233            fields,
2234        };
2235        match configured.ledger.emit(&ctx, event) {
2236            Ok(_) => Ok(()),
2237            Err(err) if configured.config.require_persistence() => Err(AuthError::Internal(
2238                format!("control-event ledger refused {}: {err}", kind.as_str()),
2239            )),
2240            Err(err) => {
2241                tracing::warn!(
2242                    target: "reddb::security",
2243                    kind = kind.as_str(),
2244                    error = %err,
2245                    "control-event ledger refused an auth event"
2246                );
2247                Ok(())
2248            }
2249        }
2250    }
2251
2252    fn authenticate_in_tenant_unaudited(
2253        &self,
2254        tenant_id: Option<&str>,
2255        username: &str,
2256        password: &str,
2257    ) -> Result<Session, AuthError> {
2258        // Initialize once on the first login, regardless of whether the
2259        // principal exists. Subsequent missing/disabled-account attempts use
2260        // this verifier instead of returning before production-cost Argon2.
2261        let dummy_hash = dummy_password_hash();
2262        // The synthetic platform-owner principal exists only as a
2263        // policy-graph anchor (see `crate::auth::self_lock_guard`); it
2264        // must never authenticate via any transport.
2265        if crate::auth::self_lock_guard::is_synthetic_principal(username) {
2266            let _ = verify_password(password, dummy_hash);
2267            return Err(AuthError::InvalidCredentials);
2268        }
2269        let id = UserId::from_parts(tenant_id, username);
2270        let users = self.users.read().map_err(lock_err)?;
2271        let (stored_hash, account_enabled) = match users.get(&id) {
2272            Some(user) => (user.password_hash.clone(), user.enabled),
2273            None => (dummy_hash.to_string(), false),
2274        };
2275        let password_matches = verify_password(password, &stored_hash);
2276        if !account_enabled || !password_matches {
2277            return Err(AuthError::InvalidCredentials);
2278        }
2279
2280        let user = users
2281            .get(&id)
2282            .expect("an enabled account came from the user map");
2283
2284        // Generate token: signed if keypair is available, random otherwise.
2285        let token = match self.keypair.read().ok().and_then(|g| {
2286            g.as_ref().map(|kp| {
2287                let token_id = random_hex(16);
2288                let sig = kp.sign(format!("session:{}", token_id).as_bytes());
2289                // Take first 16 bytes of signature for compact token.
2290                format!("rs_{}{}", token_id, hex::encode(&sig[..16]))
2291            })
2292        }) {
2293            Some(signed_token) => signed_token,
2294            None => generate_session_token(),
2295        };
2296
2297        let now = now_ms();
2298        let session = Session {
2299            token,
2300            username: username.to_string(),
2301            tenant_id: user.tenant_id.clone(),
2302            role: user.role,
2303            created_at: now,
2304            expires_at: now + (self.config.session_ttl_secs as u128 * 1000),
2305        };
2306
2307        drop(users); // release read lock before acquiring write
2308
2309        let mut sessions = self.sessions.write().map_err(lock_err)?;
2310        sessions.insert(session.token.clone(), session.clone());
2311        drop(sessions);
2312
2313        // Transparently upgrade a hash produced by the retired in-tree KDF.
2314        // This is the only point where the verified plaintext is in hand.
2315        if password_hash_needs_upgrade(&stored_hash) {
2316            let upgraded = hash_password(password);
2317            let mut rehashed = false;
2318            if let Ok(mut users) = self.users.write() {
2319                if let Some(user) = users.get_mut(&id) {
2320                    // Skip if a concurrent password change already moved it.
2321                    if user.password_hash == stored_hash {
2322                        user.password_hash = upgraded;
2323                        rehashed = true;
2324                    }
2325                }
2326            }
2327            if rehashed {
2328                self.persist_to_vault();
2329            }
2330        }
2331
2332        Ok(session)
2333    }
2334
2335    // -----------------------------------------------------------------
2336    // Token validation
2337    // -----------------------------------------------------------------
2338
2339    /// Validate a token (session or API key).
2340    ///
2341    /// Returns `(username, role)` if valid, `None` otherwise. Tenant
2342    /// scope is dropped here for compatibility with the bulk of the
2343    /// existing caller surface (routing, gRPC control, redwire). Use
2344    /// [`Self::validate_token_full`] when the caller needs the
2345    /// resolved `UserId` (e.g. to pin `CURRENT_TENANT()`).
2346    pub fn validate_token(&self, token: &str) -> Option<(String, Role)> {
2347        self.validate_token_full(token)
2348            .map(|(id, role)| (id.username, role))
2349    }
2350
2351    /// Tenant-aware token validation. Returns the resolved `UserId`
2352    /// (which carries the tenant) and the granted `Role`.
2353    pub fn validate_token_full(&self, token: &str) -> Option<(UserId, Role)> {
2354        let resolved = self.resolve_token_unchecked(token)?;
2355        // Every credential is re-checked against the live user record so
2356        // `ALTER USER … DISABLE` / `VALID UNTIL` take effect immediately for
2357        // sessions and API keys, not only at password login.
2358        if !self.principal_is_active(&resolved.0) {
2359            return None;
2360        }
2361        Some(resolved)
2362    }
2363
2364    /// Token → `(UserId, Role)` without consulting the owning user record.
2365    fn resolve_token_unchecked(&self, token: &str) -> Option<(UserId, Role)> {
2366        // Try session tokens first.
2367        if token.starts_with("rs_") {
2368            if let Ok(sessions) = self.sessions.read() {
2369                if let Some(session) = sessions.get(token) {
2370                    let now = now_ms();
2371                    if now < session.expires_at {
2372                        return Some((
2373                            UserId::from_parts(session.tenant_id.as_deref(), &session.username),
2374                            session.role,
2375                        ));
2376                    }
2377                }
2378            }
2379            return None;
2380        }
2381
2382        // Try API keys: the index is keyed by the credential id, so the
2383        // presented secret is hashed and never compared as text.
2384        if token.starts_with("rk_") {
2385            let id = api_key_credential_id(token);
2386            let (owner, role) = self
2387                .api_key_index
2388                .read()
2389                .ok()
2390                .and_then(|idx| idx.get(&id).cloned())?;
2391            if self.api_key_expired(&owner, &id) {
2392                return None;
2393            }
2394            return Some((owner, role));
2395        }
2396
2397        None
2398    }
2399
2400    /// Whether the key `id` owned by `owner` has passed its `expires_at`.
2401    /// A key the owner no longer lists counts as expired.
2402    fn api_key_expired(&self, owner: &UserId, id: &str) -> bool {
2403        let Ok(users) = self.users.read() else {
2404            return true;
2405        };
2406        users
2407            .get(owner)
2408            .and_then(|user| user.api_keys.iter().find(|key| key.key == id))
2409            .map(|key| key.expires_at.is_some_and(|deadline| now_ms() >= deadline))
2410            .unwrap_or(true)
2411    }
2412
2413    /// The current role of `id` when the account is active, else `None`.
2414    ///
2415    /// Callers holding a long-lived credential (the browser refresh cookie)
2416    /// use this to re-derive the principal's role at use time rather than
2417    /// trusting a role captured when the credential was minted.
2418    pub fn active_user_role(&self, id: &UserId) -> Option<Role> {
2419        if !self.principal_is_active(id) {
2420            return None;
2421        }
2422        self.users.read().ok()?.get(id).map(|user| user.role)
2423    }
2424
2425    /// Whether `id` names an enabled account whose `VALID UNTIL` (if any)
2426    /// has not passed. Unknown accounts are inactive.
2427    fn principal_is_active(&self, id: &UserId) -> bool {
2428        let enabled = match self.users.read() {
2429            Ok(users) => users.get(id).map(|u| u.enabled).unwrap_or(false),
2430            Err(_) => false,
2431        };
2432        if !enabled {
2433            return false;
2434        }
2435        match self.user_attributes(id).valid_until {
2436            Some(deadline) => now_ms() < deadline,
2437            None => true,
2438        }
2439    }
2440
2441    // -----------------------------------------------------------------
2442    // API Key management
2443    // -----------------------------------------------------------------
2444
2445    /// Create a persistent API key for a platform-tenant user.
2446    ///
2447    /// For tenant-scoped users use [`Self::create_api_key_in_tenant`].
2448    pub fn create_api_key(
2449        &self,
2450        username: &str,
2451        name: &str,
2452        role: Role,
2453    ) -> Result<ApiKey, AuthError> {
2454        self.create_api_key_in_tenant(None, username, name, role)
2455    }
2456
2457    pub fn create_api_key_with_control_events(
2458        &self,
2459        username: &str,
2460        name: &str,
2461        role: Role,
2462        ctx: &ControlEventCtx<'_>,
2463        ledger: &dyn ControlEventLedger,
2464        config: ControlEventConfig,
2465    ) -> Result<ApiKey, AuthError> {
2466        self.create_api_key_in_tenant_with_control_events(
2467            None, username, name, role, ctx, ledger, config,
2468        )
2469    }
2470
2471    pub fn create_api_key_in_tenant(
2472        &self,
2473        tenant_id: Option<&str>,
2474        username: &str,
2475        name: &str,
2476        role: Role,
2477    ) -> Result<ApiKey, AuthError> {
2478        self.create_api_key_in_tenant_expiring(tenant_id, username, name, role, None)
2479    }
2480
2481    /// Create an API key that stops validating at `expires_at` (ms since
2482    /// epoch). The returned value carries the secret; nothing else does.
2483    pub fn create_api_key_in_tenant_expiring(
2484        &self,
2485        tenant_id: Option<&str>,
2486        username: &str,
2487        name: &str,
2488        role: Role,
2489        expires_at: Option<u128>,
2490    ) -> Result<ApiKey, AuthError> {
2491        if let Some(configured) = self.configured_control_events() {
2492            let ctx = default_user_lifecycle_ctx();
2493            let control = UserLifecycleControl {
2494                ctx: &ctx,
2495                ledger: configured.ledger.as_ref(),
2496                config: configured.config,
2497            };
2498            self.create_api_key_in_tenant_controlled(
2499                tenant_id, username, name, role, expires_at, &control,
2500            )
2501        } else {
2502            self.create_api_key_in_tenant_unaudited(tenant_id, username, name, role, expires_at)
2503        }
2504    }
2505
2506    #[allow(clippy::too_many_arguments)]
2507    pub fn create_api_key_in_tenant_with_control_events(
2508        &self,
2509        tenant_id: Option<&str>,
2510        username: &str,
2511        name: &str,
2512        role: Role,
2513        ctx: &ControlEventCtx<'_>,
2514        ledger: &dyn ControlEventLedger,
2515        config: ControlEventConfig,
2516    ) -> Result<ApiKey, AuthError> {
2517        let control = UserLifecycleControl {
2518            ctx,
2519            ledger,
2520            config,
2521        };
2522        self.create_api_key_in_tenant_controlled(tenant_id, username, name, role, None, &control)
2523    }
2524
2525    fn create_api_key_in_tenant_unaudited(
2526        &self,
2527        tenant_id: Option<&str>,
2528        username: &str,
2529        name: &str,
2530        role: Role,
2531        expires_at: Option<u128>,
2532    ) -> Result<ApiKey, AuthError> {
2533        let id = UserId::from_parts(tenant_id, username);
2534        let mut users = self.users.write().map_err(lock_err)?;
2535        let user = users
2536            .get_mut(&id)
2537            .ok_or_else(|| AuthError::UserNotFound(id.to_string()))?;
2538
2539        // The key's role cannot exceed the user's role.
2540        if role > user.role {
2541            return Err(AuthError::RoleExceeded {
2542                requested: role,
2543                ceiling: user.role,
2544            });
2545        }
2546
2547        let secret = generate_api_key();
2548        let stored = ApiKey {
2549            key: api_key_credential_id(&secret),
2550            name: name.to_string(),
2551            role,
2552            created_at: now_ms(),
2553            expires_at,
2554        };
2555
2556        user.api_keys.push(stored.clone());
2557        user.updated_at = now_ms();
2558
2559        // Update the index.
2560        if let Ok(mut idx) = self.api_key_index.write() {
2561            idx.insert(stored.key.clone(), (id.clone(), stored.role));
2562        }
2563
2564        drop(users); // release lock before vault I/O
2565        self.persist_to_vault();
2566        // The secret leaves the store exactly once, here.
2567        Ok(ApiKey {
2568            key: secret,
2569            ..stored
2570        })
2571    }
2572
2573    fn create_api_key_in_tenant_controlled(
2574        &self,
2575        tenant_id: Option<&str>,
2576        username: &str,
2577        name: &str,
2578        role: Role,
2579        expires_at: Option<u128>,
2580        control: &UserLifecycleControl<'_>,
2581    ) -> Result<ApiKey, AuthError> {
2582        let id = UserId::from_parts(tenant_id, username);
2583        match self.create_api_key_in_tenant_unaudited(tenant_id, username, name, role, expires_at) {
2584            Ok(api_key) => {
2585                let key_id = api_key_id(&api_key.key);
2586                if let Err(err) = self.emit_api_key_allowed(
2587                    control,
2588                    EventKind::ApiKeyCreate,
2589                    "apikey.create",
2590                    &key_id,
2591                    api_key_control_fields(&id, role, &key_id),
2592                ) {
2593                    self.rollback_create_api_key(&id, &api_key.key);
2594                    return Err(err);
2595                }
2596                Ok(api_key)
2597            }
2598            Err(err) => {
2599                self.emit_user_lifecycle_outcome(
2600                    control,
2601                    if user_error_is_denied(&err) {
2602                        Outcome::Denied
2603                    } else {
2604                        Outcome::Error
2605                    },
2606                    EventKind::ApiKeyCreate,
2607                    "apikey.create",
2608                    &id,
2609                    Some(err.to_string()),
2610                    user_control_fields(&id, Some(role), None, false),
2611                );
2612                Err(err)
2613            }
2614        }
2615    }
2616
2617    /// Resolve the tenant-aware owner of an API key without exposing its record.
2618    pub fn api_key_owner(&self, key: &str) -> Option<UserId> {
2619        let key_id = api_key_credential_id(key);
2620        self.api_key_index
2621            .read()
2622            .ok()
2623            .and_then(|index| index.get(&key_id).map(|(id, _)| id.clone()))
2624    }
2625
2626    /// Revoke (delete) an API key.
2627    pub fn revoke_api_key(&self, key: &str) -> Result<(), AuthError> {
2628        if let Some(configured) = self.configured_control_events() {
2629            let ctx = default_user_lifecycle_ctx();
2630            let control = UserLifecycleControl {
2631                ctx: &ctx,
2632                ledger: configured.ledger.as_ref(),
2633                config: configured.config,
2634            };
2635            self.revoke_api_key_controlled(key, &control)
2636        } else {
2637            self.revoke_api_key_unaudited(key)
2638        }
2639    }
2640
2641    pub fn revoke_api_key_with_control_events(
2642        &self,
2643        key: &str,
2644        ctx: &ControlEventCtx<'_>,
2645        ledger: &dyn ControlEventLedger,
2646        config: ControlEventConfig,
2647    ) -> Result<(), AuthError> {
2648        let control = UserLifecycleControl {
2649            ctx,
2650            ledger,
2651            config,
2652        };
2653        self.revoke_api_key_controlled(key, &control)
2654    }
2655
2656    fn revoke_api_key_unaudited(&self, key: &str) -> Result<(), AuthError> {
2657        let key_id = api_key_credential_id(key);
2658        let key = key_id.as_str();
2659        let mut users = self.users.write().map_err(lock_err)?;
2660
2661        // Find which user owns this key (look up by the api_key_index
2662        // first; fall back to a scan for legacy state restored before
2663        // the index was reseeded).
2664        let owner_id: UserId = {
2665            if let Ok(idx) = self.api_key_index.read() {
2666                if let Some((id, _)) = idx.get(key) {
2667                    id.clone()
2668                } else {
2669                    return Err(AuthError::KeyNotFound(key.to_string()));
2670                }
2671            } else {
2672                let owner = users
2673                    .iter()
2674                    .find(|(_, u)| u.api_keys.iter().any(|k| k.key == key));
2675                match owner {
2676                    Some((id, _)) => id.clone(),
2677                    None => return Err(AuthError::KeyNotFound(key.to_string())),
2678                }
2679            }
2680        };
2681
2682        let user = users
2683            .get_mut(&owner_id)
2684            .ok_or_else(|| AuthError::KeyNotFound(key.to_string()))?;
2685        user.api_keys.retain(|k| k.key != key);
2686        user.updated_at = now_ms();
2687
2688        // Remove from index.
2689        if let Ok(mut idx) = self.api_key_index.write() {
2690            idx.remove(key);
2691        }
2692
2693        self.persist_to_vault();
2694        Ok(())
2695    }
2696
2697    fn revoke_api_key_controlled(
2698        &self,
2699        key: &str,
2700        control: &UserLifecycleControl<'_>,
2701    ) -> Result<(), AuthError> {
2702        let key_id = api_key_id(key);
2703        let rollback = self.api_key_rollback_snapshot(key);
2704        let id = rollback
2705            .as_ref()
2706            .map(|r| r.0.clone())
2707            .unwrap_or_else(|| UserId::from_parts(None, ""));
2708        let role = rollback.as_ref().map(|r| r.1.role).unwrap_or(Role::Read);
2709        match self.revoke_api_key_unaudited(key) {
2710            Ok(()) => {
2711                if let Err(err) = self.emit_api_key_allowed(
2712                    control,
2713                    EventKind::ApiKeyRevoke,
2714                    "apikey.revoke",
2715                    &key_id,
2716                    api_key_control_fields(&id, role, &key_id),
2717                ) {
2718                    if let Some((owner, api_key)) = rollback {
2719                        self.restore_api_key(&owner, api_key);
2720                    }
2721                    return Err(err);
2722                }
2723                Ok(())
2724            }
2725            Err(err) => {
2726                self.emit_user_lifecycle_outcome(
2727                    control,
2728                    if user_error_is_denied(&err) {
2729                        Outcome::Denied
2730                    } else {
2731                        Outcome::Error
2732                    },
2733                    EventKind::ApiKeyRevoke,
2734                    "apikey.revoke",
2735                    &id,
2736                    Some(err.to_string()),
2737                    api_key_control_fields(&id, role, &key_id),
2738                );
2739                Err(err)
2740            }
2741        }
2742    }
2743
2744    // -----------------------------------------------------------------
2745    // Session management
2746    // -----------------------------------------------------------------
2747
2748    /// Revoke a session token.
2749    pub fn revoke_session(&self, token: &str) {
2750        if let Some(session) = self.remove_session_without_count(token) {
2751            if self.take_counted_session(token) {
2752                let uid = UserId::from_parts(session.tenant_id.as_deref(), &session.username);
2753                self.decrement_session_count(&uid);
2754            }
2755        }
2756    }
2757
2758    fn remove_session_without_count(&self, token: &str) -> Option<Session> {
2759        self.sessions.write().ok()?.remove(token)
2760    }
2761
2762    fn take_counted_session(&self, token: &str) -> bool {
2763        self.counted_session_tokens
2764            .write()
2765            .map(|mut tokens| tokens.remove(token))
2766            .unwrap_or(false)
2767    }
2768
2769    /// Purge expired sessions (housekeeping).
2770    pub fn purge_expired_sessions(&self) -> usize {
2771        let now = now_ms();
2772        if let Ok(mut sessions) = self.sessions.write() {
2773            let expired: Vec<Session> = sessions
2774                .values()
2775                .filter(|session| session.expires_at <= now)
2776                .cloned()
2777                .collect();
2778            sessions.retain(|_, session| session.expires_at > now);
2779            drop(sessions);
2780            for session in &expired {
2781                if self.take_counted_session(&session.token) {
2782                    let uid = UserId::from_parts(session.tenant_id.as_deref(), &session.username);
2783                    self.decrement_session_count(&uid);
2784                }
2785            }
2786            return expired.len();
2787        }
2788        0
2789    }
2790
2791    // -----------------------------------------------------------------
2792    // Granular RBAC — GRANT / REVOKE
2793    //
2794    // The privilege engine lives in `super::privileges`. These helpers
2795    // are the AuthStore facade: they keep an in-memory map of grants per
2796    // user (plus a `public_grants` list), persist additions/removals to
2797    // the existing `vault_kv` store, and rebuild the per-user
2798    // `PermissionCache` so the hot path stays O(1).
2799    //
2800    // Persistence design: rather than extend the snapshot/restore
2801    // pipeline (Agent #2's territory) we serialise grants and account
2802    // attributes to the vault KV store. That gives us atomic write +
2803    // encrypted-at-rest semantics for free without touching the
2804    // existing USER/KEY/KV serializer paths. On restart `rehydrate_acl`
2805    // reads these KV entries back into the in-memory maps.
2806    // -----------------------------------------------------------------
2807
2808    /// Persist a grant. Returns `Forbidden` when the granting user is
2809    /// not Admin or attempts a cross-tenant grant.
2810    pub fn grant(
2811        &self,
2812        granter: &UserId,
2813        granter_role: Role,
2814        principal: GrantPrincipal,
2815        resource: Resource,
2816        actions: Vec<Action>,
2817        with_grant_option: bool,
2818        tenant: Option<String>,
2819    ) -> Result<(), AuthError> {
2820        if granter_role != Role::Admin {
2821            return Err(AuthError::Forbidden(format!(
2822                "GRANT requires Admin role; granter `{}` has `{:?}`",
2823                granter, granter_role
2824            )));
2825        }
2826
2827        // Cross-tenant guard: a tenant-scoped admin cannot mint grants
2828        // outside their tenant. Platform admin (tenant=None) may grant
2829        // anywhere.
2830        if granter.tenant.is_some() && granter.tenant != tenant {
2831            return Err(AuthError::Forbidden(format!(
2832                "cross-tenant GRANT denied: granter tenant `{:?}` != grant tenant `{:?}`",
2833                granter.tenant, tenant
2834            )));
2835        }
2836
2837        let mut actions_set = std::collections::BTreeSet::new();
2838        for a in actions {
2839            actions_set.insert(a);
2840        }
2841        let g = Grant {
2842            principal: principal.clone(),
2843            resource,
2844            actions: actions_set,
2845            with_grant_option,
2846            granted_by: granter.to_string(),
2847            granted_at: now_ms(),
2848            tenant,
2849            columns: None,
2850        };
2851
2852        match &principal {
2853            GrantPrincipal::User(uid) => {
2854                self.grants
2855                    .write()
2856                    .unwrap_or_else(|e| e.into_inner())
2857                    .entry(uid.clone())
2858                    .or_default()
2859                    .push(g.clone());
2860                self.invalidate_permission_cache(Some(uid));
2861            }
2862            GrantPrincipal::Public => {
2863                self.public_grants
2864                    .write()
2865                    .unwrap_or_else(|e| e.into_inner())
2866                    .push(g.clone());
2867                self.invalidate_permission_cache(None);
2868            }
2869            GrantPrincipal::Group(_) => {
2870                return Err(AuthError::Forbidden(
2871                    "GROUP principals are not yet supported; use a USER or PUBLIC".to_string(),
2872                ));
2873            }
2874        }
2875
2876        // Issue #119: a fresh grant changes the visible-collections set
2877        // for `(tenant, role)` callers under the same tenant. Drop those
2878        // cache entries so the next AI command sees the new SELECT
2879        // privilege immediately.
2880        self.invalidate_visible_collections_for_tenant(g.tenant.as_deref());
2881
2882        self.persist_acl_to_kv();
2883        Ok(())
2884    }
2885
2886    /// Drop matching grants from a principal. Returns the number of
2887    /// grants removed.
2888    pub fn revoke(
2889        &self,
2890        granter: &UserId,
2891        granter_role: Role,
2892        principal: &GrantPrincipal,
2893        resource: &Resource,
2894        actions: &[Action],
2895    ) -> Result<usize, AuthError> {
2896        if granter_role != Role::Admin {
2897            return Err(AuthError::Forbidden(format!(
2898                "REVOKE requires Admin role; granter `{}` has `{:?}`",
2899                granter, granter_role
2900            )));
2901        }
2902
2903        // Cross-tenant guard, mirroring GRANT: a tenant-scoped admin can only
2904        // touch grants of principals in its own tenant. The target tenant
2905        // comes from the statement text, so without this a tenant admin could
2906        // strip another tenant's grants. Platform admin (tenant=None) may
2907        // revoke anywhere.
2908        if let GrantPrincipal::User(uid) = principal {
2909            if granter.tenant.is_some() && granter.tenant != uid.tenant {
2910                return Err(AuthError::Forbidden(format!(
2911                    "cross-tenant REVOKE denied: granter tenant `{:?}` != principal tenant `{:?}`",
2912                    granter.tenant, uid.tenant
2913                )));
2914            }
2915        }
2916
2917        let removed = match principal {
2918            GrantPrincipal::User(uid) => {
2919                let mut g = self.grants.write().unwrap_or_else(|e| e.into_inner());
2920                let before = g.get(uid).map(|v| v.len()).unwrap_or(0);
2921                if let Some(list) = g.get_mut(uid) {
2922                    list.retain(|gr| {
2923                        !(gr.resource == *resource
2924                            && (actions.iter().any(|a| gr.actions.contains(a))
2925                                || (gr.actions.contains(&Action::All) && !actions.is_empty())))
2926                    });
2927                }
2928                let after = g.get(uid).map(|v| v.len()).unwrap_or(0);
2929                drop(g);
2930                self.invalidate_permission_cache(Some(uid));
2931                before - after
2932            }
2933            GrantPrincipal::Public => {
2934                let mut p = self
2935                    .public_grants
2936                    .write()
2937                    .unwrap_or_else(|e| e.into_inner());
2938                let before = p.len();
2939                p.retain(|gr| {
2940                    !(gr.resource == *resource
2941                        && (actions.iter().any(|a| gr.actions.contains(a))
2942                            || (gr.actions.contains(&Action::All) && !actions.is_empty())))
2943                });
2944                let after = p.len();
2945                drop(p);
2946                self.invalidate_permission_cache(None);
2947                before - after
2948            }
2949            GrantPrincipal::Group(_) => 0,
2950        };
2951
2952        if removed > 0 {
2953            // Issue #119: REVOKE may shrink the visible-collections set
2954            // for any `(tenant, role)` slot. We don't know the exact
2955            // tenant when the principal is `Public`, so a `Public`
2956            // revoke clears the whole cache; user revokes scope to the
2957            // user's tenant.
2958            match principal {
2959                GrantPrincipal::User(uid) => {
2960                    self.invalidate_visible_collections_for_tenant(uid.tenant.as_deref());
2961                }
2962                GrantPrincipal::Public | GrantPrincipal::Group(_) => {
2963                    self.invalidate_visible_collections_cache();
2964                }
2965            }
2966            self.persist_acl_to_kv();
2967        }
2968        Ok(removed)
2969    }
2970
2971    /// Compute the set of collection ids a given `(tenant, role)`
2972    /// scope can read, consulting the explicit grant table. The result
2973    /// is cached for `super::scope_cache::DEFAULT_TTL` (60s) and
2974    /// invalidated on every GRANT/REVOKE/policy/collection mutation
2975    /// that could change the answer.
2976    ///
2977    /// `all_collections` is the full list of collection ids known to
2978    /// the storage layer. The runtime hands it in so this module stays
2979    /// decoupled from the storage crate. Each collection passes through
2980    /// `check_grant(SELECT)` under a synthetic `(principal, role,
2981    /// tenant)` view. The cache key includes principal because direct
2982    /// grants can differ between users that share the same tenant and
2983    /// role.
2984    pub fn visible_collections_for_scope(
2985        &self,
2986        tenant: Option<&str>,
2987        role: Role,
2988        principal: &str,
2989        all_collections: &[String],
2990    ) -> std::collections::HashSet<String> {
2991        let key = super::scope_cache::ScopeKey::new(tenant, principal, role);
2992        if let Some(hit) = self.visible_collections_cache.get(&key) {
2993            return hit;
2994        }
2995        // Slow path: walk every collection through `check_grant`. We
2996        // build the AuthzContext once, then reuse it per resource.
2997        let ctx = AuthzContext {
2998            principal,
2999            effective_role: role,
3000            tenant,
3001        };
3002        let mut visible = std::collections::HashSet::new();
3003        for collection in all_collections {
3004            let resource = Resource::table_from_name(collection);
3005            if self.check_grant(&ctx, Action::Select, &resource).is_ok() {
3006                visible.insert(collection.clone());
3007            }
3008        }
3009        self.visible_collections_cache.insert(key, visible.clone());
3010        visible
3011    }
3012
3013    /// Stats probe required by issue #119 — exposes hit/miss counts and
3014    /// invalidations for the visible-collections cache so metrics
3015    /// pipelines can compute a hit rate.
3016    pub fn auth_cache_stats(&self) -> super::scope_cache::AuthCacheStats {
3017        self.visible_collections_cache.stats()
3018    }
3019
3020    /// Drop every cached `(tenant, role)` entry. Called from CREATE
3021    /// POLICY / DROP POLICY / DROP COLLECTION paths where the affected
3022    /// tenant set is unknown.
3023    pub fn invalidate_visible_collections_cache(&self) {
3024        self.visible_collections_cache.invalidate_all();
3025    }
3026
3027    /// Drop cached entries for one tenant. Called from GRANT / REVOKE
3028    /// where the principal's tenant is known.
3029    pub fn invalidate_visible_collections_for_tenant(&self, tenant: Option<&str>) {
3030        self.visible_collections_cache.invalidate_tenant(tenant);
3031    }
3032
3033    /// Snapshot of every grant the principal effectively has, including
3034    /// `Public` grants. Audit / introspection helper.
3035    pub fn effective_grants(&self, uid: &UserId) -> Vec<Grant> {
3036        let mut out = Vec::new();
3037        if let Ok(g) = self.grants.read() {
3038            if let Some(list) = g.get(uid) {
3039                out.extend(list.iter().cloned());
3040            }
3041        }
3042        if let Ok(p) = self.public_grants.read() {
3043            out.extend(p.iter().cloned());
3044        }
3045        out
3046    }
3047
3048    /// Run a privilege check using the in-memory grant tables. Returns
3049    /// `Ok(())` on allow, `Err(AuthzError)` on deny.
3050    pub fn check_grant(
3051        &self,
3052        ctx: &AuthzContext<'_>,
3053        action: Action,
3054        resource: &Resource,
3055    ) -> Result<(), AuthzError> {
3056        if ctx.effective_role == Role::Admin {
3057            return Ok(());
3058        }
3059
3060        let uid = UserId::from_parts(ctx.tenant, ctx.principal);
3061
3062        // Fast path: per-user pre-resolved cache.
3063        if let Ok(cache) = self.permission_cache.read() {
3064            if let Some(pc) = cache.get(&uid) {
3065                if pc.allows(resource, action) {
3066                    return Ok(());
3067                }
3068            }
3069        }
3070
3071        // Slow path: linear scan + rebuild cache as a side-effect.
3072        let user_grants = self
3073            .grants
3074            .read()
3075            .ok()
3076            .and_then(|g| g.get(&uid).cloned())
3077            .unwrap_or_default();
3078        let any_user_grants = self
3079            .grants
3080            .read()
3081            .ok()
3082            .map(|g| g.values().any(|list| !list.is_empty()))
3083            .unwrap_or(false);
3084        let public_grants = self
3085            .public_grants
3086            .read()
3087            .ok()
3088            .map(|p| p.clone())
3089            .unwrap_or_default();
3090        if user_grants.is_empty() && public_grants.is_empty() && any_user_grants {
3091            return Err(AuthzError::PermissionDenied {
3092                action,
3093                resource: resource.clone(),
3094                principal: ctx.principal.to_string(),
3095            });
3096        }
3097        let view = GrantsView {
3098            user_grants: &user_grants,
3099            public_grants: &public_grants,
3100        };
3101        let result = check_grant(ctx, action, resource, &view);
3102
3103        if result.is_ok() {
3104            let pc = PermissionCache::build(&user_grants, &public_grants);
3105            if let Ok(mut cache) = self.permission_cache.write() {
3106                cache.insert(uid, pc);
3107            }
3108        }
3109        result
3110    }
3111
3112    // -----------------------------------------------------------------
3113    // ALTER USER attributes (VALID UNTIL, CONNECTION LIMIT, etc.)
3114    // -----------------------------------------------------------------
3115
3116    /// Replace the attribute record for `uid`.
3117    pub fn set_user_attributes(
3118        &self,
3119        uid: &UserId,
3120        attrs: UserAttributes,
3121    ) -> Result<(), AuthError> {
3122        let users = self.users.read().map_err(lock_err)?;
3123        if !users.contains_key(uid) {
3124            return Err(AuthError::UserNotFound(uid.to_string()));
3125        }
3126        drop(users);
3127
3128        self.user_attributes
3129            .write()
3130            .unwrap_or_else(|e| e.into_inner())
3131            .insert(uid.clone(), attrs);
3132        self.invalidate_iam_cache(Some(uid));
3133        self.persist_acl_to_kv();
3134        Ok(())
3135    }
3136
3137    /// Read the attributes for `uid`. Returns `Default::default()` for
3138    /// users that have never been altered.
3139    pub fn user_attributes(&self, uid: &UserId) -> UserAttributes {
3140        self.user_attributes
3141            .read()
3142            .ok()
3143            .and_then(|m| m.get(uid).cloned())
3144            .unwrap_or_default()
3145    }
3146
3147    pub fn add_user_to_group(&self, uid: &UserId, group: &str) -> Result<(), AuthError> {
3148        if group.trim().is_empty() {
3149            return Err(AuthError::Forbidden("group name cannot be empty".into()));
3150        }
3151        let mut attrs = self.user_attributes(uid);
3152        if !attrs.groups.iter().any(|g| g == group) {
3153            attrs.groups.push(group.to_string());
3154            attrs.groups.sort();
3155        }
3156        self.set_user_attributes(uid, attrs)
3157    }
3158
3159    pub fn remove_user_from_group(&self, uid: &UserId, group: &str) -> Result<(), AuthError> {
3160        let mut attrs = self.user_attributes(uid);
3161        attrs.groups.retain(|g| g != group);
3162        self.set_user_attributes(uid, attrs)
3163    }
3164
3165    /// Toggle `User.enabled` without rotating credentials.
3166    pub fn set_user_enabled(&self, uid: &UserId, enabled: bool) -> Result<(), AuthError> {
3167        if let Some(configured) = self.configured_control_events() {
3168            let ctx = default_user_lifecycle_ctx();
3169            let control = UserLifecycleControl {
3170                ctx: &ctx,
3171                ledger: configured.ledger.as_ref(),
3172                config: configured.config,
3173            };
3174            self.set_user_enabled_controlled(uid, enabled, &control)
3175        } else {
3176            self.set_user_enabled_unaudited(uid, enabled)
3177        }
3178    }
3179
3180    pub fn disable_user(
3181        &self,
3182        username: &str,
3183        ctx: &ControlEventCtx<'_>,
3184        ledger: &dyn ControlEventLedger,
3185        config: ControlEventConfig,
3186    ) -> Result<(), AuthError> {
3187        self.disable_user_in_tenant(None, username, ctx, ledger, config)
3188    }
3189
3190    pub fn disable_user_in_tenant(
3191        &self,
3192        tenant_id: Option<&str>,
3193        username: &str,
3194        ctx: &ControlEventCtx<'_>,
3195        ledger: &dyn ControlEventLedger,
3196        config: ControlEventConfig,
3197    ) -> Result<(), AuthError> {
3198        let uid = UserId::from_parts(tenant_id, username);
3199        let control = UserLifecycleControl {
3200            ctx,
3201            ledger,
3202            config,
3203        };
3204        self.set_user_enabled_controlled(&uid, false, &control)
3205    }
3206
3207    fn set_user_enabled_unaudited(&self, uid: &UserId, enabled: bool) -> Result<(), AuthError> {
3208        let mut users = self.users.write().map_err(lock_err)?;
3209        let user = users
3210            .get_mut(uid)
3211            .ok_or_else(|| AuthError::UserNotFound(uid.to_string()))?;
3212        user.enabled = enabled;
3213        user.updated_at = now_ms();
3214        drop(users);
3215        if !enabled {
3216            self.purge_sessions_for(uid);
3217        }
3218        self.persist_to_vault();
3219        Ok(())
3220    }
3221
3222    /// Drop every live session belonging to `uid` (tenant + username match
3223    /// so a same-named user in another tenant is untouched).
3224    fn purge_sessions_for(&self, uid: &UserId) {
3225        if let Ok(mut sessions) = self.sessions.write() {
3226            let tokens: Vec<String> = sessions
3227                .iter()
3228                .filter(|(_, session)| {
3229                    session.username == uid.username && session.tenant_id == uid.tenant
3230                })
3231                .map(|(token, _)| token.clone())
3232                .collect();
3233            for token in &tokens {
3234                sessions.remove(token);
3235            }
3236            drop(sessions);
3237            for token in tokens {
3238                if self.take_counted_session(&token) {
3239                    self.decrement_session_count(uid);
3240                }
3241            }
3242        }
3243    }
3244
3245    fn set_user_enabled_controlled(
3246        &self,
3247        uid: &UserId,
3248        enabled: bool,
3249        control: &UserLifecycleControl<'_>,
3250    ) -> Result<(), AuthError> {
3251        let previous = self.get_user_cloned(uid);
3252        let kind = if enabled {
3253            EventKind::UserUpdate
3254        } else {
3255            EventKind::UserDisable
3256        };
3257        let action = if enabled {
3258            "user.update"
3259        } else {
3260            "user.disable"
3261        };
3262        match self.set_user_enabled_unaudited(uid, enabled) {
3263            Ok(()) => {
3264                if let Err(err) = self.emit_user_lifecycle_allowed(
3265                    control,
3266                    kind,
3267                    action,
3268                    uid,
3269                    user_control_fields(
3270                        uid,
3271                        previous.as_ref().map(|u| u.role),
3272                        Some(enabled),
3273                        false,
3274                    ),
3275                ) {
3276                    self.restore_user_snapshot(uid, previous);
3277                    return Err(err);
3278                }
3279                Ok(())
3280            }
3281            Err(err) => {
3282                self.emit_user_lifecycle_outcome(
3283                    control,
3284                    if user_error_is_denied(&err) {
3285                        Outcome::Denied
3286                    } else {
3287                        Outcome::Error
3288                    },
3289                    kind,
3290                    action,
3291                    uid,
3292                    Some(err.to_string()),
3293                    user_control_fields(
3294                        uid,
3295                        previous.as_ref().map(|u| u.role),
3296                        Some(enabled),
3297                        false,
3298                    ),
3299                );
3300                Err(err)
3301            }
3302        }
3303    }
3304
3305    fn emit_user_lifecycle_allowed(
3306        &self,
3307        control: &UserLifecycleControl<'_>,
3308        kind: EventKind,
3309        action: &'static str,
3310        id: &UserId,
3311        fields: HashMap<String, Sensitivity>,
3312    ) -> Result<(), AuthError> {
3313        self.emit_user_lifecycle_event(control, Outcome::Allowed, kind, action, id, None, fields)
3314    }
3315
3316    fn emit_api_key_allowed(
3317        &self,
3318        control: &UserLifecycleControl<'_>,
3319        kind: EventKind,
3320        action: &'static str,
3321        api_key_id: &str,
3322        fields: HashMap<String, Sensitivity>,
3323    ) -> Result<(), AuthError> {
3324        self.emit_control_event(
3325            control,
3326            Outcome::Allowed,
3327            kind,
3328            action,
3329            Some(api_key_resource(api_key_id)),
3330            None,
3331            fields,
3332        )
3333    }
3334
3335    #[allow(clippy::too_many_arguments)]
3336    fn emit_user_lifecycle_outcome(
3337        &self,
3338        control: &UserLifecycleControl<'_>,
3339        outcome: Outcome,
3340        kind: EventKind,
3341        action: &'static str,
3342        id: &UserId,
3343        reason: Option<String>,
3344        fields: HashMap<String, Sensitivity>,
3345    ) {
3346        let _ = self.emit_user_lifecycle_event(control, outcome, kind, action, id, reason, fields);
3347    }
3348
3349    #[allow(clippy::too_many_arguments)]
3350    fn emit_user_lifecycle_event(
3351        &self,
3352        control: &UserLifecycleControl<'_>,
3353        outcome: Outcome,
3354        kind: EventKind,
3355        action: &'static str,
3356        id: &UserId,
3357        reason: Option<String>,
3358        fields: HashMap<String, Sensitivity>,
3359    ) -> Result<(), AuthError> {
3360        self.emit_control_event(
3361            control,
3362            outcome,
3363            kind,
3364            action,
3365            Some(user_resource(id)),
3366            reason,
3367            fields,
3368        )
3369    }
3370
3371    #[allow(clippy::too_many_arguments)]
3372    fn emit_control_event(
3373        &self,
3374        control: &UserLifecycleControl<'_>,
3375        outcome: Outcome,
3376        kind: EventKind,
3377        action: &'static str,
3378        resource: Option<String>,
3379        reason: Option<String>,
3380        fields: HashMap<String, Sensitivity>,
3381    ) -> Result<(), AuthError> {
3382        let event = ControlEvent {
3383            kind,
3384            outcome,
3385            action: Cow::Borrowed(action),
3386            resource,
3387            reason,
3388            matched_policy_id: None,
3389            fields,
3390        };
3391        match control.ledger.emit(control.ctx, event) {
3392            Ok(_) => Ok(()),
3393            Err(err) if control.config.require_persistence() => {
3394                Err(AuthError::Internal(err.to_string()))
3395            }
3396            Err(_) => Ok(()),
3397        }
3398    }
3399
3400    fn rollback_create_user(&self, id: &UserId) {
3401        let removed = self
3402            .users
3403            .write()
3404            .unwrap_or_else(|e| e.into_inner())
3405            .remove(id);
3406        if let Some(user) = removed {
3407            self.remove_api_key_index_entries(&user);
3408        }
3409        self.persist_to_vault();
3410    }
3411
3412    fn rollback_bootstrap(&self, id: &UserId) {
3413        self.bootstrapped.store(false, Ordering::Release);
3414        self.rollback_create_user(id);
3415    }
3416
3417    fn restore_user_snapshot(&self, id: &UserId, previous: Option<User>) {
3418        match previous {
3419            Some(user) => {
3420                self.users
3421                    .write()
3422                    .unwrap_or_else(|e| e.into_inner())
3423                    .insert(id.clone(), user.clone());
3424                self.index_user_api_keys(id, &user);
3425            }
3426            None => {
3427                self.rollback_create_user(id);
3428                return;
3429            }
3430        }
3431        self.persist_to_vault();
3432    }
3433
3434    fn user_delete_rollback_snapshot(
3435        &self,
3436        id: &UserId,
3437        tenant_id: Option<&str>,
3438        username: &str,
3439    ) -> Option<(User, Vec<(String, Session)>)> {
3440        let user = self.get_user_cloned(id)?;
3441        let sessions = self
3442            .sessions
3443            .read()
3444            .map(|sessions| {
3445                sessions
3446                    .iter()
3447                    .filter(|(_, session)| {
3448                        session.username == username && session.tenant_id.as_deref() == tenant_id
3449                    })
3450                    .map(|(token, session)| (token.clone(), session.clone()))
3451                    .collect()
3452            })
3453            .unwrap_or_default();
3454        Some((user, sessions))
3455    }
3456
3457    fn restore_deleted_user(&self, id: &UserId, user: User, sessions: Vec<(String, Session)>) {
3458        self.users
3459            .write()
3460            .unwrap_or_else(|e| e.into_inner())
3461            .insert(id.clone(), user.clone());
3462        self.index_user_api_keys(id, &user);
3463        if !sessions.is_empty() {
3464            let mut guard = self.sessions.write().unwrap_or_else(|e| e.into_inner());
3465            for (token, session) in sessions {
3466                guard.insert(token, session);
3467            }
3468        }
3469        self.persist_to_vault();
3470    }
3471
3472    fn rollback_create_api_key(&self, id: &UserId, key: &str) {
3473        let key_id = api_key_credential_id(key);
3474        let key = key_id.as_str();
3475        if let Ok(mut users) = self.users.write() {
3476            if let Some(user) = users.get_mut(id) {
3477                user.api_keys.retain(|api_key| api_key.key != key);
3478                user.updated_at = now_ms();
3479            }
3480        }
3481        if let Ok(mut idx) = self.api_key_index.write() {
3482            idx.remove(key);
3483        }
3484        self.persist_to_vault();
3485    }
3486
3487    fn api_key_rollback_snapshot(&self, key: &str) -> Option<(UserId, ApiKey)> {
3488        let key = api_key_credential_id(key);
3489        let users = self.users.read().ok()?;
3490        users.iter().find_map(|(id, user)| {
3491            user.api_keys
3492                .iter()
3493                .find(|api_key| api_key.key == key)
3494                .cloned()
3495                .map(|api_key| (id.clone(), api_key))
3496        })
3497    }
3498
3499    fn restore_api_key(&self, id: &UserId, api_key: ApiKey) {
3500        if let Ok(mut users) = self.users.write() {
3501            if let Some(user) = users.get_mut(id) {
3502                if !user
3503                    .api_keys
3504                    .iter()
3505                    .any(|candidate| candidate.key == api_key.key)
3506                {
3507                    user.api_keys.push(api_key.clone());
3508                    user.updated_at = now_ms();
3509                }
3510            }
3511        }
3512        if let Ok(mut idx) = self.api_key_index.write() {
3513            idx.insert(api_key.key.clone(), (id.clone(), api_key.role));
3514        }
3515        self.persist_to_vault();
3516    }
3517
3518    fn remove_api_key_index_entries(&self, user: &User) {
3519        if let Ok(mut idx) = self.api_key_index.write() {
3520            for api_key in &user.api_keys {
3521                idx.remove(&api_key.key);
3522            }
3523        }
3524    }
3525
3526    fn index_user_api_keys(&self, id: &UserId, user: &User) {
3527        if let Ok(mut idx) = self.api_key_index.write() {
3528            for api_key in &user.api_keys {
3529                idx.insert(api_key.key.clone(), (id.clone(), api_key.role));
3530            }
3531        }
3532    }
3533
3534    // -----------------------------------------------------------------
3535    // Login-side enforcement (HTTP path)
3536    // -----------------------------------------------------------------
3537
3538    /// Authenticate with VALID UNTIL / CONNECTION LIMIT enforcement.
3539    /// Wraps `authenticate_in_tenant` and additionally:
3540    ///   * rejects logins after `valid_until`,
3541    ///   * rejects logins when the live session count would exceed the
3542    ///     `connection_limit` attribute.
3543    pub fn authenticate_with_attrs(
3544        &self,
3545        tenant_id: Option<&str>,
3546        username: &str,
3547        password: &str,
3548    ) -> Result<Session, AuthError> {
3549        let uid = UserId::from_parts(tenant_id, username);
3550        let session = self.authenticate_active(tenant_id, username, password)?;
3551        if let Err(error) = self.reserve_connection_slot(&uid) {
3552            self.remove_session_without_count(&session.token);
3553            return Err(error);
3554        }
3555        let mut tokens = self.counted_session_tokens.write().map_err(|_| {
3556            self.remove_session_without_count(&session.token);
3557            self.decrement_session_count(&uid);
3558            AuthError::Internal("session admission lock poisoned".to_string())
3559        })?;
3560        tokens.insert(session.token.clone());
3561        Ok(session)
3562    }
3563
3564    /// Authenticate a stateless/token login while enforcing account expiry.
3565    /// CONNECTION LIMIT is reserved for live wire leases, not for merely
3566    /// issuing a reusable bearer token.
3567    pub fn authenticate_active(
3568        &self,
3569        tenant_id: Option<&str>,
3570        username: &str,
3571        password: &str,
3572    ) -> Result<Session, AuthError> {
3573        let uid = UserId::from_parts(tenant_id, username);
3574        if let Some(deadline) = self.user_attributes(&uid).valid_until {
3575            if now_ms() >= deadline {
3576                return Err(AuthError::Forbidden(format!(
3577                    "account `{uid}` expired (VALID UNTIL exceeded)"
3578                )));
3579            }
3580        }
3581        self.authenticate_in_tenant(tenant_id, username, password)
3582    }
3583
3584    /// Authenticate a wire connection and hold its admission slot until the
3585    /// returned lease is dropped at EOF/Bye/error.
3586    pub fn authenticate_connection(
3587        self: &Arc<Self>,
3588        tenant_id: Option<&str>,
3589        username: &str,
3590        password: &str,
3591    ) -> Result<(Session, ConnectionLease), AuthError> {
3592        let session = self.authenticate_active(tenant_id, username, password)?;
3593        let principal = UserId::from_parts(tenant_id, username);
3594        if let Err(error) = self.reserve_connection_slot(&principal) {
3595            self.remove_session_without_count(&session.token);
3596            return Err(error);
3597        }
3598        // PG password authentication needs the verified identity but does not
3599        // expose a bearer token. Remove the transient login session so its TTL
3600        // and revocation lifecycle cannot release the socket-owned lease.
3601        self.remove_session_without_count(&session.token);
3602        let lease = ConnectionLease {
3603            store: Arc::clone(self),
3604            principal,
3605        };
3606        Ok((session, lease))
3607    }
3608
3609    /// Reserve a slot after a non-password proof (SCRAM, API key, session,
3610    /// or a locally mapped OAuth principal) has already been validated.
3611    pub fn acquire_connection_lease(
3612        self: &Arc<Self>,
3613        principal: &UserId,
3614    ) -> Result<ConnectionLease, AuthError> {
3615        if !self.principal_is_active(principal) {
3616            return Err(AuthError::InvalidCredentials);
3617        }
3618        self.reserve_connection_slot(principal)?;
3619        Ok(ConnectionLease {
3620            store: Arc::clone(self),
3621            principal: principal.clone(),
3622        })
3623    }
3624
3625    fn reserve_connection_slot(&self, uid: &UserId) -> Result<(), AuthError> {
3626        let attrs = self.user_attributes(uid);
3627        if let Some(deadline) = attrs.valid_until {
3628            if now_ms() >= deadline {
3629                return Err(AuthError::Forbidden(format!(
3630                    "account `{uid}` expired (VALID UNTIL exceeded)"
3631                )));
3632            }
3633        }
3634        let mut counts = self
3635            .session_count_by_user
3636            .write()
3637            .map_err(|_| AuthError::Internal("connection admission lock poisoned".to_string()))?;
3638        let current = counts.get(uid).copied().unwrap_or(0);
3639        if let Some(limit) = attrs.connection_limit {
3640            if current >= limit {
3641                return Err(AuthError::Forbidden(format!(
3642                    "account `{uid}` exceeded CONNECTION LIMIT ({limit})"
3643                )));
3644            }
3645        }
3646        counts.insert(uid.clone(), current.saturating_add(1));
3647        Ok(())
3648    }
3649
3650    /// Decrement the live-session count for `uid`. Call from session
3651    /// revoke / expiry paths so CONNECTION LIMIT stays accurate.
3652    pub fn decrement_session_count(&self, uid: &UserId) {
3653        if let Ok(mut counts) = self.session_count_by_user.write() {
3654            if let Some(c) = counts.get_mut(uid) {
3655                *c = c.saturating_sub(1);
3656            }
3657        }
3658    }
3659
3660    // -----------------------------------------------------------------
3661    // ACL persistence — vault_kv backed
3662    // -----------------------------------------------------------------
3663
3664    /// Re-read the ACL state from `vault_kv`. Call after vault load /
3665    /// restore so the in-memory maps reflect the persisted data.
3666    pub fn rehydrate_acl(&self) {
3667        let kv_snapshot: Vec<(String, String)> = self
3668            .vault_kv
3669            .read()
3670            .map(|kv| {
3671                kv.iter()
3672                    .filter(|(k, _)| {
3673                        k.starts_with("red.acl.grants.")
3674                            || k.starts_with("red.acl.attrs.")
3675                            || k == &"red.acl.public_grants"
3676                    })
3677                    .map(|(k, v)| (k.clone(), v.clone()))
3678                    .collect()
3679            })
3680            .unwrap_or_default();
3681
3682        for (k, v) in kv_snapshot {
3683            if k == "red.acl.public_grants" {
3684                if let Some(parsed) = decode_grants_blob(&v) {
3685                    *self
3686                        .public_grants
3687                        .write()
3688                        .unwrap_or_else(|e| e.into_inner()) = parsed;
3689                }
3690            } else if let Some(suffix) = k.strip_prefix("red.acl.grants.") {
3691                if let Some(uid) = decode_uid(suffix) {
3692                    if let Some(mut parsed) = decode_grants_blob(&v) {
3693                        // Restore the principal field — the on-disk
3694                        // line stores only resource+action shape.
3695                        for g in parsed.iter_mut() {
3696                            g.principal = GrantPrincipal::User(uid.clone());
3697                        }
3698                        self.grants
3699                            .write()
3700                            .unwrap_or_else(|e| e.into_inner())
3701                            .insert(uid, parsed);
3702                    }
3703                }
3704            } else if let Some(suffix) = k.strip_prefix("red.acl.attrs.") {
3705                if let Some(uid) = decode_uid(suffix) {
3706                    if let Some(parsed) = decode_attrs_blob(&v) {
3707                        self.user_attributes
3708                            .write()
3709                            .unwrap_or_else(|e| e.into_inner())
3710                            .insert(uid, parsed);
3711                    }
3712                }
3713            }
3714        }
3715
3716        self.permission_cache
3717            .write()
3718            .unwrap_or_else(|e| e.into_inner())
3719            .clear();
3720    }
3721
3722    /// Snapshot every ACL change back into the vault KV store.
3723    fn persist_acl_to_kv(&self) {
3724        let public = self
3725            .public_grants
3726            .read()
3727            .ok()
3728            .map(|p| encode_grants_blob(&p))
3729            .unwrap_or_default();
3730        self.vault_kv_set("red.acl.public_grants".to_string(), public);
3731
3732        let snapshot: Vec<(UserId, Vec<Grant>)> = self
3733            .grants
3734            .read()
3735            .ok()
3736            .map(|g| g.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
3737            .unwrap_or_default();
3738        for (uid, list) in snapshot {
3739            let key = format!("red.acl.grants.{}", encode_uid(&uid));
3740            let val = encode_grants_blob(&list);
3741            self.vault_kv_set(key, val);
3742        }
3743
3744        let attrs_snapshot: Vec<(UserId, UserAttributes)> = self
3745            .user_attributes
3746            .read()
3747            .ok()
3748            .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
3749            .unwrap_or_default();
3750        for (uid, attrs) in attrs_snapshot {
3751            let key = format!("red.acl.attrs.{}", encode_uid(&uid));
3752            let val = encode_attrs_blob(&attrs);
3753            self.vault_kv_set(key, val);
3754        }
3755    }
3756
3757    fn invalidate_permission_cache(&self, uid: Option<&UserId>) {
3758        if let Ok(mut cache) = self.permission_cache.write() {
3759            match uid {
3760                Some(u) => {
3761                    cache.remove(u);
3762                }
3763                None => cache.clear(),
3764            }
3765        }
3766    }
3767
3768    // -----------------------------------------------------------------
3769    // IAM policies — put / delete / attach / detach / simulate
3770    //
3771    // The kernel in `super::policies` owns the Policy type and the
3772    // evaluator. AuthStore handles persistence + per-user cache + the
3773    // GRANT translation layer (synthetic `_grant_*` policies).
3774    // -----------------------------------------------------------------
3775
3776    pub fn put_policy_with_control_events(
3777        &self,
3778        p: Policy,
3779        control: &PolicyMutationControl<'_>,
3780    ) -> Result<(), AuthError> {
3781        let policy_id = p.id.clone();
3782        let kind = if self.get_policy(&policy_id).is_some() {
3783            EventKind::PolicyUpdate
3784        } else {
3785            EventKind::PolicyCreate
3786        };
3787
3788        if p.id.starts_with("_grant_") || p.id.starts_with("_default_") {
3789            let err = AuthError::Forbidden(format!("policy id `{}` is reserved", p.id));
3790            self.emit_policy_error(
3791                control,
3792                kind,
3793                "policy:put",
3794                &policy_id,
3795                Some(&p),
3796                None,
3797                &err,
3798            );
3799            return Err(err);
3800        }
3801
3802        if let Some(ManagedPolicyDecision::Deny {
3803            entry_id,
3804            matched_action,
3805            matched_resource,
3806            reason,
3807            ..
3808        }) = self.managed_policy_decision(&policy_id, PolicyOp::Put, control)
3809        {
3810            self.emit_policy_denied(
3811                control,
3812                kind,
3813                "policy:put",
3814                &policy_id,
3815                Some(&p),
3816                None,
3817                &reason.to_string(),
3818                Some(entry_id),
3819                Some((matched_action, matched_resource)),
3820            );
3821            return Err(Self::managed_policy_error(&policy_id, &reason.to_string()));
3822        }
3823
3824        let previous = self.get_policy(&policy_id);
3825        let was_enabled = self.iam_authorization_enabled();
3826        match self.put_policy_internal(p.clone()) {
3827            Ok(()) => match self.emit_policy_allowed(
3828                control,
3829                kind,
3830                "policy:put",
3831                &policy_id,
3832                Some(&p),
3833                None,
3834                None,
3835            ) {
3836                Ok(()) => Ok(()),
3837                Err(err) => {
3838                    self.restore_policy_put(&policy_id, previous, was_enabled);
3839                    Err(err)
3840                }
3841            },
3842            Err(err) => {
3843                self.emit_policy_error(
3844                    control,
3845                    kind,
3846                    "policy:put",
3847                    &policy_id,
3848                    Some(&p),
3849                    None,
3850                    &err,
3851                );
3852                Err(err)
3853            }
3854        }
3855    }
3856
3857    pub fn delete_policy_with_control_events(
3858        &self,
3859        id: &str,
3860        control: &PolicyMutationControl<'_>,
3861    ) -> Result<(), AuthError> {
3862        let existing = self.get_policy(id);
3863        if let Some(ManagedPolicyDecision::Deny {
3864            entry_id,
3865            matched_action,
3866            matched_resource,
3867            reason,
3868            ..
3869        }) = self.managed_policy_decision(id, PolicyOp::Drop, control)
3870        {
3871            self.emit_policy_denied(
3872                control,
3873                EventKind::PolicyDelete,
3874                "policy:drop",
3875                id,
3876                existing.as_deref(),
3877                None,
3878                &reason.to_string(),
3879                Some(entry_id),
3880                Some((matched_action, matched_resource)),
3881            );
3882            return Err(Self::managed_policy_error(id, &reason.to_string()));
3883        }
3884
3885        let user_attachments = self
3886            .user_attachments
3887            .read()
3888            .map(|m| m.clone())
3889            .unwrap_or_default();
3890        let group_attachments = self
3891            .group_attachments
3892            .read()
3893            .map(|m| m.clone())
3894            .unwrap_or_default();
3895        let was_enabled = self.iam_authorization_enabled();
3896        match self.delete_policy(id) {
3897            Ok(()) => match self.emit_policy_allowed(
3898                control,
3899                EventKind::PolicyDelete,
3900                "policy:drop",
3901                id,
3902                existing.as_deref(),
3903                None,
3904                None,
3905            ) {
3906                Ok(()) => Ok(()),
3907                Err(err) => {
3908                    if let Some(policy) = existing {
3909                        self.restore_policy_delete(
3910                            id,
3911                            policy,
3912                            user_attachments,
3913                            group_attachments,
3914                            was_enabled,
3915                        );
3916                    }
3917                    Err(err)
3918                }
3919            },
3920            Err(err) => {
3921                self.emit_policy_error(
3922                    control,
3923                    EventKind::PolicyDelete,
3924                    "policy:drop",
3925                    id,
3926                    existing.as_deref(),
3927                    None,
3928                    &err,
3929                );
3930                Err(err)
3931            }
3932        }
3933    }
3934
3935    pub fn attach_policy_with_control_events(
3936        &self,
3937        principal: PrincipalRef,
3938        policy_id: &str,
3939        control: &PolicyMutationControl<'_>,
3940    ) -> Result<(), AuthError> {
3941        let existing = self.get_policy(policy_id);
3942        if let Some(ManagedPolicyDecision::Deny {
3943            entry_id,
3944            matched_action,
3945            matched_resource,
3946            reason,
3947            ..
3948        }) = self.managed_policy_decision(policy_id, PolicyOp::Attach, control)
3949        {
3950            self.emit_policy_denied(
3951                control,
3952                EventKind::PolicyAttach,
3953                "policy:attach",
3954                policy_id,
3955                existing.as_deref(),
3956                Some(&principal),
3957                &reason.to_string(),
3958                Some(entry_id),
3959                Some((matched_action, matched_resource)),
3960            );
3961            return Err(Self::managed_policy_error(policy_id, &reason.to_string()));
3962        }
3963
3964        if let Some(err_msg) = self.check_self_lock_invariant_for_attach(policy_id) {
3965            self.emit_policy_denied(
3966                control,
3967                EventKind::PolicyAttach,
3968                "policy:attach",
3969                policy_id,
3970                existing.as_deref(),
3971                Some(&principal),
3972                &err_msg,
3973                None,
3974                None,
3975            );
3976            return Err(AuthError::Forbidden(err_msg));
3977        }
3978
3979        let user_attachments = self
3980            .user_attachments
3981            .read()
3982            .map(|m| m.clone())
3983            .unwrap_or_default();
3984        let group_attachments = self
3985            .group_attachments
3986            .read()
3987            .map(|m| m.clone())
3988            .unwrap_or_default();
3989        match self.attach_policy(principal.clone(), policy_id) {
3990            Ok(()) => match self.emit_policy_allowed(
3991                control,
3992                EventKind::PolicyAttach,
3993                "policy:attach",
3994                policy_id,
3995                existing.as_deref(),
3996                Some(&principal),
3997                None,
3998            ) {
3999                Ok(()) => Ok(()),
4000                Err(err) => {
4001                    self.restore_policy_attachments(user_attachments, group_attachments);
4002                    Err(err)
4003                }
4004            },
4005            Err(err) => {
4006                self.emit_policy_error(
4007                    control,
4008                    EventKind::PolicyAttach,
4009                    "policy:attach",
4010                    policy_id,
4011                    existing.as_deref(),
4012                    Some(&principal),
4013                    &err,
4014                );
4015                Err(err)
4016            }
4017        }
4018    }
4019
4020    pub fn detach_policy_with_control_events(
4021        &self,
4022        principal: PrincipalRef,
4023        policy_id: &str,
4024        control: &PolicyMutationControl<'_>,
4025    ) -> Result<(), AuthError> {
4026        let existing = self.get_policy(policy_id);
4027        if let Some(ManagedPolicyDecision::Deny {
4028            entry_id,
4029            matched_action,
4030            matched_resource,
4031            reason,
4032            ..
4033        }) = self.managed_policy_decision(policy_id, PolicyOp::Detach, control)
4034        {
4035            self.emit_policy_denied(
4036                control,
4037                EventKind::PolicyDetach,
4038                "policy:detach",
4039                policy_id,
4040                existing.as_deref(),
4041                Some(&principal),
4042                &reason.to_string(),
4043                Some(entry_id),
4044                Some((matched_action, matched_resource)),
4045            );
4046            return Err(Self::managed_policy_error(policy_id, &reason.to_string()));
4047        }
4048        let Some(existing_policy) = existing else {
4049            let err = AuthError::Forbidden(format!("policy `{policy_id}` not found"));
4050            self.emit_policy_error(
4051                control,
4052                EventKind::PolicyDetach,
4053                "policy:detach",
4054                policy_id,
4055                None,
4056                Some(&principal),
4057                &err,
4058            );
4059            return Err(err);
4060        };
4061
4062        let user_attachments = self
4063            .user_attachments
4064            .read()
4065            .map(|m| m.clone())
4066            .unwrap_or_default();
4067        let group_attachments = self
4068            .group_attachments
4069            .read()
4070            .map(|m| m.clone())
4071            .unwrap_or_default();
4072        match self.detach_policy(principal.clone(), policy_id) {
4073            Ok(()) => match self.emit_policy_allowed(
4074                control,
4075                EventKind::PolicyDetach,
4076                "policy:detach",
4077                policy_id,
4078                Some(existing_policy.as_ref()),
4079                Some(&principal),
4080                None,
4081            ) {
4082                Ok(()) => Ok(()),
4083                Err(err) => {
4084                    self.restore_policy_attachments(user_attachments, group_attachments);
4085                    Err(err)
4086                }
4087            },
4088            Err(err) => {
4089                self.emit_policy_error(
4090                    control,
4091                    EventKind::PolicyDetach,
4092                    "policy:detach",
4093                    policy_id,
4094                    Some(existing_policy.as_ref()),
4095                    Some(&principal),
4096                    &err,
4097                );
4098                Err(err)
4099            }
4100        }
4101    }
4102
4103    fn managed_policy_decision(
4104        &self,
4105        policy_id: &str,
4106        op: PolicyOp,
4107        control: &PolicyMutationControl<'_>,
4108    ) -> Option<ManagedPolicyDecision> {
4109        control.registry.map(|registry| {
4110            ManagedPolicyGate::new(registry).check_mutation(
4111                self,
4112                control.actor,
4113                control.eval_ctx,
4114                policy_id,
4115                op,
4116            )
4117        })
4118    }
4119
4120    /// Run the [`crate::auth::self_lock_guard`] invariant against the
4121    /// current policy set. Returns `Some(error_message)` when an attach
4122    /// must be refused, `None` when the system would remain unlockable.
4123    ///
4124    /// `_policy_id` is informational — the invariant is on the full
4125    /// graph and doesn't currently special-case the new entry.
4126    fn check_self_lock_invariant_for_attach(&self, _policy_id: &str) -> Option<String> {
4127        use crate::auth::self_lock_guard::{
4128            check_self_lock_invariant, format_block_error, InvariantOutcome,
4129        };
4130        let policies: Vec<Arc<Policy>> = match self.policies.read() {
4131            Ok(map) => map.values().cloned().collect(),
4132            Err(_) => return None,
4133        };
4134        let outcome = check_self_lock_invariant(&policies);
4135        match &outcome {
4136            InvariantOutcome::Ok => None,
4137            InvariantOutcome::Blocked { .. } => format_block_error(&outcome),
4138        }
4139    }
4140
4141    fn managed_policy_error(policy_id: &str, reason: &str) -> AuthError {
4142        AuthError::Forbidden(format!(
4143            "managed policy mutation blocked for `{policy_id}`: {reason}"
4144        ))
4145    }
4146
4147    fn emit_policy_allowed(
4148        &self,
4149        control: &PolicyMutationControl<'_>,
4150        kind: EventKind,
4151        action: &'static str,
4152        policy_id: &str,
4153        policy: Option<&Policy>,
4154        principal: Option<&PrincipalRef>,
4155        matched_policy_id: Option<String>,
4156    ) -> Result<(), AuthError> {
4157        let event = ControlEvent {
4158            kind,
4159            outcome: Outcome::Allowed,
4160            action: std::borrow::Cow::Borrowed(action),
4161            resource: Some(format!("policy:{policy_id}")),
4162            reason: None,
4163            matched_policy_id,
4164            fields: policy_control_fields(policy_id, policy, principal),
4165        };
4166        match control.ledger.emit(control.ctx, event) {
4167            Ok(_) => Ok(()),
4168            Err(err) if control.config.require_persistence() => {
4169                Err(AuthError::Internal(err.to_string()))
4170            }
4171            Err(_) => Ok(()),
4172        }
4173    }
4174
4175    #[allow(clippy::too_many_arguments)]
4176    fn emit_policy_denied(
4177        &self,
4178        control: &PolicyMutationControl<'_>,
4179        kind: EventKind,
4180        action: &'static str,
4181        policy_id: &str,
4182        policy: Option<&Policy>,
4183        principal: Option<&PrincipalRef>,
4184        reason: &str,
4185        matched_policy_id: Option<String>,
4186        matched: Option<(String, String)>,
4187    ) {
4188        let mut fields = policy_control_fields(policy_id, policy, principal);
4189        if let Some((matched_action, matched_resource)) = matched {
4190            fields.insert(
4191                "matched_action".to_string(),
4192                Sensitivity::raw(matched_action),
4193            );
4194            fields.insert(
4195                "matched_resource".to_string(),
4196                Sensitivity::raw(matched_resource),
4197            );
4198        }
4199        let event = ControlEvent {
4200            kind,
4201            outcome: Outcome::Denied,
4202            action: std::borrow::Cow::Borrowed(action),
4203            resource: Some(format!("policy:{policy_id}")),
4204            reason: Some(reason.to_string()),
4205            matched_policy_id,
4206            fields,
4207        };
4208        let _ = control.ledger.emit(control.ctx, event);
4209    }
4210
4211    fn emit_policy_error(
4212        &self,
4213        control: &PolicyMutationControl<'_>,
4214        kind: EventKind,
4215        action: &'static str,
4216        policy_id: &str,
4217        policy: Option<&Policy>,
4218        principal: Option<&PrincipalRef>,
4219        err: &AuthError,
4220    ) {
4221        let event = ControlEvent {
4222            kind,
4223            outcome: Outcome::Error,
4224            action: std::borrow::Cow::Borrowed(action),
4225            resource: Some(format!("policy:{policy_id}")),
4226            reason: Some(err.to_string()),
4227            matched_policy_id: None,
4228            fields: policy_control_fields(policy_id, policy, principal),
4229        };
4230        let _ = control.ledger.emit(control.ctx, event);
4231    }
4232
4233    fn restore_policy_put(
4234        &self,
4235        policy_id: &str,
4236        previous: Option<Arc<Policy>>,
4237        was_enabled: bool,
4238    ) {
4239        let mut policies = self.policies.write().unwrap_or_else(|e| e.into_inner());
4240        match previous {
4241            Some(policy) => {
4242                policies.insert(policy_id.to_string(), policy);
4243            }
4244            None => {
4245                policies.remove(policy_id);
4246            }
4247        }
4248        drop(policies);
4249        self.iam_authorization_enabled
4250            .store(was_enabled, Ordering::Release);
4251        self.iam_effective_cache
4252            .write()
4253            .unwrap_or_else(|e| e.into_inner())
4254            .clear();
4255        self.invalidate_visible_collections_cache();
4256        self.persist_iam_to_kv();
4257    }
4258
4259    fn restore_policy_delete(
4260        &self,
4261        policy_id: &str,
4262        policy: Arc<Policy>,
4263        user_attachments: HashMap<UserId, Vec<String>>,
4264        group_attachments: HashMap<String, Vec<String>>,
4265        was_enabled: bool,
4266    ) {
4267        self.policies
4268            .write()
4269            .unwrap_or_else(|e| e.into_inner())
4270            .insert(policy_id.to_string(), policy);
4271        self.restore_policy_attachments(user_attachments, group_attachments);
4272        self.iam_authorization_enabled
4273            .store(was_enabled, Ordering::Release);
4274        self.persist_iam_to_kv();
4275    }
4276
4277    fn restore_policy_attachments(
4278        &self,
4279        user_attachments: HashMap<UserId, Vec<String>>,
4280        group_attachments: HashMap<String, Vec<String>>,
4281    ) {
4282        *self
4283            .user_attachments
4284            .write()
4285            .unwrap_or_else(|e| e.into_inner()) = user_attachments;
4286        *self
4287            .group_attachments
4288            .write()
4289            .unwrap_or_else(|e| e.into_inner()) = group_attachments;
4290        self.iam_effective_cache
4291            .write()
4292            .unwrap_or_else(|e| e.into_inner())
4293            .clear();
4294        self.invalidate_visible_collections_cache();
4295        self.persist_iam_to_kv();
4296    }
4297
4298    /// Insert or replace a policy by id. Rejects synthetic ids
4299    /// (`_grant_*` / `_default_*`) so callers can't hand-write them
4300    /// from the public API. Use `put_policy_internal` for synthetic
4301    /// inserts.
4302    pub fn put_policy(&self, p: Policy) -> Result<(), AuthError> {
4303        if p.id.starts_with("_grant_") || p.id.starts_with("_default_") {
4304            return Err(AuthError::Forbidden(format!(
4305                "policy id `{}` is reserved",
4306                p.id
4307            )));
4308        }
4309        self.put_policy_internal(p)
4310    }
4311
4312    /// Internal put bypassing the synthetic-namespace guard. Used by
4313    /// the GRANT translation layer; exposed publicly so integration
4314    /// tests can register synthetic `_grant_*` policies without going
4315    /// through the SQL frontend.
4316    pub fn put_policy_internal(&self, p: Policy) -> Result<(), AuthError> {
4317        p.validate()
4318            .map_err(|e| AuthError::Forbidden(format!("invalid policy `{}`: {e}", p.id)))?;
4319        let id = p.id.clone();
4320        self.policies
4321            .write()
4322            .unwrap_or_else(|e| e.into_inner())
4323            .insert(id, Arc::new(p));
4324        self.iam_authorization_enabled
4325            .store(true, Ordering::Release);
4326        self.iam_effective_cache
4327            .write()
4328            .unwrap_or_else(|e| e.into_inner())
4329            .clear();
4330        // Issue #119: a policy mutation can change the visible-
4331        // collections answer for any (tenant, role); we don't know
4332        // which up-front, so blow the whole cache.
4333        self.invalidate_visible_collections_cache();
4334        self.persist_iam_to_kv();
4335        Ok(())
4336    }
4337
4338    /// Whether the IAM evaluator should be authoritative for runtime
4339    /// authorization. This flips on the first policy write and remains
4340    /// on after deletes so dropping all policies leaves the instance in
4341    /// default-deny rather than silently returning to role fallback.
4342    pub fn iam_authorization_enabled(&self) -> bool {
4343        self.iam_authorization_enabled.load(Ordering::Acquire)
4344    }
4345
4346    /// Remove a policy and any attachments referencing it.
4347    pub fn delete_policy(&self, id: &str) -> Result<(), AuthError> {
4348        let removed = self
4349            .policies
4350            .write()
4351            .unwrap_or_else(|e| e.into_inner())
4352            .remove(id)
4353            .is_some();
4354        if !removed {
4355            return Err(AuthError::Forbidden(format!("policy `{id}` not found")));
4356        }
4357        // Detach from every user / group.
4358        if let Ok(mut ua) = self.user_attachments.write() {
4359            for ids in ua.values_mut() {
4360                ids.retain(|p| p != id);
4361            }
4362            ua.retain(|_, v| !v.is_empty());
4363        }
4364        if let Ok(mut ga) = self.group_attachments.write() {
4365            for ids in ga.values_mut() {
4366                ids.retain(|p| p != id);
4367            }
4368            ga.retain(|_, v| !v.is_empty());
4369        }
4370        self.iam_effective_cache
4371            .write()
4372            .unwrap_or_else(|e| e.into_inner())
4373            .clear();
4374        // Issue #119: dropping a policy can shrink any caller's visible
4375        // set; clear the (tenant, role) cache so AI commands re-resolve.
4376        self.invalidate_visible_collections_cache();
4377        self.persist_iam_to_kv();
4378        Ok(())
4379    }
4380
4381    /// List all policies (id-sorted for deterministic output).
4382    pub fn list_policies(&self) -> Vec<Arc<Policy>> {
4383        let map = match self.policies.read() {
4384            Ok(g) => g,
4385            Err(_) => return Vec::new(),
4386        };
4387        let mut out: Vec<Arc<Policy>> = map.values().cloned().collect();
4388        out.sort_by(|a, b| a.id.cmp(&b.id));
4389        out
4390    }
4391
4392    /// Fetch a single policy by id.
4393    pub fn get_policy(&self, id: &str) -> Option<Arc<Policy>> {
4394        self.policies.read().ok().and_then(|m| m.get(id).cloned())
4395    }
4396
4397    /// List policies directly attached to a group.
4398    pub fn group_policies(&self, group: &str) -> Vec<Arc<Policy>> {
4399        let policies = self.policies.read();
4400        let attachments = self.group_attachments.read();
4401        let mut out = Vec::new();
4402        if let (Ok(p_map), Ok(ga_map)) = (policies, attachments) {
4403            if let Some(ids) = ga_map.get(group) {
4404                for id in ids {
4405                    if let Some(p) = p_map.get(id) {
4406                        out.push(p.clone());
4407                    }
4408                }
4409            }
4410        }
4411        out.sort_by(|a, b| a.id.cmp(&b.id));
4412        out
4413    }
4414
4415    /// Delete synthetic policies produced by SQL GRANT translation.
4416    /// REVOKE uses this to keep the IAM lane and the legacy grant table
4417    /// in lock-step.
4418    pub fn delete_synthetic_grant_policies(
4419        &self,
4420        principal: &GrantPrincipal,
4421        resource: &Resource,
4422        actions: &[Action],
4423    ) -> usize {
4424        let attached = match principal {
4425            GrantPrincipal::User(uid) => self
4426                .user_attachments
4427                .read()
4428                .ok()
4429                .and_then(|m| m.get(uid).cloned())
4430                .unwrap_or_default(),
4431            GrantPrincipal::Group(group) => self
4432                .group_attachments
4433                .read()
4434                .ok()
4435                .and_then(|m| m.get(group).cloned())
4436                .unwrap_or_default(),
4437            GrantPrincipal::Public => self
4438                .group_attachments
4439                .read()
4440                .ok()
4441                .and_then(|m| m.get(PUBLIC_IAM_GROUP).cloned())
4442                .unwrap_or_default(),
4443        };
4444        if attached.is_empty() {
4445            return 0;
4446        }
4447
4448        let mut delete_ids = Vec::new();
4449        if let Ok(policies) = self.policies.read() {
4450            for id in attached {
4451                let Some(policy) = policies.get(&id) else {
4452                    continue;
4453                };
4454                if !policy.id.starts_with("_grant_") {
4455                    continue;
4456                }
4457                if synthetic_grant_matches(policy, resource, actions) {
4458                    delete_ids.push(policy.id.clone());
4459                }
4460            }
4461        }
4462
4463        let mut deleted = 0usize;
4464        for id in delete_ids {
4465            if self.delete_policy(&id).is_ok() {
4466                deleted += 1;
4467            }
4468        }
4469        deleted
4470    }
4471
4472    /// Apply the `REDDB_POLICY_BREAK_GLASS` recovery path: install or
4473    /// refresh the unlock policy, ensure the synthetic platform-owner
4474    /// user record exists, rebind the unlock policy to it, and emit a
4475    /// `policy.break_glass` audit event. Idempotent across reboots.
4476    ///
4477    /// `boot_ts_ms` is recorded in the audit event so operators can
4478    /// correlate the recovery with the boot transcript.
4479    pub fn apply_policy_break_glass(&self, boot_ts_ms: u128) -> Result<(), AuthError> {
4480        use crate::auth::self_lock_guard::{
4481            break_glass_audit_fields, unlock_policy, PLATFORM_OWNER_UNLOCK_POLICY_ID,
4482            PLATFORM_OWNER_USERNAME,
4483        };
4484
4485        // (1) Install or refresh the unlock policy. `put_policy_internal`
4486        // bypasses the synthetic-namespace guard, lets us replace any
4487        // tampered persisted copy, and persists to vault on success.
4488        let policy = unlock_policy();
4489        self.put_policy_internal(policy)?;
4490
4491        // (2) Ensure the synthetic platform-owner user record exists.
4492        // We bypass the public user-creation API to keep this user
4493        // immutable, disabled, and absent from operator listings even
4494        // if our filter ever regresses.
4495        let owner_id = UserId::platform(PLATFORM_OWNER_USERNAME);
4496        {
4497            let mut users = self.users.write().unwrap_or_else(|e| e.into_inner());
4498            users.entry(owner_id.clone()).or_insert_with(|| User {
4499                username: PLATFORM_OWNER_USERNAME.to_string(),
4500                tenant_id: None,
4501                // No usable password; even if `authenticate` were
4502                // wired to accept the principal it would fail
4503                // verification. Defence in depth: `enabled = false`
4504                // and the username filter both block it.
4505                password_hash: String::new(),
4506                scram_verifier: None,
4507                role: Role::Admin,
4508                api_keys: Vec::new(),
4509                created_at: boot_ts_ms,
4510                updated_at: boot_ts_ms,
4511                enabled: false,
4512            });
4513        }
4514
4515        // (3) Rebind the unlock policy to the synthetic principal.
4516        // `attach_policy` is idempotent (the helper checks for an
4517        // existing binding before appending).
4518        self.attach_policy(
4519            PrincipalRef::User(owner_id.clone()),
4520            PLATFORM_OWNER_UNLOCK_POLICY_ID,
4521        )?;
4522
4523        // (4) Emit a loud audit event. If the control-event ledger
4524        // isn't configured (e.g. unit tests without runtime wiring),
4525        // log a warn and proceed — recovery must not depend on audit
4526        // wiring being live.
4527        let fields_raw = break_glass_audit_fields(boot_ts_ms);
4528        let mut fields: HashMap<String, Sensitivity> = HashMap::new();
4529        for (k, v) in fields_raw {
4530            fields.insert(k, Sensitivity::raw(v));
4531        }
4532        if let Some(configured) = self.configured_control_events() {
4533            let ctx = ControlEventCtx {
4534                actor: crate::runtime::control_events::ActorRef::System("break_glass"),
4535                scope: None,
4536                request_id: None,
4537                trace_id: None,
4538            };
4539            let event = ControlEvent {
4540                kind: EventKind::PolicyBreakGlass,
4541                outcome: Outcome::Allowed,
4542                action: std::borrow::Cow::Borrowed("policy:break_glass"),
4543                resource: Some(format!("policy:{PLATFORM_OWNER_UNLOCK_POLICY_ID}")),
4544                reason: Some("REDDB_POLICY_BREAK_GLASS=1 at boot".into()),
4545                matched_policy_id: None,
4546                fields,
4547            };
4548            let _ = configured.ledger.emit(&ctx, event);
4549        }
4550        tracing::warn!(
4551            policy_id = PLATFORM_OWNER_UNLOCK_POLICY_ID,
4552            principal = PLATFORM_OWNER_USERNAME,
4553            boot_ts_ms = %boot_ts_ms,
4554            "REDDB_POLICY_BREAK_GLASS=1 — installed/refreshed platform-owner unlock policy and rebound to synthetic principal"
4555        );
4556        Ok(())
4557    }
4558
4559    /// Attach a policy to a user or group. Returns an error if the
4560    /// policy id doesn't exist.
4561    pub fn attach_policy(&self, principal: PrincipalRef, policy_id: &str) -> Result<(), AuthError> {
4562        if !self
4563            .policies
4564            .read()
4565            .map(|m| m.contains_key(policy_id))
4566            .unwrap_or(false)
4567        {
4568            return Err(AuthError::Forbidden(format!(
4569                "policy `{policy_id}` not found"
4570            )));
4571        }
4572        match &principal {
4573            PrincipalRef::User(uid) => {
4574                let mut ua = self
4575                    .user_attachments
4576                    .write()
4577                    .unwrap_or_else(|e| e.into_inner());
4578                let list = ua.entry(uid.clone()).or_default();
4579                if !list.iter().any(|p| p == policy_id) {
4580                    list.push(policy_id.to_string());
4581                }
4582                drop(ua);
4583                self.invalidate_iam_cache(Some(uid));
4584            }
4585            PrincipalRef::Group(g) => {
4586                let mut ga = self
4587                    .group_attachments
4588                    .write()
4589                    .unwrap_or_else(|e| e.into_inner());
4590                let list = ga.entry(g.clone()).or_default();
4591                if !list.iter().any(|p| p == policy_id) {
4592                    list.push(policy_id.to_string());
4593                }
4594                drop(ga);
4595                self.invalidate_iam_cache(None);
4596            }
4597        }
4598        self.persist_iam_to_kv();
4599        Ok(())
4600    }
4601
4602    /// Remove a policy attachment from a user or group.
4603    pub fn detach_policy(&self, principal: PrincipalRef, policy_id: &str) -> Result<(), AuthError> {
4604        match &principal {
4605            PrincipalRef::User(uid) => {
4606                if let Ok(mut ua) = self.user_attachments.write() {
4607                    if let Some(list) = ua.get_mut(uid) {
4608                        list.retain(|p| p != policy_id);
4609                        if list.is_empty() {
4610                            ua.remove(uid);
4611                        }
4612                    }
4613                }
4614                self.invalidate_iam_cache(Some(uid));
4615            }
4616            PrincipalRef::Group(g) => {
4617                if let Ok(mut ga) = self.group_attachments.write() {
4618                    if let Some(list) = ga.get_mut(g) {
4619                        list.retain(|p| p != policy_id);
4620                        if list.is_empty() {
4621                            ga.remove(g);
4622                        }
4623                    }
4624                }
4625                self.invalidate_iam_cache(None);
4626            }
4627        }
4628        self.persist_iam_to_kv();
4629        Ok(())
4630    }
4631
4632    /// Resolve the ordered list of effective policies for a user:
4633    /// group attachments first (least specific), then user
4634    /// attachments (most specific). Cached per user.
4635    pub fn effective_policies(&self, user: &UserId) -> Vec<Arc<Policy>> {
4636        if let Ok(cache) = self.iam_effective_cache.read() {
4637            if let Some(hit) = cache.get(user) {
4638                return hit.clone();
4639            }
4640        }
4641        let policies = self.policies.read();
4642        let user_attachments = self.user_attachments.read();
4643        let group_attachments = self.group_attachments.read();
4644        let mut groups = self
4645            .user_attributes
4646            .read()
4647            .ok()
4648            .and_then(|m| m.get(user).map(|attrs| attrs.groups.clone()))
4649            .unwrap_or_default();
4650        groups.insert(0, PUBLIC_IAM_GROUP.to_string());
4651        let mut out: Vec<Arc<Policy>> = Vec::new();
4652        if let (Ok(p_map), Ok(ua_map), Ok(ga_map)) = (policies, user_attachments, group_attachments)
4653        {
4654            for group in groups {
4655                if let Some(ids) = ga_map.get(&group) {
4656                    for id in ids {
4657                        if let Some(p) = p_map.get(id) {
4658                            out.push(p.clone());
4659                        }
4660                    }
4661                }
4662            }
4663            if let Some(ids) = ua_map.get(user) {
4664                for id in ids {
4665                    if let Some(p) = p_map.get(id) {
4666                        out.push(p.clone());
4667                    }
4668                }
4669            }
4670        }
4671        if let Ok(mut cache) = self.iam_effective_cache.write() {
4672            cache.insert(user.clone(), out.clone());
4673        }
4674        out
4675    }
4676
4677    /// Run the policy simulator for a principal. Synthesises an
4678    /// `EvalContext` from the user record + caller-supplied extras.
4679    pub fn simulate(
4680        &self,
4681        principal: &UserId,
4682        action: &str,
4683        resource: &ResourceRef,
4684        ctx_extras: SimCtx,
4685    ) -> SimulationOutcome {
4686        let user_role = self
4687            .users
4688            .read()
4689            .ok()
4690            .and_then(|u| u.get(principal).map(|u| u.role));
4691        let principal_is_admin_role = user_role == Some(Role::Admin);
4692        let now = ctx_extras.now_ms.unwrap_or_else(now_ms);
4693        let ctx = EvalContext {
4694            principal_tenant: principal.tenant.clone(),
4695            current_tenant: ctx_extras.current_tenant,
4696            peer_ip: ctx_extras.peer_ip,
4697            mfa_present: ctx_extras.mfa_present,
4698            now_ms: now,
4699            principal_is_admin_role,
4700            principal_is_platform_scoped: principal.tenant.is_none(),
4701        };
4702        let pols = self.effective_policies(principal);
4703        let refs: Vec<&Policy> = pols.iter().map(|p| p.as_ref()).collect();
4704        iam_policies::simulate(&refs, action, resource, &ctx)
4705    }
4706
4707    /// Production hot-path policy evaluation. Returns `true` on Allow
4708    /// / AdminBypass, `false` on Deny / DefaultDeny.
4709    ///
4710    /// This entry point is **strict**: it never consults the
4711    /// [`PolicyEnforcementMode`] fallback. Governance APIs that gate
4712    /// admin-tier mutations (managed-config writes, registry
4713    /// supersedes, managed-policy lifecycle) call this so they cannot
4714    /// accidentally pick up the lenient `LegacyRbac` posture. Runtime
4715    /// hot-path callers that should respect the mode call
4716    /// [`AuthStore::check_policy_authz_with_role`] instead.
4717    pub fn check_policy_authz(
4718        &self,
4719        principal: &UserId,
4720        action: &str,
4721        resource: &ResourceRef,
4722        ctx: &EvalContext,
4723    ) -> bool {
4724        let pols = self.effective_policies(principal);
4725        let refs: Vec<&Policy> = pols.iter().map(|p| p.as_ref()).collect();
4726        let decision = iam_policies::evaluate(&refs, action, resource, ctx);
4727        matches!(
4728            decision,
4729            iam_policies::Decision::Allow { .. } | iam_policies::Decision::AdminBypass
4730        )
4731    }
4732
4733    /// Mode-aware policy evaluation for runtime SQL/HTTP/wire surfaces.
4734    ///
4735    /// Returns `true` on `Allow`/`AdminBypass`, `false` on an explicit
4736    /// `Deny`. On `DefaultDeny` (no statement matched) the result
4737    /// depends on the active [`PolicyEnforcementMode`]:
4738    ///
4739    /// * `LegacyRbac` — defer to
4740    ///   [`legacy_rbac_decision`][super::enforcement_mode::legacy_rbac_decision]
4741    ///   using the caller's `role`. This preserves the pre-#712
4742    ///   behaviour where a principal with no attached policy is gated
4743    ///   only by their role.
4744    /// * `PolicyOnly` — return `false`. A principal needs an explicit
4745    ///   matching `Allow` to be authorized.
4746    ///
4747    /// An explicit `Deny` always wins, irrespective of mode and role.
4748    pub fn check_policy_authz_with_role(
4749        &self,
4750        principal: &UserId,
4751        action: &str,
4752        resource: &ResourceRef,
4753        ctx: &EvalContext,
4754        role: Role,
4755    ) -> bool {
4756        let pols = self.effective_policies(principal);
4757        let refs: Vec<&Policy> = pols.iter().map(|p| p.as_ref()).collect();
4758        let decision = iam_policies::evaluate(&refs, action, resource, ctx);
4759        match decision {
4760            iam_policies::Decision::Allow { .. } | iam_policies::Decision::AdminBypass => true,
4761            iam_policies::Decision::Deny { .. } => false,
4762            iam_policies::Decision::DefaultDeny => match self.enforcement_mode() {
4763                PolicyEnforcementMode::LegacyRbac => legacy_rbac_decision(role, action),
4764                PolicyEnforcementMode::PolicyOnly => false,
4765            },
4766        }
4767    }
4768
4769    // -----------------------------------------------------------------
4770    // PolicyEnforcementMode (#712)
4771    // -----------------------------------------------------------------
4772
4773    /// Read the active enforcement mode. Cheap (a single `RwLock` read);
4774    /// safe to call on the hot path.
4775    pub fn enforcement_mode(&self) -> PolicyEnforcementMode {
4776        *self
4777            .enforcement_mode
4778            .read()
4779            .unwrap_or_else(|e| e.into_inner())
4780    }
4781
4782    /// Overwrite the active enforcement mode and persist it to the
4783    /// vault KV so the new value survives a restart. Returns the
4784    /// previous value so callers logging a transition (e.g. the
4785    /// boot-time loader, the S5B migration command) can record the
4786    /// before/after.
4787    pub fn set_enforcement_mode(&self, mode: PolicyEnforcementMode) -> PolicyEnforcementMode {
4788        let prev = {
4789            let mut guard = self
4790                .enforcement_mode
4791                .write()
4792                .unwrap_or_else(|e| e.into_inner());
4793            let prev = *guard;
4794            *guard = mode;
4795            prev
4796        };
4797        self.vault_kv_set(
4798            "red.config.policy.enforcement_mode".to_string(),
4799            mode.as_str().to_string(),
4800        );
4801        prev
4802    }
4803
4804    /// Claim the "emit the one-time legacy-RBAC boot warning" token.
4805    /// Returns `true` on the first call after construction (or after a
4806    /// process restart) **iff** the active mode is `LegacyRbac`;
4807    /// returns `false` on every subsequent call so the boot path can
4808    /// guarantee the warning is logged at most once per boot.
4809    pub fn take_legacy_rbac_warn_once(&self) -> bool {
4810        if self.enforcement_mode() != PolicyEnforcementMode::LegacyRbac {
4811            return false;
4812        }
4813        !self
4814            .legacy_rbac_boot_warn_emitted
4815            .swap(true, Ordering::AcqRel)
4816    }
4817
4818    /// Evaluate a resolved table projection through the column policy
4819    /// gate. Query paths should pass already-resolved column names; this
4820    /// helper intentionally does not parse SQL projection syntax.
4821    pub fn check_column_projection_authz(
4822        &self,
4823        principal: &UserId,
4824        request: &ColumnAccessRequest,
4825        ctx: &EvalContext,
4826    ) -> ColumnPolicyOutcome {
4827        let pols = self.effective_policies(principal);
4828        let refs: Vec<&Policy> = pols.iter().map(|p| p.as_ref()).collect();
4829        ColumnPolicyGate::new(&refs).evaluate(request, ctx)
4830    }
4831
4832    fn invalidate_iam_cache(&self, uid: Option<&UserId>) {
4833        if let Ok(mut cache) = self.iam_effective_cache.write() {
4834            match uid {
4835                Some(u) => {
4836                    cache.remove(u);
4837                }
4838                None => cache.clear(),
4839            }
4840        }
4841    }
4842
4843    /// Drop every effective-policy cache entry. Called from execution
4844    /// paths that mutate policies/attachments without knowing which
4845    /// users will be affected.
4846    pub fn invalidate_all_iam_cache(&self) {
4847        self.invalidate_iam_cache(None);
4848    }
4849
4850    // -----------------------------------------------------------------
4851    // IAM persistence — vault_kv backed under `red.iam.*` keys
4852    // -----------------------------------------------------------------
4853
4854    /// Reload IAM state (policies + attachments) from the vault KV.
4855    /// Replaces the legacy `rehydrate_acl` reader — pre-1.0 we drop
4856    /// the old `red.acl.*` blob format entirely.
4857    pub fn rehydrate_iam(&self) {
4858        let mut enabled = self
4859            .vault_kv_get("red.iam.enabled")
4860            .map(|v| v == "true")
4861            .unwrap_or(false);
4862        // Policies — single JSON object keyed by id.
4863        if let Some(blob) = self.vault_kv_get("red.iam.policies") {
4864            if let Ok(val) = crate::serde_json::from_str::<crate::serde_json::Value>(&blob) {
4865                if let Some(obj) = val.as_object() {
4866                    let mut map = HashMap::new();
4867                    for (id, body) in obj.iter() {
4868                        let s = body.to_string_compact();
4869                        if let Ok(p) = Policy::from_json_str(&s) {
4870                            map.insert(id.clone(), Arc::new(p));
4871                        }
4872                    }
4873                    if !map.is_empty() {
4874                        enabled = true;
4875                    }
4876                    *self.policies.write().unwrap_or_else(|e| e.into_inner()) = map;
4877                }
4878            }
4879        }
4880        // User attachments.
4881        if let Some(blob) = self.vault_kv_get("red.iam.attachments.users") {
4882            if let Ok(val) = crate::serde_json::from_str::<crate::serde_json::Value>(&blob) {
4883                if let Some(obj) = val.as_object() {
4884                    let mut map: HashMap<UserId, Vec<String>> = HashMap::new();
4885                    for (encoded_uid, ids_v) in obj.iter() {
4886                        let Some(uid) = decode_uid(encoded_uid) else {
4887                            continue;
4888                        };
4889                        if let Some(arr) = ids_v.as_array() {
4890                            let ids: Vec<String> = arr
4891                                .iter()
4892                                .filter_map(|v| v.as_str().map(|s| s.to_string()))
4893                                .collect();
4894                            map.insert(uid, ids);
4895                        }
4896                    }
4897                    *self
4898                        .user_attachments
4899                        .write()
4900                        .unwrap_or_else(|e| e.into_inner()) = map;
4901                }
4902            }
4903        }
4904        // Group attachments.
4905        if let Some(blob) = self.vault_kv_get("red.iam.attachments.groups") {
4906            if let Ok(val) = crate::serde_json::from_str::<crate::serde_json::Value>(&blob) {
4907                if let Some(obj) = val.as_object() {
4908                    let mut map: HashMap<String, Vec<String>> = HashMap::new();
4909                    for (g, ids_v) in obj.iter() {
4910                        if let Some(arr) = ids_v.as_array() {
4911                            let ids: Vec<String> = arr
4912                                .iter()
4913                                .filter_map(|v| v.as_str().map(|s| s.to_string()))
4914                                .collect();
4915                            map.insert(g.clone(), ids);
4916                        }
4917                    }
4918                    *self
4919                        .group_attachments
4920                        .write()
4921                        .unwrap_or_else(|e| e.into_inner()) = map;
4922                }
4923            }
4924        }
4925        self.iam_authorization_enabled
4926            .store(enabled, Ordering::Release);
4927        self.invalidate_iam_cache(None);
4928
4929        // #712 / S5A: load persisted enforcement mode, if any. Absence
4930        // of the key means an existing install upgrading past #712 —
4931        // stay on the lenient default written at construction time.
4932        // Fresh installs receive `PolicyOnly` via the bootstrap path
4933        // (`mark_bootstrap_complete`) before the first restart, so this
4934        // hydrate path picks it up on every subsequent boot.
4935        if let Some(stored) = self.vault_kv_get("red.config.policy.enforcement_mode") {
4936            if let Some(mode) = PolicyEnforcementMode::parse(&stored) {
4937                *self
4938                    .enforcement_mode
4939                    .write()
4940                    .unwrap_or_else(|e| e.into_inner()) = mode;
4941            }
4942        }
4943
4944        // #712 / S5A: legacy_rbac is a transitional posture; emit one
4945        // boot-time warning so operators see the migration nudge in
4946        // their logs. The flag inside take_legacy_rbac_warn_once
4947        // guarantees we don't spam reload cycles.
4948        if self.take_legacy_rbac_warn_once() {
4949            tracing::warn!(
4950                target: "reddb::auth::enforcement_mode",
4951                mode = "legacy_rbac",
4952                hard_version = crate::auth::enforcement_mode::POLICY_ONLY_HARD_VERSION,
4953                "policy enforcement_mode=legacy_rbac (transitional). Run \
4954                 `MIGRATE POLICY MODE TO 'policy_only'` (next slice S5B) \
4955                 before {} to flip this install over to the strict posture.",
4956                crate::auth::enforcement_mode::POLICY_ONLY_HARD_VERSION,
4957            );
4958        }
4959    }
4960
4961    /// Snapshot policies + attachments into the vault KV. Called
4962    /// after every mutation.
4963    fn persist_iam_to_kv(&self) {
4964        let enabled = if self.iam_authorization_enabled() {
4965            "true"
4966        } else {
4967            "false"
4968        };
4969        self.vault_kv_set("red.iam.enabled".to_string(), enabled.to_string());
4970
4971        // Policies: `{ "<id>": <policy_json>, ... }`
4972        let policies_obj = {
4973            let map = self.policies.read().unwrap_or_else(|e| e.into_inner());
4974            let mut obj = crate::serde_json::Map::new();
4975            for (id, p) in map.iter() {
4976                let s = p.to_json_string();
4977                if let Ok(v) = crate::serde_json::from_str::<crate::serde_json::Value>(&s) {
4978                    obj.insert(id.clone(), v);
4979                }
4980            }
4981            crate::serde_json::Value::Object(obj).to_string_compact()
4982        };
4983        self.vault_kv_set("red.iam.policies".to_string(), policies_obj);
4984
4985        // User attachments: `{ "<encoded_uid>": [ "<policy_id>", ... ], ... }`
4986        let users_obj = {
4987            let map = self
4988                .user_attachments
4989                .read()
4990                .unwrap_or_else(|e| e.into_inner());
4991            let mut obj = crate::serde_json::Map::new();
4992            for (uid, ids) in map.iter() {
4993                let arr = crate::serde_json::Value::Array(
4994                    ids.iter()
4995                        .map(|s| crate::serde_json::Value::String(s.clone()))
4996                        .collect(),
4997                );
4998                obj.insert(encode_uid(uid), arr);
4999            }
5000            crate::serde_json::Value::Object(obj).to_string_compact()
5001        };
5002        self.vault_kv_set("red.iam.attachments.users".to_string(), users_obj);
5003
5004        // Group attachments.
5005        let groups_obj = {
5006            let map = self
5007                .group_attachments
5008                .read()
5009                .unwrap_or_else(|e| e.into_inner());
5010            let mut obj = crate::serde_json::Map::new();
5011            for (g, ids) in map.iter() {
5012                let arr = crate::serde_json::Value::Array(
5013                    ids.iter()
5014                        .map(|s| crate::serde_json::Value::String(s.clone()))
5015                        .collect(),
5016                );
5017                obj.insert(g.clone(), arr);
5018            }
5019            crate::serde_json::Value::Object(obj).to_string_compact()
5020        };
5021        self.vault_kv_set("red.iam.attachments.groups".to_string(), groups_obj);
5022    }
5023}
5024
5025fn synthetic_grant_matches(policy: &Policy, resource: &Resource, actions: &[Action]) -> bool {
5026    policy.statements.iter().any(|st| {
5027        st.effect == crate::auth::policies::Effect::Allow
5028            && st.condition.is_none()
5029            && grant_actions_overlap(&st.actions, actions)
5030            && grant_resource_matches(&st.resources, resource)
5031    })
5032}
5033
5034fn grant_actions_overlap(
5035    patterns: &[crate::auth::policies::ActionPattern],
5036    actions: &[Action],
5037) -> bool {
5038    if actions.contains(&Action::All) {
5039        return true;
5040    }
5041    patterns.iter().any(|pat| match pat {
5042        crate::auth::policies::ActionPattern::Wildcard => true,
5043        crate::auth::policies::ActionPattern::Exact(s) => {
5044            actions.iter().any(|a| s.eq_ignore_ascii_case(a.as_str()))
5045        }
5046        crate::auth::policies::ActionPattern::Prefix(_) => false,
5047    })
5048}
5049
5050fn grant_resource_matches(
5051    patterns: &[crate::auth::policies::ResourcePattern],
5052    resource: &Resource,
5053) -> bool {
5054    let expected = grant_resource_pattern(resource);
5055    patterns.iter().any(|pat| pat == &expected)
5056}
5057
5058fn grant_resource_pattern(resource: &Resource) -> crate::auth::policies::ResourcePattern {
5059    use crate::auth::policies::ResourcePattern;
5060
5061    match resource {
5062        Resource::Database => ResourcePattern::Glob("table:*".to_string()),
5063        Resource::Schema(s) => ResourcePattern::Glob(format!("table:{s}.*")),
5064        Resource::Table { schema, table } => ResourcePattern::Exact {
5065            kind: "table".to_string(),
5066            name: match schema {
5067                Some(s) => format!("{s}.{table}"),
5068                None => table.clone(),
5069            },
5070        },
5071        Resource::Function { schema, name } => ResourcePattern::Exact {
5072            kind: "function".to_string(),
5073            name: match schema {
5074                Some(s) => format!("{s}.{name}"),
5075                None => name.clone(),
5076            },
5077        },
5078    }
5079}
5080
5081// ===========================================================================
5082// ACL serialization helpers — line-oriented, human-readable so an
5083// operator inspecting the vault dump can spot misconfigurations.
5084//
5085// Format (one record per line):
5086//   GRANT|<resource>|<actions_csv>|<with_grant_option>|<tenant_or_*>|<granted_by>|<granted_at>
5087//   ATTR|<valid_until>|<connection_limit>|<search_path>
5088//
5089// Resources are encoded as:
5090//   db                          → Database
5091//   schema:<name>               → Schema(name)
5092//   table:<schema_or_*>:<name>  → Table { schema, table }
5093//   func:<schema_or_*>:<name>   → Function { schema, name }
5094// ===========================================================================
5095
5096fn encode_uid(uid: &UserId) -> String {
5097    match &uid.tenant {
5098        Some(t) => format!("{}/{}", t, uid.username),
5099        None => format!("*/{}", uid.username),
5100    }
5101}
5102
5103fn decode_uid(s: &str) -> Option<UserId> {
5104    let (tenant, username) = s.split_once('/')?;
5105    Some(if tenant == "*" {
5106        UserId::platform(username)
5107    } else {
5108        UserId::scoped(tenant, username)
5109    })
5110}
5111
5112fn encode_resource(r: &Resource) -> String {
5113    match r {
5114        Resource::Database => "db".into(),
5115        Resource::Schema(s) => format!("schema:{}", s),
5116        Resource::Table { schema, table } => {
5117            format!("table:{}:{}", schema.as_deref().unwrap_or("*"), table)
5118        }
5119        Resource::Function { schema, name } => {
5120            format!("func:{}:{}", schema.as_deref().unwrap_or("*"), name)
5121        }
5122    }
5123}
5124
5125fn decode_resource(s: &str) -> Option<Resource> {
5126    if s == "db" {
5127        return Some(Resource::Database);
5128    }
5129    if let Some(rest) = s.strip_prefix("schema:") {
5130        return Some(Resource::Schema(rest.to_string()));
5131    }
5132    if let Some(rest) = s.strip_prefix("table:") {
5133        let (schema, table) = rest.split_once(':')?;
5134        return Some(Resource::Table {
5135            schema: if schema == "*" {
5136                None
5137            } else {
5138                Some(schema.to_string())
5139            },
5140            table: table.to_string(),
5141        });
5142    }
5143    if let Some(rest) = s.strip_prefix("func:") {
5144        let (schema, name) = rest.split_once(':')?;
5145        return Some(Resource::Function {
5146            schema: if schema == "*" {
5147                None
5148            } else {
5149                Some(schema.to_string())
5150            },
5151            name: name.to_string(),
5152        });
5153    }
5154    None
5155}
5156
5157fn encode_grants_blob(grants: &[Grant]) -> String {
5158    let mut out = String::new();
5159    for g in grants {
5160        let actions: Vec<&str> = g.actions.iter().map(|a| a.as_str()).collect();
5161        out.push_str(&format!(
5162            "GRANT|{}|{}|{}|{}|{}|{}\n",
5163            encode_resource(&g.resource),
5164            actions.join(","),
5165            g.with_grant_option,
5166            g.tenant.as_deref().unwrap_or("*"),
5167            g.granted_by,
5168            g.granted_at,
5169        ));
5170    }
5171    out
5172}
5173
5174fn decode_grants_blob(s: &str) -> Option<Vec<Grant>> {
5175    let mut out = Vec::new();
5176    for line in s.lines() {
5177        if line.is_empty() {
5178            continue;
5179        }
5180        let parts: Vec<&str> = line.split('|').collect();
5181        if parts.len() != 7 || parts[0] != "GRANT" {
5182            return None;
5183        }
5184        let resource = decode_resource(parts[1])?;
5185        let mut actions = std::collections::BTreeSet::new();
5186        for token in parts[2].split(',') {
5187            if let Some(a) = Action::from_keyword(token) {
5188                actions.insert(a);
5189            }
5190        }
5191        let with_grant_option = parts[3] == "true";
5192        let tenant = if parts[4] == "*" {
5193            None
5194        } else {
5195            Some(parts[4].to_string())
5196        };
5197        let granted_by = parts[5].to_string();
5198        let granted_at: u128 = parts[6].parse().unwrap_or(0);
5199        out.push(Grant {
5200            // Principal field is reconstructed by the loader from the
5201            // storage-key prefix; default to `Public` here.
5202            principal: GrantPrincipal::Public,
5203            resource,
5204            actions,
5205            with_grant_option,
5206            granted_by,
5207            granted_at,
5208            tenant,
5209            columns: None,
5210        });
5211    }
5212    Some(out)
5213}
5214
5215fn encode_attrs_blob(a: &UserAttributes) -> String {
5216    let valid = a
5217        .valid_until
5218        .map(|t| t.to_string())
5219        .unwrap_or_else(|| "*".into());
5220    let limit = a
5221        .connection_limit
5222        .map(|l| l.to_string())
5223        .unwrap_or_else(|| "*".into());
5224    let path = a.search_path.clone().unwrap_or_else(|| "*".into());
5225    let groups = if a.groups.is_empty() {
5226        "*".to_string()
5227    } else {
5228        a.groups.join(",")
5229    };
5230    format!("ATTR|{}|{}|{}|{}\n", valid, limit, path, groups)
5231}
5232
5233fn decode_attrs_blob(s: &str) -> Option<UserAttributes> {
5234    let line = s.lines().next()?;
5235    let parts: Vec<&str> = line.split('|').collect();
5236    if !(parts.len() == 4 || parts.len() == 5) || parts[0] != "ATTR" {
5237        return None;
5238    }
5239    let groups = if parts.get(4).copied().unwrap_or("*") == "*" {
5240        Vec::new()
5241    } else {
5242        parts[4]
5243            .split(',')
5244            .filter(|g| !g.is_empty())
5245            .map(|g| g.to_string())
5246            .collect()
5247    };
5248    Some(UserAttributes {
5249        valid_until: if parts[1] == "*" {
5250            None
5251        } else {
5252            parts[1].parse().ok()
5253        },
5254        connection_limit: if parts[2] == "*" {
5255            None
5256        } else {
5257            parts[2].parse().ok()
5258        },
5259        search_path: if parts[3] == "*" {
5260            None
5261        } else {
5262            Some(parts[3].to_string())
5263        },
5264        groups,
5265    })
5266}
5267
5268// ===========================================================================
5269// Password hashing
5270// ===========================================================================
5271
5272/// Derive a SCRAM-SHA-256 verifier for a fresh user / password
5273/// rotation. Salt is 16 random bytes; iter is the engine default
5274/// (`scram::DEFAULT_ITER`). Stored alongside the Argon2 password
5275/// hash so HTTP login + v2 SCRAM can both authenticate the same
5276/// user.
5277fn make_scram_verifier(password: &str) -> crate::auth::scram::ScramVerifier {
5278    let salt = random_bytes(16);
5279    crate::auth::scram::ScramVerifier::from_password(
5280        password,
5281        salt,
5282        crate::auth::scram::DEFAULT_ITER,
5283    )
5284}
5285
5286/// Argon2id cost parameters for password hashing, at the OWASP floor
5287/// (19 MiB, t=2, p=1). The previous in-tree KDF ran at 4 MiB / t=3, which
5288/// is below that floor; the parameters are encoded in the PHC string so
5289/// they can be raised again without another format change.
5290const PASSWORD_M_COST_KIB: u32 = 19 * 1024;
5291const PASSWORD_T_COST: u32 = 2;
5292const PASSWORD_P_COST: u32 = 1;
5293const PASSWORD_TAG_LEN: usize = 32;
5294
5295static DUMMY_PASSWORD_HASH: OnceLock<String> = OnceLock::new();
5296
5297fn dummy_password_hash() -> &'static str {
5298    DUMMY_PASSWORD_HASH
5299        .get_or_init(|| hash_password("reddb-invalid-login-dummy"))
5300        .as_str()
5301}
5302
5303fn password_hasher() -> argon2::Argon2<'static> {
5304    let params = argon2::Params::new(
5305        PASSWORD_M_COST_KIB,
5306        PASSWORD_T_COST,
5307        PASSWORD_P_COST,
5308        Some(PASSWORD_TAG_LEN),
5309    )
5310    .expect("Argon2id parameters are constant and valid");
5311    argon2::Argon2::new(argon2::Algorithm::Argon2id, argon2::Version::V0x13, params)
5312}
5313
5314/// Hash a password with Argon2id (RFC 9106) via the RustCrypto `argon2`
5315/// crate, encoded as a PHC string:
5316/// `$argon2id$v=19$m=19456,t=2,p=1$<salt_b64>$<hash_b64>`.
5317///
5318/// The previous in-tree implementation was not RFC 9106 — its `G` mixing
5319/// function added the block indices where the specification calls for the
5320/// `2·lo32(a)·lo32(b)` term, and its compression function was ad hoc — so
5321/// its memory-hardness was unknown. Hashes in the old
5322/// `argon2id$<salt_hex>$<hash_hex>` form still verify (see
5323/// [`verify_password`]) and are transparently upgraded on the owner's next
5324/// successful login.
5325pub(crate) fn hash_password(password: &str) -> String {
5326    use argon2::PasswordHasher;
5327    // The salt comes from the OS CSPRNG inside the hasher and, like
5328    // `random_bytes`, fails closed rather than degrading.
5329    password_hasher()
5330        .hash_password(password.as_bytes())
5331        .expect("Argon2id hashing cannot fail for valid parameters and a working CSPRNG")
5332        .to_string()
5333}
5334
5335/// Whether `stored_hash` was produced by the retired in-tree KDF and should
5336/// be re-hashed the next time its owner authenticates.
5337pub(crate) fn password_hash_needs_upgrade(stored_hash: &str) -> bool {
5338    stored_hash.starts_with("argon2id$")
5339}
5340
5341/// Verify a password against either encoding: the current PHC string, or a
5342/// legacy `argon2id$<salt_hex>$<hash_hex>` produced by the retired in-tree
5343/// KDF. Legacy hashes stay verifiable so an upgrade does not lock every
5344/// account out; [`password_hash_needs_upgrade`] drives the rehash.
5345pub(crate) fn verify_password(password: &str, stored_hash: &str) -> bool {
5346    if let Some(rest) = stored_hash.strip_prefix("argon2id$") {
5347        return verify_legacy_password(password, rest);
5348    }
5349    use argon2::{PasswordHash, PasswordVerifier};
5350    let Ok(parsed) = stored_hash.parse::<PasswordHash>() else {
5351        return false;
5352    };
5353    password_hasher()
5354        .verify_password(password.as_bytes(), &parsed)
5355        .is_ok()
5356}
5357
5358/// Verify against the retired in-tree KDF, whose stored form is
5359/// `<salt_hex>$<hash_hex>` (the `argon2id$` prefix is already stripped).
5360fn verify_legacy_password(password: &str, salt_and_hash: &str) -> bool {
5361    let Some((salt_hex, hash_hex)) = salt_and_hash.split_once('$') else {
5362        return false;
5363    };
5364    let (Ok(salt), Ok(expected)) = (hex::decode(salt_hex), hex::decode(hash_hex)) else {
5365        return false;
5366    };
5367    let computed = derive_key(password.as_bytes(), &salt, &auth_argon2_params());
5368    constant_time_eq(&computed, &expected)
5369}
5370
5371/// Constant-time byte comparison — one implementation for the crate.
5372fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
5373    crate::crypto::constant_time_eq(a, b)
5374}
5375
5376// ===========================================================================
5377// Token generation
5378// ===========================================================================
5379
5380fn generate_session_token() -> String {
5381    format!("rs_{}", hex::encode(random_bytes(32)))
5382}
5383
5384fn generate_api_key() -> String {
5385    format!("rk_{}", hex::encode(random_bytes(32)))
5386}
5387
5388/// Generate `n` random bytes and return as a hex string.
5389fn random_hex(n: usize) -> String {
5390    hex::encode(random_bytes(n))
5391}
5392
5393/// Generate `n` cryptographically random bytes using the OS CSPRNG,
5394/// then mix with SHA-256 for domain separation.
5395pub(crate) fn random_bytes(n: usize) -> Vec<u8> {
5396    let mut buf = vec![0u8; n.max(32)];
5397    // Fail closed: these bytes become session tokens, API keys, salts and
5398    // AES-GCM nonces. A predictable fallback (the old time-derived seed)
5399    // is strictly worse than refusing the operation.
5400    if let Err(err) = os_random::fill_bytes(&mut buf) {
5401        panic!("OS CSPRNG unavailable; refusing to generate credential material: {err}");
5402    }
5403    // SHA-256 mix to ensure uniform distribution.
5404    let digest = sha256(&buf);
5405    if n <= 32 {
5406        digest[..n].to_vec()
5407    } else {
5408        // Chain SHA-256 for longer outputs (unusual but supported).
5409        let mut out = Vec::with_capacity(n);
5410        let mut prev = digest;
5411        while out.len() < n {
5412            out.extend_from_slice(&prev[..std::cmp::min(32, n - out.len())]);
5413            prev = sha256(&prev);
5414        }
5415        out
5416    }
5417}
5418
5419// ===========================================================================
5420// Helpers
5421// ===========================================================================
5422
5423fn lock_err<T>(_: T) -> AuthError {
5424    AuthError::Internal("lock poisoned".to_string())
5425}
5426
5427// ===========================================================================
5428// Tests
5429// ===========================================================================
5430
5431#[cfg(test)]
5432mod tests {
5433    use super::*;
5434
5435    fn test_config() -> AuthConfig {
5436        AuthConfig {
5437            enabled: true,
5438            session_ttl_secs: 60,
5439            require_auth: true,
5440            auto_encrypt_storage: false,
5441            vault_enabled: false,
5442            cert: Default::default(),
5443            oauth: Default::default(),
5444        }
5445    }
5446
5447    #[test]
5448    fn test_create_and_list_users() {
5449        let store = AuthStore::new(test_config());
5450        store.create_user("alice", "pass1", Role::Admin).unwrap();
5451        store.create_user("bob", "pass2", Role::Read).unwrap();
5452
5453        let users = store.list_users();
5454        assert_eq!(users.len(), 2);
5455        // Password hashes should be redacted.
5456        for u in &users {
5457            assert!(u.password_hash.is_empty());
5458        }
5459    }
5460
5461    #[test]
5462    fn test_create_duplicate_user() {
5463        let store = AuthStore::new(test_config());
5464        store.create_user("alice", "pass", Role::Admin).unwrap();
5465        let err = store.create_user("alice", "pass2", Role::Read).unwrap_err();
5466        assert!(matches!(err, AuthError::UserExists(_)));
5467    }
5468
5469    struct RecordingLedger(std::sync::Mutex<Vec<(String, String)>>);
5470
5471    impl crate::runtime::control_events::ControlEventLedger for RecordingLedger {
5472        fn emit(
5473            &self,
5474            _ctx: &ControlEventCtx<'_>,
5475            event: ControlEvent,
5476        ) -> Result<
5477            crate::runtime::control_events::EventId,
5478            crate::runtime::control_events::ControlEventError,
5479        > {
5480            self.0.lock().unwrap().push((
5481                event.kind.as_str().to_string(),
5482                event.outcome.as_str().to_string(),
5483            ));
5484            Ok(crate::runtime::control_events::EventId("evt".to_string()))
5485        }
5486    }
5487
5488    #[test]
5489    fn logins_land_in_the_control_event_ledger() {
5490        let store = AuthStore::new(test_config());
5491        store.create_user("dave", "pw", Role::Read).unwrap();
5492        let ledger = std::sync::Arc::new(RecordingLedger(std::sync::Mutex::new(Vec::new())));
5493        store.configure_control_events(
5494            ledger.clone(),
5495            crate::runtime::control_events::ControlEventConfig::default(),
5496        );
5497        store.authenticate("dave", "pw").expect("login");
5498        store
5499            .authenticate("dave", "nope")
5500            .expect_err("bad password");
5501        store
5502            .authenticate("nobody", "pw")
5503            .expect_err("unknown user");
5504        let seen = ledger.0.lock().unwrap().clone();
5505        let allowed = Outcome::Allowed.as_str().to_string();
5506        let denied = Outcome::Denied.as_str().to_string();
5507        assert!(
5508            seen.contains(&("auth.login".to_string(), allowed)),
5509            "success not recorded: {seen:?}"
5510        );
5511        assert_eq!(
5512            seen.iter()
5513                .filter(|(k, o)| k == "auth.login_failed" && *o == denied)
5514                .count(),
5515            2,
5516            "both refusals must be recorded: {seen:?}"
5517        );
5518    }
5519
5520    #[test]
5521    fn api_keys_are_hashed_at_rest_and_expire() {
5522        let store = AuthStore::new(test_config());
5523        store.create_user("erin", "pw", Role::Write).unwrap();
5524        let key = store
5525            .create_api_key_in_tenant_expiring(None, "erin", "ci", Role::Read, Some(now_ms() + 60))
5526            .unwrap();
5527        assert!(key.key.starts_with("rk_"));
5528
5529        let stored = store
5530            .users
5531            .read()
5532            .unwrap()
5533            .get(&UserId::from_parts(None, "erin"))
5534            .unwrap()
5535            .api_keys[0]
5536            .clone();
5537        assert!(stored.key.starts_with("rkh_"), "{}", stored.key);
5538        assert_ne!(stored.key, key.key, "the secret must not be stored");
5539        assert_eq!(stored.key, api_key_credential_id(&key.key));
5540
5541        assert!(store.validate_token(&key.key).is_some());
5542        std::thread::sleep(std::time::Duration::from_millis(80));
5543        assert!(
5544            store.validate_token(&key.key).is_none(),
5545            "an expired key must not validate"
5546        );
5547
5548        // Revocation accepts the stored id as well as the secret.
5549        store.revoke_api_key(&stored.key).unwrap();
5550        assert!(store.revoke_api_key(&key.key).is_err());
5551    }
5552
5553    #[test]
5554    fn revoke_is_clamped_to_the_granter_tenant() {
5555        use crate::auth::privileges::{Action, GrantPrincipal, Resource};
5556        let store = AuthStore::new(test_config());
5557        let acme_admin = UserId::from_parts(Some("acme"), "adm");
5558        let globex_user = UserId::from_parts(Some("globex"), "u");
5559        let err = store
5560            .revoke(
5561                &acme_admin,
5562                Role::Admin,
5563                &GrantPrincipal::User(globex_user),
5564                &Resource::table_from_name("orders"),
5565                &[Action::Select],
5566            )
5567            .expect_err("a tenant admin must not touch another tenant's grants");
5568        assert!(err.to_string().contains("cross-tenant"), "{err}");
5569        // Platform admin may.
5570        store
5571            .revoke(
5572                &UserId::from_parts(None, "platform"),
5573                Role::Admin,
5574                &GrantPrincipal::User(UserId::from_parts(Some("globex"), "u")),
5575                &Resource::table_from_name("orders"),
5576                &[Action::Select],
5577            )
5578            .expect("platform admin revokes anywhere");
5579    }
5580
5581    #[test]
5582    fn role_change_revokes_sessions_minted_under_the_old_role() {
5583        let store = AuthStore::new(test_config());
5584        store.create_user("carol", "pw", Role::Write).unwrap();
5585        let session = store.authenticate("carol", "pw").unwrap();
5586        assert!(store.validate_token(&session.token).is_some());
5587        store.change_role("carol", Role::Read).unwrap();
5588        assert!(
5589            store.validate_token(&session.token).is_none(),
5590            "a demotion must not keep riding on the old session"
5591        );
5592    }
5593
5594    #[test]
5595    fn test_authenticate_and_validate() {
5596        let store = AuthStore::new(test_config());
5597        store.create_user("alice", "secret", Role::Write).unwrap();
5598
5599        let session = store.authenticate("alice", "secret").unwrap();
5600        assert!(session.token.starts_with("rs_"));
5601
5602        let (username, role) = store.validate_token(&session.token).unwrap();
5603        assert_eq!(username, "alice");
5604        assert_eq!(role, Role::Write);
5605    }
5606
5607    #[test]
5608    fn test_authenticate_wrong_password() {
5609        let store = AuthStore::new(test_config());
5610        store.create_user("alice", "secret", Role::Read).unwrap();
5611
5612        let err = store.authenticate("alice", "wrong").unwrap_err();
5613        assert!(matches!(err, AuthError::InvalidCredentials));
5614    }
5615
5616    #[test]
5617    fn unknown_disabled_and_wrong_password_logins_share_the_public_error() {
5618        let store = AuthStore::new(test_config());
5619        store.create_user("alice", "secret", Role::Read).unwrap();
5620        store.create_user("disabled", "secret", Role::Read).unwrap();
5621        store
5622            .set_user_enabled(&UserId::platform("disabled"), false)
5623            .unwrap();
5624
5625        for result in [
5626            store.authenticate("unknown", "secret"),
5627            store.authenticate("alice", "wrong"),
5628            store.authenticate("disabled", "secret"),
5629        ] {
5630            assert!(matches!(result, Err(AuthError::InvalidCredentials)));
5631        }
5632    }
5633
5634    #[test]
5635    fn test_api_key_lifecycle() {
5636        let store = AuthStore::new(test_config());
5637        store.create_user("alice", "pass", Role::Admin).unwrap();
5638
5639        let key = store
5640            .create_api_key("alice", "ci-token", Role::Write)
5641            .unwrap();
5642        assert!(key.key.starts_with("rk_"));
5643
5644        let (username, role) = store.validate_token(&key.key).unwrap();
5645        assert_eq!(username, "alice");
5646        assert_eq!(role, Role::Write);
5647
5648        store.revoke_api_key(&key.key).unwrap();
5649        assert!(store.validate_token(&key.key).is_none());
5650    }
5651
5652    #[test]
5653    fn test_api_key_role_exceeded() {
5654        let store = AuthStore::new(test_config());
5655        store.create_user("bob", "pass", Role::Read).unwrap();
5656
5657        let err = store
5658            .create_api_key("bob", "escalate", Role::Admin)
5659            .unwrap_err();
5660        assert!(matches!(err, AuthError::RoleExceeded { .. }));
5661    }
5662
5663    #[test]
5664    fn test_change_password() {
5665        let store = AuthStore::new(test_config());
5666        store.create_user("alice", "old", Role::Write).unwrap();
5667
5668        store.change_password("alice", "old", "new").unwrap();
5669
5670        // Old password should fail.
5671        assert!(store.authenticate("alice", "old").is_err());
5672        // New password should succeed.
5673        assert!(store.authenticate("alice", "new").is_ok());
5674    }
5675
5676    #[test]
5677    fn test_change_role() {
5678        let store = AuthStore::new(test_config());
5679        store.create_user("alice", "pass", Role::Admin).unwrap();
5680        store.create_api_key("alice", "key1", Role::Admin).unwrap();
5681
5682        store.change_role("alice", Role::Read).unwrap();
5683
5684        // User's role should be Read now.
5685        let users = store.list_users();
5686        let alice = users.iter().find(|u| u.username == "alice").unwrap();
5687        assert_eq!(alice.role, Role::Read);
5688
5689        // API keys should have been downgraded.
5690        assert_eq!(alice.api_keys[0].role, Role::Read);
5691    }
5692
5693    #[test]
5694    fn test_admin_user_mutations_are_policy_controlled_not_flag_guarded() {
5695        let store = AuthStore::new(test_config());
5696        store
5697            .create_admin_user("system", "pass", Role::Admin, None)
5698            .unwrap();
5699
5700        let uid = UserId::platform("system");
5701        store.change_password("system", "pass", "new").unwrap();
5702        store.change_role("system", Role::Read).unwrap();
5703        store.set_user_enabled(&uid, false).unwrap();
5704
5705        let key = store
5706            .create_api_key("system", "rotation", Role::Read)
5707            .unwrap();
5708        // The account is disabled at this point: every credential it owns
5709        // must stop validating, API keys included (they used to be resolved
5710        // straight from the index without consulting the user record).
5711        assert!(store.validate_token(&key.key).is_none());
5712        store.set_user_enabled(&uid, true).unwrap();
5713        assert!(store.validate_token(&key.key).is_some());
5714        store.revoke_api_key(&key.key).unwrap();
5715        assert!(store.validate_token(&key.key).is_none());
5716        store.delete_user("system").unwrap();
5717    }
5718
5719    #[test]
5720    fn test_user_lifecycle_deny_policy_blocks_cloud_admin_deletion() {
5721        use crate::auth::policies::Policy;
5722
5723        let store = AuthStore::new(test_config());
5724        store
5725            .create_user("cloud-admin", "pass", Role::Admin)
5726            .unwrap();
5727        store
5728            .create_user("customer-admin", "pass", Role::Admin)
5729            .unwrap();
5730
5731        store
5732            .put_policy(
5733                Policy::from_json_str(
5734                    r#"{
5735                        "id": "customer-admin-allow-all",
5736                        "version": 1,
5737                        "statements": [{
5738                            "effect": "allow",
5739                            "actions": ["*"],
5740                            "resources": ["*"]
5741                        }]
5742                    }"#,
5743                )
5744                .unwrap(),
5745            )
5746            .unwrap();
5747        store
5748            .put_policy(
5749                Policy::from_json_str(
5750                    r#"{
5751                        "id": "cloud-admin-protection",
5752                        "version": 1,
5753                        "statements": [{
5754                            "effect": "deny",
5755                            "actions": [
5756                                "user:delete",
5757                                "user:disable",
5758                                "user:password:change",
5759                                "user:role:update"
5760                            ],
5761                            "resources": ["user:cloud-admin"]
5762                        }]
5763                    }"#,
5764                )
5765                .unwrap(),
5766            )
5767            .unwrap();
5768        let customer = UserId::platform("customer-admin");
5769        store
5770            .attach_policy(
5771                PrincipalRef::User(customer.clone()),
5772                "customer-admin-allow-all",
5773            )
5774            .unwrap();
5775        store
5776            .attach_policy(
5777                PrincipalRef::User(customer.clone()),
5778                "cloud-admin-protection",
5779            )
5780            .unwrap();
5781
5782        let cloud_admin = UserId::platform("cloud-admin");
5783        assert!(!store.check_user_lifecycle_authz(
5784            &customer,
5785            Role::Admin,
5786            "user:delete",
5787            &cloud_admin,
5788        ));
5789        assert!(!store.check_user_lifecycle_authz(
5790            &customer,
5791            Role::Admin,
5792            "user:disable",
5793            &cloud_admin,
5794        ));
5795        assert!(!store.check_user_lifecycle_authz(
5796            &customer,
5797            Role::Admin,
5798            "user:password:change",
5799            &cloud_admin,
5800        ));
5801        assert!(!store.check_user_lifecycle_authz(
5802            &customer,
5803            Role::Admin,
5804            "user:role:update",
5805            &cloud_admin,
5806        ));
5807
5808        let another_user = UserId::platform("someone-else");
5809        assert!(store.check_user_lifecycle_authz(
5810            &customer,
5811            Role::Admin,
5812            "user:delete",
5813            &another_user,
5814        ));
5815    }
5816
5817    #[test]
5818    fn admin_user_lifecycle_keeps_role_baseline_in_policy_only_mode() {
5819        let store = AuthStore::new(test_config());
5820        store.set_enforcement_mode(PolicyEnforcementMode::PolicyOnly);
5821        store.create_user("admin", "pass", Role::Admin).unwrap();
5822        store.create_user("alice", "pass", Role::Read).unwrap();
5823
5824        assert!(store.check_user_lifecycle_authz(
5825            &UserId::platform("admin"),
5826            Role::Admin,
5827            "user:update",
5828            &UserId::platform("alice"),
5829        ));
5830    }
5831
5832    #[test]
5833    fn test_regular_user_mutations_still_work() {
5834        let store = AuthStore::new(test_config());
5835        store.create_user("alice", "old", Role::Admin).unwrap();
5836
5837        let uid = UserId::platform("alice");
5838        store.set_user_enabled(&uid, false).unwrap();
5839        assert!(matches!(
5840            store.authenticate("alice", "old"),
5841            Err(AuthError::InvalidCredentials)
5842        ));
5843
5844        store.set_user_enabled(&uid, true).unwrap();
5845        store.change_password("alice", "old", "new").unwrap();
5846        store.change_role("alice", Role::Read).unwrap();
5847        store.delete_user("alice").unwrap();
5848        assert!(matches!(
5849            store.authenticate("alice", "new"),
5850            Err(AuthError::InvalidCredentials)
5851        ));
5852    }
5853
5854    #[test]
5855    fn test_delete_user() {
5856        let store = AuthStore::new(test_config());
5857        store.create_user("alice", "pass", Role::Admin).unwrap();
5858        let key = store.create_api_key("alice", "key1", Role::Read).unwrap();
5859        let session = store.authenticate("alice", "pass").unwrap();
5860
5861        store.delete_user("alice").unwrap();
5862
5863        assert!(store.validate_token(&key.key).is_none());
5864        assert!(store.validate_token(&session.token).is_none());
5865        assert!(store.list_users().is_empty());
5866    }
5867
5868    #[test]
5869    fn test_revoke_session() {
5870        let store = AuthStore::new(test_config());
5871        store.create_user("alice", "pass", Role::Read).unwrap();
5872        let session = store.authenticate("alice", "pass").unwrap();
5873
5874        store.revoke_session(&session.token);
5875        assert!(store.validate_token(&session.token).is_none());
5876    }
5877
5878    #[test]
5879    fn test_password_hash_format() {
5880        // PHC string: `$argon2id$v=19$m=19456,t=2,p=1$<salt_b64>$<hash_b64>`.
5881        // Encoding the parameters is what lets the cost be raised later
5882        // without another format change — the retired in-tree form stored
5883        // only salt and tag, so its 4 MiB / t=3 cost was frozen.
5884        let hash = hash_password("test");
5885        assert!(
5886            hash.starts_with("$argon2id$v=19$"),
5887            "expected a PHC-encoded Argon2id hash, got: {hash}"
5888        );
5889        assert!(
5890            hash.contains(&format!(
5891                "m={PASSWORD_M_COST_KIB},t={PASSWORD_T_COST},p={PASSWORD_P_COST}"
5892            )),
5893            "the PHC string must carry the cost parameters: {hash}"
5894        );
5895        assert!(verify_password("test", &hash));
5896        assert!(!verify_password("wrong", &hash));
5897        assert!(
5898            !password_hash_needs_upgrade(&hash),
5899            "a freshly minted hash must not be flagged for upgrade"
5900        );
5901    }
5902
5903    #[test]
5904    fn legacy_password_hashes_still_verify_and_upgrade_on_login() {
5905        // Records written by the retired in-tree KDF must keep verifying —
5906        // otherwise the upgrade locks every existing account out — and must
5907        // be re-hashed with the real KDF once the plaintext is in hand.
5908        let legacy = {
5909            let salt = random_bytes(16);
5910            let hash = derive_key(b"hunter2", &salt, &auth_argon2_params());
5911            format!("argon2id${}${}", hex::encode(&salt), hex::encode(&hash))
5912        };
5913        assert!(verify_password("hunter2", &legacy));
5914        assert!(!verify_password("wrong", &legacy));
5915        assert!(password_hash_needs_upgrade(&legacy));
5916
5917        let store = AuthStore::new(test_config());
5918        store
5919            .create_user("erin", "placeholder", Role::Write)
5920            .unwrap();
5921        {
5922            let mut users = store.users.write().unwrap();
5923            users
5924                .get_mut(&UserId::platform("erin"))
5925                .unwrap()
5926                .password_hash = legacy.clone();
5927        }
5928
5929        store
5930            .authenticate("erin", "hunter2")
5931            .expect("a legacy hash must still authenticate");
5932
5933        let stored = store
5934            .users
5935            .read()
5936            .unwrap()
5937            .get(&UserId::platform("erin"))
5938            .unwrap()
5939            .password_hash
5940            .clone();
5941        assert!(
5942            stored.starts_with("$argon2id$v=19$"),
5943            "a successful login must upgrade the stored hash, got: {stored}"
5944        );
5945        assert!(store.authenticate("erin", "hunter2").is_ok());
5946    }
5947
5948    #[test]
5949    fn test_constant_time_eq() {
5950        assert!(constant_time_eq(b"hello", b"hello"));
5951        assert!(!constant_time_eq(b"hello", b"world"));
5952        assert!(!constant_time_eq(b"short", b"longer"));
5953    }
5954
5955    #[test]
5956    fn test_bootstrap_seals_permanently() {
5957        let store = AuthStore::new(test_config());
5958
5959        assert!(store.needs_bootstrap());
5960        assert!(!store.is_bootstrapped());
5961
5962        // First bootstrap succeeds
5963        let result = store.bootstrap("admin", "secret");
5964        assert!(result.is_ok());
5965        let br = result.unwrap();
5966        assert_eq!(br.user.username, "admin");
5967        assert_eq!(br.user.role, Role::Admin);
5968        assert!(br.api_key.key.starts_with("rk_"));
5969        // No vault configured, so no certificate.
5970        assert!(br.certificate.is_none());
5971
5972        // Sealed now
5973        assert!(!store.needs_bootstrap());
5974        assert!(store.is_bootstrapped());
5975
5976        // Second bootstrap fails -- sealed permanently
5977        let result = store.bootstrap("admin2", "secret2");
5978        assert!(result.is_err());
5979        let err = result.unwrap_err();
5980        assert!(err.to_string().contains("sealed permanently"));
5981
5982        // Only 1 user exists (the first one)
5983        assert_eq!(store.list_users().len(), 1);
5984        assert_eq!(store.list_users()[0].username, "admin");
5985    }
5986
5987    #[test]
5988    fn certificate_delivery_failure_precedes_vault_activation() {
5989        use crate::storage::engine::pager::PagerConfig;
5990        use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
5991
5992        static COUNTER: AtomicU64 = AtomicU64::new(0);
5993        let id = COUNTER.fetch_add(1, AtomicOrdering::Relaxed);
5994        let directory = std::env::temp_dir().join(format!(
5995            "reddb-auth-delivery-failure-{}-{id}",
5996            std::process::id()
5997        ));
5998        std::fs::create_dir_all(&directory).expect("temp directory");
5999        let pager = Arc::new(
6000            Pager::open(directory.join("test.rdb"), PagerConfig::default()).expect("pager"),
6001        );
6002        let mut config = test_config();
6003        config.vault_enabled = true;
6004        let store = AuthStore::with_vault_optional_certificate(config, Arc::clone(&pager), None)
6005            .expect("fresh vault store");
6006
6007        let error = store
6008            .bootstrap_with_certificate_delivery("admin", "secret", |_| {
6009                Err(AuthError::Internal(
6010                    "simulated delivery failure".to_string(),
6011                ))
6012            })
6013            .expect_err("delivery failure must abort bootstrap");
6014        assert!(error.to_string().contains("simulated delivery failure"));
6015        assert!(store.needs_bootstrap());
6016        assert!(store.list_users().is_empty());
6017        assert!(!Vault::has_saved_state(&pager));
6018
6019        drop(store);
6020        drop(pager);
6021        let _ = std::fs::remove_dir_all(directory);
6022    }
6023
6024    #[test]
6025    fn test_bootstrap_after_manual_user_creation() {
6026        let store = AuthStore::new(test_config());
6027
6028        // Create a user manually first
6029        store.create_user("existing", "pass", Role::Read).unwrap();
6030
6031        // Bootstrap sees the seal hasn't been set but users exist
6032        // The atomic seal fires first, then the users check catches it
6033        assert!(!store.needs_bootstrap()); // users exist → false
6034    }
6035
6036    // ---------------------------------------------------------------
6037    // Tenant scoping
6038    // ---------------------------------------------------------------
6039
6040    #[test]
6041    fn test_same_username_two_tenants_distinct() {
6042        let store = AuthStore::new(test_config());
6043        store
6044            .create_user_in_tenant(Some("acme"), "alice", "pw-acme", Role::Write)
6045            .unwrap();
6046        store
6047            .create_user_in_tenant(Some("globex"), "alice", "pw-globex", Role::Read)
6048            .unwrap();
6049
6050        // Two distinct users.
6051        let users = store.list_users();
6052        assert_eq!(users.len(), 2);
6053
6054        // Each verifies its own password under its own tenant.
6055        assert!(store
6056            .authenticate_in_tenant(Some("acme"), "alice", "pw-acme")
6057            .is_ok());
6058        assert!(store
6059            .authenticate_in_tenant(Some("globex"), "alice", "pw-globex")
6060            .is_ok());
6061
6062        // Cross-tenant credentials are rejected.
6063        assert!(store
6064            .authenticate_in_tenant(Some("acme"), "alice", "pw-globex")
6065            .is_err());
6066        assert!(store
6067            .authenticate_in_tenant(Some("globex"), "alice", "pw-acme")
6068            .is_err());
6069    }
6070
6071    #[test]
6072    fn test_session_carries_tenant() {
6073        let store = AuthStore::new(test_config());
6074        store
6075            .create_user_in_tenant(Some("acme"), "alice", "pw", Role::Admin)
6076            .unwrap();
6077        let session = store
6078            .authenticate_in_tenant(Some("acme"), "alice", "pw")
6079            .unwrap();
6080        assert_eq!(session.tenant_id.as_deref(), Some("acme"));
6081
6082        let (id, role) = store.validate_token_full(&session.token).unwrap();
6083        assert_eq!(id.tenant.as_deref(), Some("acme"));
6084        assert_eq!(id.username, "alice");
6085        assert_eq!(role, Role::Admin);
6086    }
6087
6088    #[test]
6089    fn test_platform_user_has_no_tenant() {
6090        let store = AuthStore::new(test_config());
6091        store.create_user("admin", "pw", Role::Admin).unwrap();
6092        let session = store.authenticate("admin", "pw").unwrap();
6093        assert!(session.tenant_id.is_none());
6094
6095        let (id, _) = store.validate_token_full(&session.token).unwrap();
6096        assert!(id.tenant.is_none());
6097    }
6098
6099    #[test]
6100    fn test_lookup_scram_verifier_global_resolves_platform() {
6101        let store = AuthStore::new(test_config());
6102        store.create_user("admin", "pw", Role::Admin).unwrap();
6103        store
6104            .create_user_in_tenant(Some("acme"), "admin", "pw", Role::Admin)
6105            .unwrap();
6106
6107        // The global helper picks the platform-tenant user only.
6108        let v = store.lookup_scram_verifier_global("admin");
6109        assert!(v.is_some());
6110
6111        // The tenant-scoped user has its own verifier.
6112        let v_acme = store.lookup_scram_verifier(&UserId::scoped("acme", "admin"));
6113        assert!(v_acme.is_some());
6114
6115        // The two verifiers carry independent salts.
6116        assert_ne!(v.unwrap().salt, v_acme.unwrap().salt);
6117    }
6118
6119    #[test]
6120    fn test_delete_in_tenant_does_not_touch_other_tenant() {
6121        let store = AuthStore::new(test_config());
6122        store
6123            .create_user_in_tenant(Some("acme"), "alice", "pw", Role::Admin)
6124            .unwrap();
6125        store
6126            .create_user_in_tenant(Some("globex"), "alice", "pw", Role::Admin)
6127            .unwrap();
6128
6129        store.delete_user_in_tenant(Some("acme"), "alice").unwrap();
6130
6131        // Globex still alive.
6132        assert!(store
6133            .authenticate_in_tenant(Some("globex"), "alice", "pw")
6134            .is_ok());
6135        // Acme gone.
6136        assert!(store
6137            .authenticate_in_tenant(Some("acme"), "alice", "pw")
6138            .is_err());
6139    }
6140
6141    #[test]
6142    fn test_user_id_display() {
6143        assert_eq!(UserId::platform("admin").to_string(), "admin");
6144        assert_eq!(UserId::scoped("acme", "alice").to_string(), "acme/alice");
6145    }
6146
6147    // -----------------------------------------------------------------
6148    // PolicyEnforcementMode wiring (#712 / S5A)
6149    // -----------------------------------------------------------------
6150
6151    fn enforcement_eval_ctx(role: Role) -> EvalContext {
6152        EvalContext {
6153            principal_tenant: None,
6154            current_tenant: None,
6155            peer_ip: None,
6156            mfa_present: false,
6157            now_ms: 1_700_000_000_000,
6158            principal_is_admin_role: role == Role::Admin,
6159            principal_is_platform_scoped: true,
6160        }
6161    }
6162
6163    /// Helper: install a single unrelated allow policy so the IAM
6164    /// path is the authoritative one — without any installed policy
6165    /// the auth flow short-circuits before we ever consult the mode.
6166    fn install_unrelated_policy(store: &AuthStore) {
6167        store
6168            .put_policy(
6169                Policy::from_json_str(
6170                    r#"{"id":"p-unrelated","version":1,"statements":[{"effect":"allow","actions":["select"],"resources":["table:public.other"]}]}"#,
6171                )
6172                .unwrap(),
6173            )
6174            .unwrap();
6175    }
6176
6177    #[test]
6178    fn enforcement_mode_default_for_new_store_is_legacy_rbac() {
6179        // Pre-bootstrap construction = "existing install" path.
6180        // Defaulting to LegacyRbac preserves pre-#712 behaviour on
6181        // upgrade so an operator that has not yet attached IAM
6182        // policies does not lose access after the upgrade.
6183        let store = AuthStore::new(test_config());
6184        assert_eq!(store.enforcement_mode(), PolicyEnforcementMode::LegacyRbac);
6185    }
6186
6187    #[test]
6188    fn enforcement_mode_set_round_trips_and_returns_previous() {
6189        let store = AuthStore::new(test_config());
6190        let prev = store.set_enforcement_mode(PolicyEnforcementMode::PolicyOnly);
6191        assert_eq!(prev, PolicyEnforcementMode::LegacyRbac);
6192        assert_eq!(store.enforcement_mode(), PolicyEnforcementMode::PolicyOnly);
6193        let prev = store.set_enforcement_mode(PolicyEnforcementMode::LegacyRbac);
6194        assert_eq!(prev, PolicyEnforcementMode::PolicyOnly);
6195    }
6196
6197    #[test]
6198    fn policy_only_mode_treats_no_matching_policy_as_deny() {
6199        // Acceptance #2: in `policy_only` mode a principal with no
6200        // matching policy is denied even when the action's role
6201        // floor would otherwise have allowed it.
6202        let store = AuthStore::new(test_config());
6203        store.create_user("alice", "p", Role::Admin).unwrap();
6204        let uid = UserId::platform("alice");
6205        install_unrelated_policy(&store);
6206        store.set_enforcement_mode(PolicyEnforcementMode::PolicyOnly);
6207
6208        let resource = ResourceRef::new("table", "public.t");
6209        // Even Role::Admin is denied — DefaultDeny is the deciding
6210        // signal in `policy_only`, regardless of role.
6211        assert!(!store.check_policy_authz_with_role(
6212            &uid,
6213            "select",
6214            &resource,
6215            &enforcement_eval_ctx(Role::Admin),
6216            Role::Admin,
6217        ));
6218    }
6219
6220    #[test]
6221    fn legacy_rbac_mode_falls_back_to_role_decision() {
6222        // Acceptance #3: in `legacy_rbac` mode the same principal
6223        // falls through to the role-based decision. `select` requires
6224        // only `Read`, so all three roles satisfy it; a write verb
6225        // requires `Write`, so `Read` does not.
6226        let store = AuthStore::new(test_config());
6227        store.create_user("alice", "p", Role::Read).unwrap();
6228        let uid = UserId::platform("alice");
6229        install_unrelated_policy(&store);
6230        store.set_enforcement_mode(PolicyEnforcementMode::LegacyRbac);
6231
6232        let table = ResourceRef::new("table", "public.t");
6233        // Read role + select action: allowed via the legacy fallback.
6234        assert!(store.check_policy_authz_with_role(
6235            &uid,
6236            "select",
6237            &table,
6238            &enforcement_eval_ctx(Role::Read),
6239            Role::Read,
6240        ));
6241        // Read role + insert action: legacy decision says no.
6242        assert!(!store.check_policy_authz_with_role(
6243            &uid,
6244            "insert",
6245            &table,
6246            &enforcement_eval_ctx(Role::Read),
6247            Role::Read,
6248        ));
6249        // Admin role + insert action: legacy decision says yes.
6250        assert!(store.check_policy_authz_with_role(
6251            &uid,
6252            "insert",
6253            &table,
6254            &enforcement_eval_ctx(Role::Admin),
6255            Role::Admin,
6256        ));
6257    }
6258
6259    #[test]
6260    fn explicit_deny_wins_irrespective_of_mode_or_role() {
6261        // Even in `legacy_rbac` with an Admin role, an explicit
6262        // matching Deny statement always denies. The mode only
6263        // influences the `DefaultDeny` arm, never an explicit one.
6264        let store = AuthStore::new(test_config());
6265        store.create_user("alice", "p", Role::Admin).unwrap();
6266        let uid = UserId::platform("alice");
6267        store
6268            .put_policy(
6269                Policy::from_json_str(
6270                    r#"{"id":"p-deny-select","version":1,"statements":[{"effect":"deny","actions":["select"],"resources":["table:public.t"]}]}"#,
6271                )
6272                .unwrap(),
6273            )
6274            .unwrap();
6275        store
6276            .attach_policy(PrincipalRef::User(uid.clone()), "p-deny-select")
6277            .unwrap();
6278
6279        let resource = ResourceRef::new("table", "public.t");
6280        for mode in [
6281            PolicyEnforcementMode::LegacyRbac,
6282            PolicyEnforcementMode::PolicyOnly,
6283        ] {
6284            store.set_enforcement_mode(mode);
6285            assert!(
6286                !store.check_policy_authz_with_role(
6287                    &uid,
6288                    "select",
6289                    &resource,
6290                    &enforcement_eval_ctx(Role::Admin),
6291                    Role::Admin,
6292                ),
6293                "explicit deny must win under mode {mode:?}"
6294            );
6295        }
6296    }
6297
6298    #[test]
6299    fn explicit_allow_wins_irrespective_of_mode_or_role() {
6300        // Symmetric guarantee: an explicit Allow is honoured even in
6301        // `policy_only` mode for a role that the legacy fallback
6302        // would have rejected (Read role + admin-tier action).
6303        let store = AuthStore::new(test_config());
6304        store.create_user("alice", "p", Role::Read).unwrap();
6305        let uid = UserId::platform("alice");
6306        store
6307            .put_policy(
6308                Policy::from_json_str(
6309                    r#"{"id":"p-allow-create","version":1,"statements":[{"effect":"allow","actions":["create"],"resources":["table:public.t"]}]}"#,
6310                )
6311                .unwrap(),
6312            )
6313            .unwrap();
6314        store
6315            .attach_policy(PrincipalRef::User(uid.clone()), "p-allow-create")
6316            .unwrap();
6317        store.set_enforcement_mode(PolicyEnforcementMode::PolicyOnly);
6318
6319        let resource = ResourceRef::new("table", "public.t");
6320        assert!(store.check_policy_authz_with_role(
6321            &uid,
6322            "create",
6323            &resource,
6324            &enforcement_eval_ctx(Role::Read),
6325            Role::Read,
6326        ));
6327    }
6328
6329    #[test]
6330    fn take_legacy_rbac_warn_once_is_at_most_once_per_boot() {
6331        let store = AuthStore::new(test_config());
6332        // Default mode = LegacyRbac, so the first call wins the token.
6333        assert!(store.take_legacy_rbac_warn_once());
6334        // Every subsequent call is a no-op for the lifetime of this
6335        // process / AuthStore — that is the "once per boot" guarantee.
6336        for _ in 0..3 {
6337            assert!(!store.take_legacy_rbac_warn_once());
6338        }
6339    }
6340
6341    #[test]
6342    fn take_legacy_rbac_warn_once_is_silent_under_policy_only() {
6343        // Acceptance #5 phrases the warning as "Boot in legacy_rbac
6344        // mode emits exactly one warn entry per boot" — implicitly,
6345        // `policy_only` boots emit zero. The token must not be
6346        // claimable when the mode is strict so the boot path stays
6347        // quiet.
6348        let store = AuthStore::new(test_config());
6349        store.set_enforcement_mode(PolicyEnforcementMode::PolicyOnly);
6350        assert!(!store.take_legacy_rbac_warn_once());
6351        // And, importantly, switching back to LegacyRbac later does
6352        // not "owe" us a warning — the token was never claimed under
6353        // policy_only, so it is still available the first time the
6354        // mode is legacy_rbac.
6355        store.set_enforcement_mode(PolicyEnforcementMode::LegacyRbac);
6356        assert!(store.take_legacy_rbac_warn_once());
6357        assert!(!store.take_legacy_rbac_warn_once());
6358    }
6359
6360    #[test]
6361    fn strict_check_policy_authz_ignores_enforcement_mode() {
6362        // Governance APIs (managed_config, registry, managed_policy)
6363        // call the strict `check_policy_authz` — it must return
6364        // `false` on DefaultDeny regardless of mode so the lenient
6365        // posture cannot accidentally elevate an admin-tier mutation.
6366        let store = AuthStore::new(test_config());
6367        store.create_user("alice", "p", Role::Admin).unwrap();
6368        let uid = UserId::platform("alice");
6369        install_unrelated_policy(&store);
6370        store.set_enforcement_mode(PolicyEnforcementMode::LegacyRbac);
6371
6372        let resource = ResourceRef::new("table", "public.t");
6373        assert!(!store.check_policy_authz(
6374            &uid,
6375            "select",
6376            &resource,
6377            &enforcement_eval_ctx(Role::Admin),
6378        ));
6379    }
6380
6381    #[test]
6382    fn disabling_a_user_invalidates_its_api_keys_and_sessions() {
6383        let store = AuthStore::new(test_config());
6384        store.create_user("carol", "pw", Role::Write).unwrap();
6385        let key = store
6386            .create_api_key("carol", "ci", Role::Write)
6387            .unwrap()
6388            .key;
6389        let session = store.authenticate("carol", "pw").unwrap();
6390        assert!(store.validate_token(&key).is_some());
6391        assert!(store.validate_token(&session.token).is_some());
6392
6393        let uid = UserId::platform("carol");
6394        store.set_user_enabled(&uid, false).unwrap();
6395
6396        // API keys resolved straight from the index and never consulted the
6397        // user record, so a disabled (or offboarded) account kept a working
6398        // credential forever.
6399        assert!(
6400            store.validate_token(&key).is_none(),
6401            "an API key must stop working the moment its owner is disabled"
6402        );
6403        assert!(
6404            store.validate_token(&session.token).is_none(),
6405            "live sessions must not survive disabling the account"
6406        );
6407        assert!(
6408            store.lookup_scram_verifier(&uid).is_none(),
6409            "SCRAM must not authenticate a disabled account"
6410        );
6411    }
6412
6413    #[test]
6414    fn admin_password_reset_applies_and_revokes_sessions() {
6415        let store = AuthStore::new(test_config());
6416        store.create_user("dave", "old", Role::Write).unwrap();
6417        let session = store.authenticate("dave", "old").unwrap();
6418
6419        store
6420            .set_password_by_admin(&UserId::platform("dave"), "rotated")
6421            .expect("admin reset");
6422
6423        assert!(
6424            store.authenticate("dave", "old").is_err(),
6425            "the old password must stop working after a reset"
6426        );
6427        assert!(
6428            store.authenticate("dave", "rotated").is_ok(),
6429            "the new password must work after a reset"
6430        );
6431        assert!(
6432            store.validate_token(&session.token).is_none(),
6433            "a password rotation must revoke live sessions"
6434        );
6435    }
6436}