reddb_server/runtime/
control_events.rs

1//! Control Event Ledger — skeleton (issue #652).
2//!
3//! Cross-cutting types + the [`ControlEventLedger`] trait that the
4//! policy / config / user-lifecycle producer slices (issues 665/666/
5//! 667) will call into. Ships ONE implementor — [`RuntimeLedger`] —
6//! which writes one row per `emit()` to the `red.control_events`
7//! collection via the unified entity API.
8//!
9//! This module deliberately does NOT wire `emit()` into any producer
10//! call site (`AuthStore::*`, `ConfigRegistry::*`, etc.); that is the
11//! scope of 652b/c/d. It also does not decide what counts as
12//! sensitive — producers call [`Sensitivity::hashed`] /
13//! [`Sensitivity::redacted`] at their own emit sites.
14
15use std::borrow::Cow;
16use std::collections::HashMap;
17use std::sync::Arc;
18
19use crate::auth::UserId;
20use crate::crypto::uuid::Uuid;
21use crate::storage::unified::{EntityData, EntityId, EntityKind, RowData, UnifiedEntity};
22use crate::storage::UnifiedStore;
23use crate::utils::now_unix_millis;
24use reddb_types::encoding::json_escape;
25use reddb_types::types::Value;
26
27/// Canonical name of the managed control-event collection.
28pub const CONTROL_EVENTS_COLLECTION: &str = "red.control_events";
29
30// ---------------------------------------------------------------------------
31// EventKind
32// ---------------------------------------------------------------------------
33
34/// Strong enum of every control-plane mutation the ledger records.
35/// Mirrors the `kind` column in `red.control_events`.
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
37pub enum EventKind {
38    PolicyCreate,
39    PolicyUpdate,
40    PolicyDelete,
41    PolicyAttach,
42    PolicyDetach,
43    ConfigWrite,
44    ConfigDelete,
45    UserCreate,
46    UserUpdate,
47    UserDelete,
48    UserDisable,
49    ApiKeyCreate,
50    ApiKeyRevoke,
51    VaultMetadataRead,
52    VaultRead,
53    VaultUnseal,
54    VaultReveal,
55    VaultUse,
56    VaultRotate,
57    VaultPurge,
58    SchemaDdl,
59    TenantGovernance,
60    RlsGovernance,
61    BackupRun,
62    RestoreRun,
63    FailoverPromotion,
64    ReplicationSafety,
65    EvidenceExport,
66    /// Emitted at startup when the `REDDB_POLICY_BREAK_GLASS` env var
67    /// triggers the [`crate::auth::self_lock_guard`] recovery path —
68    /// see issue #713.
69    PolicyBreakGlass,
70    /// A password login succeeded.
71    AuthLogin,
72    /// A password login was refused (unknown principal, disabled account,
73    /// wrong password). The reason stays generic on purpose.
74    AuthLoginFailed,
75    /// The server booted with authentication disabled (`--no-auth` /
76    /// `--dev`).
77    AuthDisabled,
78}
79
80impl EventKind {
81    pub fn as_str(self) -> &'static str {
82        match self {
83            Self::PolicyCreate => "policy.create",
84            Self::PolicyUpdate => "policy.update",
85            Self::PolicyDelete => "policy.delete",
86            Self::PolicyAttach => "policy.attach",
87            Self::PolicyDetach => "policy.detach",
88            Self::ConfigWrite => "config.write",
89            Self::ConfigDelete => "config.delete",
90            Self::UserCreate => "user.create",
91            Self::UserUpdate => "user.update",
92            Self::UserDelete => "user.delete",
93            Self::UserDisable => "user.disable",
94            Self::ApiKeyCreate => "apikey.create",
95            Self::ApiKeyRevoke => "apikey.revoke",
96            Self::VaultMetadataRead => "vault.metadata_read",
97            Self::VaultRead => "vault.read",
98            Self::VaultUnseal => "vault.unseal",
99            Self::VaultReveal => "vault.reveal",
100            Self::VaultUse => "vault.use",
101            Self::VaultRotate => "vault.rotate",
102            Self::VaultPurge => "vault.purge",
103            Self::SchemaDdl => "schema.ddl",
104            Self::TenantGovernance => "tenant.governance",
105            Self::RlsGovernance => "rls.governance",
106            Self::BackupRun => "backup.run",
107            Self::RestoreRun => "restore.run",
108            Self::FailoverPromotion => "failover.promotion",
109            Self::ReplicationSafety => "replication.safety",
110            Self::EvidenceExport => "evidence.export",
111            Self::PolicyBreakGlass => "policy.break_glass",
112            Self::AuthLogin => "auth.login",
113            Self::AuthLoginFailed => "auth.login_failed",
114            Self::AuthDisabled => "auth.disabled",
115        }
116    }
117}
118
119// ---------------------------------------------------------------------------
120// Outcome
121// ---------------------------------------------------------------------------
122
123#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
124pub enum Outcome {
125    Allowed,
126    Denied,
127    Error,
128}
129
130impl Outcome {
131    pub fn as_str(self) -> &'static str {
132        match self {
133            Self::Allowed => "allowed",
134            Self::Denied => "denied",
135            Self::Error => "error",
136        }
137    }
138}
139
140// ---------------------------------------------------------------------------
141// ActorRef / Context
142// ---------------------------------------------------------------------------
143
144/// Who attempted the mutation. Borrowed so producer call-sites don't
145/// allocate at every emit; the ledger copies into the persisted row.
146#[derive(Debug)]
147pub enum ActorRef<'a> {
148    User(&'a UserId),
149    /// A static system label (e.g. `"bootstrap"`, `"wal_replay"`).
150    /// Static-lifetime to keep the enum cheap and to disambiguate from
151    /// user identifiers — never a runtime tenant/user string.
152    System(&'static str),
153    Anonymous,
154}
155
156impl<'a> ActorRef<'a> {
157    pub fn user_id(&self) -> Option<&UserId> {
158        match self {
159            Self::User(u) => Some(u),
160            _ => None,
161        }
162    }
163
164    pub fn kind_str(&self) -> &'static str {
165        match self {
166            Self::User(_) => "user",
167            Self::System(_) => "system",
168            Self::Anonymous => "anonymous",
169        }
170    }
171}
172
173/// Request-scoped context attached to every emit. Producer call-sites
174/// fill what they have; missing fields land as `Null` in the row.
175pub struct ControlEventCtx<'a> {
176    pub actor: ActorRef<'a>,
177    pub scope: Option<Cow<'a, str>>,
178    pub request_id: Option<Cow<'a, str>>,
179    pub trace_id: Option<Cow<'a, str>>,
180}
181
182// ---------------------------------------------------------------------------
183// ControlEvent
184// ---------------------------------------------------------------------------
185
186pub struct ControlEvent {
187    pub kind: EventKind,
188    pub outcome: Outcome,
189    pub action: Cow<'static, str>,
190    pub resource: Option<String>,
191    pub reason: Option<String>,
192    pub matched_policy_id: Option<String>,
193    pub fields: HashMap<String, Sensitivity>,
194}
195
196// ---------------------------------------------------------------------------
197// Sensitivity
198// ---------------------------------------------------------------------------
199
200/// How a payload value is rendered when it lands in `fields_json`.
201/// Producer slices choose per-field; the skeleton does not decide what
202/// counts as sensitive.
203#[derive(Debug, Clone, PartialEq, Eq)]
204pub enum Sensitivity {
205    /// Value persisted as-is. Producer guarantees it isn't sensitive.
206    Raw(String),
207    /// Fingerprint instead of the value. `algo` is the hash name
208    /// (always `"blake3"` for the skeleton helper) and `hex` is the
209    /// lowercase hex digest.
210    Hashed { algo: &'static str, hex: String },
211    /// Placeholder: "a value existed at this field but we are not
212    /// logging it." Distinguishable from absence (no key at all).
213    Redacted,
214}
215
216impl Sensitivity {
217    pub fn raw<S: Into<String>>(s: S) -> Self {
218        Self::Raw(s.into())
219    }
220
221    pub fn hashed(value: &[u8]) -> Self {
222        let hex = blake3::hash(value).to_hex().to_string();
223        Self::Hashed {
224            algo: "blake3",
225            hex,
226        }
227    }
228
229    pub fn redacted() -> Self {
230        Self::Redacted
231    }
232}
233
234// ---------------------------------------------------------------------------
235// Ledger trait
236// ---------------------------------------------------------------------------
237
238/// Opaque id of a persisted event. Producers may store this to chain a
239/// follow-up audit entry to the original.
240#[derive(Debug, Clone, PartialEq, Eq)]
241pub struct EventId(pub String);
242
243#[derive(Debug)]
244pub enum ControlEventError {
245    Persistence(String),
246}
247
248impl std::fmt::Display for ControlEventError {
249    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
250        match self {
251            Self::Persistence(msg) => write!(f, "control-event persistence failed: {msg}"),
252        }
253    }
254}
255
256impl std::error::Error for ControlEventError {}
257
258/// Persistence sink for control events.
259///
260/// `emit` is synchronous. Callers running under
261/// [`ControlEventConfig::compliance_mode`] MUST treat `Err` as
262/// "abort the originating mutation" (fail closed). Callers in the
263/// default mode may log-and-continue, but the trait never swallows
264/// the failure on their behalf — that policy is per-caller.
265pub trait ControlEventLedger: Send + Sync {
266    fn emit(
267        &self,
268        ctx: &ControlEventCtx<'_>,
269        event: ControlEvent,
270    ) -> Result<EventId, ControlEventError>;
271}
272
273// ---------------------------------------------------------------------------
274// ControlEventConfig
275// ---------------------------------------------------------------------------
276
277/// Runtime knob for the ledger. Lives on
278/// `RedDBOptions::control_events` and is read at boot from
279/// `REDDB_COMPLIANCE_MODE`.
280#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
281pub struct ControlEventConfig {
282    /// When true, the producer slices MUST abort their originating
283    /// mutation on `emit` failure (fail closed). Default: false.
284    pub compliance_mode: bool,
285}
286
287impl ControlEventConfig {
288    /// Convenience: do callers need durable evidence before letting
289    /// the originating mutation complete?
290    pub fn require_persistence(&self) -> bool {
291        self.compliance_mode
292    }
293}
294
295// ---------------------------------------------------------------------------
296// RuntimeLedger — the skeleton's single implementor
297// ---------------------------------------------------------------------------
298
299/// Writes one row per `emit()` to `red.control_events` via the
300/// unified entity API. The collection is created on construction if
301/// it doesn't already exist (idempotent across re-opens).
302pub struct RuntimeLedger {
303    store: Arc<UnifiedStore>,
304}
305
306impl RuntimeLedger {
307    pub fn new(store: Arc<UnifiedStore>) -> Self {
308        let _ = store.get_or_create_collection(CONTROL_EVENTS_COLLECTION);
309        Self { store }
310    }
311}
312
313impl ControlEventLedger for RuntimeLedger {
314    fn emit(
315        &self,
316        ctx: &ControlEventCtx<'_>,
317        event: ControlEvent,
318    ) -> Result<EventId, ControlEventError> {
319        let ts_ms = now_unix_millis();
320        // ns since epoch: ms * 1e6, clamped to i64 range. The actual
321        // call site doesn't have nanosecond precision today; producer
322        // slices that need it can carry it via `fields`.
323        let ts_ns = (ts_ms as i128)
324            .saturating_mul(1_000_000)
325            .min(i64::MAX as i128) as i64;
326        // UUIDv7 is time-sortable in its first 48 bits, satisfying the
327        // brief's "ULID, sortable by time" requirement without adding
328        // a ulid dep.
329        let id = Uuid::new_v7().to_string();
330
331        let mut named: HashMap<String, Value> = HashMap::with_capacity(14);
332        named.insert("id".into(), Value::text(id.clone()));
333        named.insert("ts".into(), Value::Integer(ts_ns));
334        named.insert("kind".into(), Value::text(event.kind.as_str()));
335        named.insert("outcome".into(), Value::text(event.outcome.as_str()));
336        named.insert("actor_kind".into(), Value::text(ctx.actor.kind_str()));
337        named.insert(
338            "actor_user_id".into(),
339            ctx.actor
340                .user_id()
341                .map(|u| Value::text(u.to_string()))
342                .unwrap_or(Value::Null),
343        );
344        named.insert(
345            "scope".into(),
346            ctx.scope
347                .as_ref()
348                .map(|s| Value::text(s.to_string()))
349                .unwrap_or(Value::Null),
350        );
351        named.insert("action".into(), Value::text(event.action.to_string()));
352        named.insert(
353            "resource".into(),
354            event.resource.map(Value::text).unwrap_or(Value::Null),
355        );
356        named.insert(
357            "reason".into(),
358            event.reason.map(Value::text).unwrap_or(Value::Null),
359        );
360        named.insert(
361            "matched_policy_id".into(),
362            event
363                .matched_policy_id
364                .map(Value::text)
365                .unwrap_or(Value::Null),
366        );
367        named.insert(
368            "request_id".into(),
369            ctx.request_id
370                .as_ref()
371                .map(|s| Value::text(s.to_string()))
372                .unwrap_or(Value::Null),
373        );
374        named.insert(
375            "trace_id".into(),
376            ctx.trace_id
377                .as_ref()
378                .map(|s| Value::text(s.to_string()))
379                .unwrap_or(Value::Null),
380        );
381        named.insert(
382            "fields_json".into(),
383            Value::text(serialise_fields(&event.fields)),
384        );
385
386        let entity = UnifiedEntity::new(
387            EntityId::new(0),
388            EntityKind::TableRow {
389                table: Arc::from(CONTROL_EVENTS_COLLECTION),
390                row_id: 0,
391            },
392            EntityData::Row(RowData {
393                columns: Vec::new(),
394                named: Some(named),
395                schema: None,
396            }),
397        );
398        self.store
399            .insert_auto(CONTROL_EVENTS_COLLECTION, entity)
400            .map_err(|e| ControlEventError::Persistence(e.to_string()))?;
401        Ok(EventId(id))
402    }
403}
404
405/// Deterministic JSON serialiser for the `fields` map. Keys are sorted
406/// so two events with the same logical payload hash-compare equal at
407/// the row level (helpful for tests and dedup). Hand-rolled to avoid
408/// pulling serde into the ledger hot path.
409fn serialise_fields(fields: &HashMap<String, Sensitivity>) -> String {
410    let mut keys: Vec<&String> = fields.keys().collect();
411    keys.sort();
412    let mut out = String::from("{");
413    for (i, k) in keys.iter().enumerate() {
414        if i > 0 {
415            out.push(',');
416        }
417        out.push('"');
418        out.push_str(&json_escape(k));
419        out.push_str("\":");
420        match &fields[*k] {
421            Sensitivity::Raw(s) => {
422                out.push_str(r#"{"kind":"raw","value":""#);
423                out.push_str(&json_escape(s));
424                out.push_str(r#""}"#);
425            }
426            Sensitivity::Hashed { algo, hex } => {
427                out.push_str(r#"{"kind":"hashed","algo":""#);
428                out.push_str(algo);
429                out.push_str(r#"","hex":""#);
430                out.push_str(hex);
431                out.push_str(r#""}"#);
432            }
433            Sensitivity::Redacted => {
434                out.push_str(r#"{"kind":"redacted"}"#);
435            }
436        }
437    }
438    out.push('}');
439    out
440}
441
442#[cfg(test)]
443mod tests {
444    use super::*;
445
446    fn anon_ctx<'a>() -> ControlEventCtx<'a> {
447        ControlEventCtx {
448            actor: ActorRef::Anonymous,
449            scope: None,
450            request_id: None,
451            trace_id: None,
452        }
453    }
454
455    fn sample_event() -> ControlEvent {
456        ControlEvent {
457            kind: EventKind::PolicyCreate,
458            outcome: Outcome::Allowed,
459            action: Cow::Borrowed("policy.write"),
460            resource: Some("policy:test".into()),
461            reason: None,
462            matched_policy_id: Some("p-abc".into()),
463            fields: HashMap::new(),
464        }
465    }
466
467    #[test]
468    fn collection_is_created_on_first_open_and_reopen_is_idempotent() {
469        let store = Arc::new(UnifiedStore::new());
470        let _l1 = RuntimeLedger::new(store.clone());
471        assert!(store.get_collection(CONTROL_EVENTS_COLLECTION).is_some());
472        // A second ledger over the same store must not explode —
473        // `create_collection` would error on duplicate, but
474        // `RuntimeLedger::new` goes through `get_or_create_collection`.
475        let _l2 = RuntimeLedger::new(store.clone());
476        assert!(store.get_collection(CONTROL_EVENTS_COLLECTION).is_some());
477    }
478
479    #[test]
480    fn runtime_ledger_emit_persists_row_with_every_schema_column() {
481        let store = Arc::new(UnifiedStore::new());
482        let ledger = RuntimeLedger::new(store.clone());
483        let id = ledger.emit(&anon_ctx(), sample_event()).expect("emit ok");
484        assert!(!id.0.is_empty(), "row id must be populated");
485
486        let manager = store
487            .get_collection(CONTROL_EVENTS_COLLECTION)
488            .expect("table exists");
489        let rows = manager.query_all(|_| true);
490        assert_eq!(rows.len(), 1);
491
492        match &rows[0].data {
493            EntityData::Row(row) => {
494                let named = row.named.as_ref().expect("named columns present");
495                for col in [
496                    "id",
497                    "ts",
498                    "kind",
499                    "outcome",
500                    "actor_kind",
501                    "actor_user_id",
502                    "scope",
503                    "action",
504                    "resource",
505                    "reason",
506                    "matched_policy_id",
507                    "request_id",
508                    "trace_id",
509                    "fields_json",
510                ] {
511                    assert!(named.contains_key(col), "missing schema column {col}");
512                }
513                assert_eq!(named["kind"], Value::text("policy.create"));
514                assert_eq!(named["outcome"], Value::text("allowed"));
515                assert_eq!(named["actor_kind"], Value::text("anonymous"));
516                assert_eq!(named["actor_user_id"], Value::Null);
517                assert_eq!(named["scope"], Value::Null);
518                assert_eq!(named["resource"], Value::text("policy:test"));
519                assert_eq!(named["matched_policy_id"], Value::text("p-abc"));
520            }
521            other => panic!("expected Row, got {other:?}"),
522        }
523    }
524
525    #[test]
526    fn emit_with_user_actor_records_kind_and_label() {
527        let store = Arc::new(UnifiedStore::new());
528        let ledger = RuntimeLedger::new(store.clone());
529        let user = UserId::scoped("acme", "alice");
530        let ctx = ControlEventCtx {
531            actor: ActorRef::User(&user),
532            scope: Some(Cow::Borrowed("acme")),
533            request_id: Some(Cow::Borrowed("req-42")),
534            trace_id: None,
535        };
536        ledger.emit(&ctx, sample_event()).unwrap();
537        let manager = store.get_collection(CONTROL_EVENTS_COLLECTION).unwrap();
538        let rows = manager.query_all(|_| true);
539        let named = match &rows[0].data {
540            EntityData::Row(r) => r.named.as_ref().unwrap(),
541            _ => panic!(),
542        };
543        assert_eq!(named["actor_kind"], Value::text("user"));
544        assert_eq!(named["actor_user_id"], Value::text("acme/alice"));
545        assert_eq!(named["scope"], Value::text("acme"));
546        assert_eq!(named["request_id"], Value::text("req-42"));
547        assert_eq!(named["trace_id"], Value::Null);
548    }
549
550    #[test]
551    fn sensitivity_hashed_is_stable_blake3() {
552        let a = Sensitivity::hashed(b"hunter2");
553        let b = Sensitivity::hashed(b"hunter2");
554        assert_eq!(a, b);
555        match a {
556            Sensitivity::Hashed { algo, hex } => {
557                assert_eq!(algo, "blake3");
558                assert_eq!(hex.len(), 64);
559                // Pin the digest so producer-slice tests (652b/c/d)
560                // can rely on it across runs.
561                assert_eq!(hex, blake3::hash(b"hunter2").to_hex().to_string(),);
562            }
563            other => panic!("expected Hashed, got {other:?}"),
564        }
565    }
566
567    #[test]
568    fn sensitivity_redacted_serialises_without_value() {
569        let mut fields = HashMap::new();
570        fields.insert("password".to_string(), Sensitivity::redacted());
571        let s = serialise_fields(&fields);
572        assert_eq!(s, r#"{"password":{"kind":"redacted"}}"#);
573    }
574
575    // ---- ComplianceMode caller-decision shape ---------------------------
576    //
577    // The trait never swallows persistence failures; the per-caller
578    // policy (`abort the originating mutation` vs `log and continue`)
579    // is encoded by the caller against [`ControlEventConfig::require_persistence`].
580    // The tests below pin that contract via a deliberately failing fake.
581
582    struct FailingLedger;
583    impl ControlEventLedger for FailingLedger {
584        fn emit(
585            &self,
586            _: &ControlEventCtx<'_>,
587            _: ControlEvent,
588        ) -> Result<EventId, ControlEventError> {
589            Err(ControlEventError::Persistence("simulated".into()))
590        }
591    }
592
593    fn caller_decides(
594        cfg: ControlEventConfig,
595        ledger: &dyn ControlEventLedger,
596    ) -> Result<(), &'static str> {
597        match ledger.emit(&anon_ctx(), sample_event()) {
598            Ok(_) => Ok(()),
599            Err(_) if cfg.require_persistence() => Err("aborted"),
600            Err(_) => Ok(()),
601        }
602    }
603
604    #[test]
605    fn compliance_mode_makes_callers_fail_closed_on_persistence_failure() {
606        let cfg = ControlEventConfig {
607            compliance_mode: true,
608        };
609        assert!(cfg.require_persistence());
610        assert_eq!(caller_decides(cfg, &FailingLedger), Err("aborted"));
611    }
612
613    #[test]
614    fn default_mode_lets_callers_continue_on_persistence_failure() {
615        let cfg = ControlEventConfig::default();
616        assert!(!cfg.require_persistence());
617        // The trait still surfaces the error...
618        assert!(FailingLedger.emit(&anon_ctx(), sample_event()).is_err());
619        // ...but the caller is free to swallow it.
620        assert_eq!(caller_decides(cfg, &FailingLedger), Ok(()));
621    }
622}
reddb_server/runtime/control_events.rs

reddb_server/runtime/
control_events.rs
