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

reddb_server/runtime/
control_events.rs
