vcon_server/
audit.rs

1//! The audit log: one record per request, written in the same transaction
2//! as the change it describes. A record holds no vCon content, no token
3//! and no client address.
4
5use std::io::{self, Write};
6
7use rusqlite::{Connection, params};
8use serde_json::json;
9use time::OffsetDateTime;
10use time::format_description::well_known::Rfc3339;
11
12use crate::ids::{KeyId, VersionId};
13use crate::query::{self, BadQuery};
14
15/// The audit table, part of every catalog that holds a log.
16pub const SCHEMA: &str = "
17CREATE TABLE audit (
18    seq INTEGER PRIMARY KEY AUTOINCREMENT,
19    at TEXT NOT NULL,
20    actor TEXT NOT NULL,
21    action TEXT NOT NULL,
22    target TEXT,
23    version TEXT,
24    status INTEGER NOT NULL,
25    count INTEGER NOT NULL
26);
27CREATE INDEX audit_counted ON audit (actor, action, status, at);
28";
29
30/// Who made a request.
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub enum Actor {
33    /// A caller's key.
34    Key(KeyId),
35    /// The purge sweep.
36    Sweep,
37    /// A command run on the host.
38    Cli,
39    /// A caller with no known key: its token did not parse, or names a key
40    /// id the key table does not hold.
41    Unparsed,
42}
43
44/// What a request asked for. A refused request records what it asked for.
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum Action {
47    Create,
48    Replace,
49    Read,
50    Delete,
51    Purge,
52    LegalRead,
53    LegalVersions,
54    Restore,
55    Hold,
56    Release,
57    KeyCreate,
58    KeyRevoke,
59    KeyList,
60    KeyInfo,
61    Retention,
62    Health,
63    AuthFailed,
64    TenantCreate,
65    TenantList,
66    TenantRetention,
67    TenantRemove,
68    RekeyMaster,
69    RekeyTenant,
70    List,
71    AuditRead,
72    Openapi,
73    Backup,
74    /// An export of the tenant's versions, as Parquet files.
75    Export,
76    /// A tenant's metadata level set, in the admin log.
77    TenantMetadata,
78    /// The tenant's metadata level set, in its own log.
79    Metadata,
80    /// A request to the MCP server, counted (`count`), not recorded.
81    Mcp,
82}
83
84impl Action {
85    /// The name written in the log.
86    pub fn as_str(self) -> &'static str {
87        match self {
88            Action::Create => "create",
89            Action::Replace => "replace",
90            Action::Read => "read",
91            Action::Delete => "delete",
92            Action::Purge => "purge",
93            Action::LegalRead => "legal_read",
94            Action::LegalVersions => "legal_versions",
95            Action::Restore => "restore",
96            Action::Hold => "hold",
97            Action::Release => "release",
98            Action::KeyCreate => "key_create",
99            Action::KeyRevoke => "key_revoke",
100            Action::KeyList => "key_list",
101            Action::KeyInfo => "key_info",
102            Action::Retention => "retention",
103            Action::Health => "health",
104            Action::AuthFailed => "auth_failed",
105            Action::TenantCreate => "tenant_create",
106            Action::TenantList => "tenant_list",
107            Action::TenantRetention => "tenant_retention",
108            Action::TenantRemove => "tenant_remove",
109            Action::RekeyMaster => "rekey_master",
110            Action::RekeyTenant => "rekey_tenant",
111            Action::List => "list",
112            Action::AuditRead => "audit_read",
113            Action::Openapi => "openapi",
114            Action::Backup => "backup",
115            Action::Export => "export",
116            Action::TenantMetadata => "tenant_metadata",
117            Action::Metadata => "metadata",
118            Action::Mcp => "mcp",
119        }
120    }
121}
122
123/// One request, as recorded.
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub struct Entry {
126    pub actor: Actor,
127    pub action: Action,
128    /// The vCon's uuid, or the tenant or key id for an admin action.
129    pub target: Option<String>,
130    /// The version written, read or purged, if there is one.
131    pub version: Option<VersionId>,
132    /// The HTTP status answered, or the command's result.
133    pub status: u16,
134}
135
136/// Why a record could not be written or exported.
137#[derive(Debug)]
138pub enum AuditError {
139    Sqlite(rusqlite::Error),
140    Write(io::Error),
141}
142
143impl From<rusqlite::Error> for AuditError {
144    fn from(e: rusqlite::Error) -> AuditError {
145        AuditError::Sqlite(e)
146    }
147}
148
149impl Actor {
150    /// The name written in the log: the key id, `sweep`, `cli`, or empty.
151    fn name(self) -> String {
152        match self {
153            Actor::Key(id) => id.to_string(),
154            Actor::Sweep => "sweep".to_owned(),
155            Actor::Cli => "cli".to_owned(),
156            Actor::Unparsed => String::new(),
157        }
158    }
159}
160
161/// `now` in RFC 3339, in UTC.
162fn rfc3339(now: OffsetDateTime) -> String {
163    now.to_offset(time::UtcOffset::UTC)
164        .format(&Rfc3339)
165        .expect("every UTC time from the clock formats")
166}
167
168/// Appends one record, at `now`, in `tx`.
169pub fn record(tx: &Connection, entry: &Entry, now: OffsetDateTime) -> Result<(), AuditError> {
170    tx.execute(
171        "INSERT INTO audit (at, actor, action, target, version, status, count)
172         VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1)",
173        params![
174            rfc3339(now),
175            entry.actor.name(),
176            entry.action.as_str(),
177            entry.target,
178            entry.version.map(|v| v.to_string()),
179            entry.status,
180        ],
181    )?;
182    Ok(())
183}
184
185/// Adds one to the record for this actor, action and status in the minute
186/// of `now`, making it if needed. For requests that could otherwise grow a
187/// log one row each: failed authentications and health checks.
188pub fn count(tx: &Connection, entry: &Entry, now: OffsetDateTime) -> Result<(), AuditError> {
189    let minute = now
190        .to_offset(time::UtcOffset::UTC)
191        .replace_second(0)
192        .and_then(|t| t.replace_nanosecond(0))
193        .expect("0 is a valid second and nanosecond");
194    let fields = params![
195        rfc3339(minute),
196        entry.actor.name(),
197        entry.action.as_str(),
198        entry.target,
199        entry.version.map(|v| v.to_string()),
200        entry.status,
201    ];
202    let added = tx.execute(
203        "UPDATE audit SET count = count + 1
204         WHERE at = ?1 AND actor = ?2 AND action = ?3 AND target IS ?4 AND version IS ?5
205           AND status = ?6",
206        fields,
207    )?;
208    if added == 0 {
209        tx.execute(
210            "INSERT INTO audit (at, actor, action, target, version, status, count)
211             VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1)",
212            fields,
213        )?;
214    }
215    Ok(())
216}
217
218/// Writes every record after `since` as JSON lines, oldest first.
219pub fn export(log: &Connection, since: Option<i64>, out: &mut dyn Write) -> Result<(), AuditError> {
220    // SQLite reads a negative limit as none.
221    lines(log, since.unwrap_or(0), -1, out)
222}
223
224/// Writes at most `limit` records after `after` as JSON lines, oldest
225/// first: the one rendering both `export` and `page` write.
226fn lines(log: &Connection, after: i64, limit: i64, out: &mut dyn Write) -> Result<(), AuditError> {
227    let mut rows = log.prepare(
228        "SELECT seq, at, actor, action, target, version, status, count
229         FROM audit WHERE seq > ?1 ORDER BY seq LIMIT ?2",
230    )?;
231    let mut rows = rows.query(params![after, limit])?;
232    while let Some(row) = rows.next()? {
233        let record = json!({
234            "seq": row.get::<_, i64>(0)?,
235            "at": row.get::<_, String>(1)?,
236            "actor": row.get::<_, String>(2)?,
237            "action": row.get::<_, String>(3)?,
238            "target": row.get::<_, Option<String>>(4)?,
239            "version": row.get::<_, Option<String>>(5)?,
240            "status": row.get::<_, u16>(6)?,
241            "count": row.get::<_, i64>(7)?,
242        });
243        writeln!(out, "{record}").map_err(AuditError::Write)?;
244    }
245    Ok(())
246}
247
248/// What an audit read asks for: at most `limit` records after the seq
249/// `after`, oldest first.
250#[derive(Debug, Clone, Copy, PartialEq, Eq)]
251pub struct AuditQuery {
252    pub after: i64,
253    pub limit: u32,
254}
255
256impl AuditQuery {
257    pub const DEFAULT_LIMIT: u32 = 1000;
258    pub const MAX_LIMIT: u32 = 10000;
259
260    /// An audit query from its parameters' values, the one rule for them
261    /// wherever they come from: `after` a seq (0, the default, is the
262    /// start of the log), `limit` 1 to `MAX_LIMIT`.
263    pub fn new(after: Option<u64>, limit: Option<u64>) -> Result<AuditQuery, BadQuery> {
264        let after = after
265            .map(|n| i64::try_from(n).map_err(|_| BadQuery))
266            .transpose()?
267            .unwrap_or(0);
268        let limit = match limit {
269            None => AuditQuery::DEFAULT_LIMIT,
270            Some(n) => u32::try_from(n)
271                .ok()
272                .filter(|n| (1..=AuditQuery::MAX_LIMIT).contains(n))
273                .ok_or(BadQuery)?,
274        };
275        Ok(AuditQuery { after, limit })
276    }
277
278    /// An audit query from an HTTP query string, which may name `after` and
279    /// `limit`, each at most once.
280    pub fn from_query(text: Option<&str>) -> Result<AuditQuery, BadQuery> {
281        let read = query::pairs(text, &["after", "limit"])?;
282        let number = |name| {
283            read.get(name)
284                .map(|n: &String| n.parse::<u64>().map_err(|_| BadQuery))
285                .transpose()
286        };
287        AuditQuery::new(number("after")?, number("limit")?)
288    }
289}
290
291/// Writes at most `query.limit` records after `query.after` as JSON lines,
292/// oldest first: the first lines `export` writes from the same place.
293pub fn page(log: &Connection, query: &AuditQuery, out: &mut dyn Write) -> Result<(), AuditError> {
294    lines(log, query.after, i64::from(query.limit), out)
295}
296
297/// Reading a log back, for tests.
298#[cfg(test)]
299pub(crate) mod testing {
300    use rusqlite::Connection;
301    use serde_json::Value;
302
303    /// The records `export` writes after `since`, parsed.
304    pub(crate) fn exported(log: &Connection, since: Option<i64>) -> Vec<Value> {
305        let mut out = Vec::new();
306        super::export(log, since, &mut out).expect("export");
307        String::from_utf8(out)
308            .expect("UTF-8")
309            .lines()
310            .map(|line| serde_json::from_str(line).expect("one JSON object per line"))
311            .collect()
312    }
313}
314
315#[cfg(test)]
316mod tests {
317    use serde_json::{Value, json};
318    use time::macros::datetime;
319
320    use super::testing::exported;
321    use super::*;
322    use crate::catalog;
323    use crate::envelope::Key;
324
325    struct Log {
326        _dir: tempfile::TempDir,
327        conn: Connection,
328    }
329
330    fn log() -> Log {
331        let dir = tempfile::tempdir().expect("temp dir");
332        let key = Key::generate().expect("key");
333        let conn = catalog::open(&dir.path().join("catalog.sqlite"), &key, SCHEMA).expect("opens");
334        Log { _dir: dir, conn }
335    }
336
337    fn read_by(key: KeyId, status: u16) -> Entry {
338        Entry {
339            actor: Actor::Key(key),
340            action: Action::Read,
341            target: Some("01a07da8-c2bb-83e5-b9a2-279e0d16bc46".into()),
342            version: Some(VersionId([0x22; 16])),
343            status,
344        }
345    }
346
347    #[test]
348    fn records_are_written_and_exported_back() {
349        let mut log = log();
350        let key = Key::generate().expect("key").id();
351        let tx = log.conn.transaction().expect("tx");
352        record(
353            &tx,
354            &read_by(key, 200),
355            datetime!(2026-10-23 12:34:56.789 UTC),
356        )
357        .expect("record");
358        let purge = Entry {
359            actor: Actor::Sweep,
360            action: Action::Purge,
361            target: Some("01a07da8-c2bb-83e5-b9a2-279e0d16bc46".into()),
362            version: None,
363            status: 200,
364        };
365        record(&tx, &purge, datetime!(2026-10-23 12:35:00 UTC)).expect("record");
366        tx.commit().expect("commit");
367        assert_eq!(
368            exported(&log.conn, None),
369            [
370                json!({"seq": 1, "at": "2026-10-23T12:34:56.789Z", "actor": key.to_string(),
371                       "action": "read", "target": "01a07da8-c2bb-83e5-b9a2-279e0d16bc46",
372                       "version": "22".repeat(16), "status": 200, "count": 1}),
373                json!({"seq": 2, "at": "2026-10-23T12:35:00Z", "actor": "sweep",
374                       "action": "purge", "target": "01a07da8-c2bb-83e5-b9a2-279e0d16bc46",
375                       "version": null, "status": 200, "count": 1}),
376            ]
377        );
378        assert_eq!(
379            exported(&log.conn, Some(1)).len(),
380            1,
381            "only records after seq 1"
382        );
383        assert_eq!(exported(&log.conn, Some(2)).len(), 0);
384    }
385
386    #[test]
387    fn times_are_written_in_utc() {
388        let mut log = log();
389        let key = Key::generate().expect("key").id();
390        let tx = log.conn.transaction().expect("tx");
391        record(
392            &tx,
393            &read_by(key, 200),
394            datetime!(2026-10-23 14:34:56 +02:00),
395        )
396        .expect("record");
397        tx.commit().expect("commit");
398        assert_eq!(exported(&log.conn, None)[0]["at"], "2026-10-23T12:34:56Z");
399    }
400
401    #[test]
402    fn counted_records_hold_one_row_per_actor_action_status_and_minute() {
403        let mut log = log();
404        let health = |status| Entry {
405            actor: Actor::Unparsed,
406            action: Action::Health,
407            target: None,
408            version: None,
409            status,
410        };
411        let tx = log.conn.transaction().expect("tx");
412        for second in [0, 30, 59] {
413            let now = datetime!(2026-10-23 12:34:00 UTC) + time::Duration::seconds(second);
414            count(&tx, &health(200), now).expect("count");
415        }
416        count(&tx, &health(200), datetime!(2026-10-23 12:35:00 UTC)).expect("next minute");
417        count(&tx, &health(503), datetime!(2026-10-23 12:34:10 UTC)).expect("other status");
418        // Same actor, status and minute as the 503 health record above,
419        // so only the action keeps the two apart.
420        let failed = Entry {
421            action: Action::AuthFailed,
422            ..health(503)
423        };
424        count(&tx, &failed, datetime!(2026-10-23 12:34:20 UTC)).expect("other action");
425        let cli = Entry {
426            actor: Actor::Cli,
427            ..health(200)
428        };
429        count(&tx, &cli, datetime!(2026-10-23 12:34:40 UTC)).expect("other actor");
430        tx.commit().expect("commit");
431        let rows: Vec<(String, String, String, u64, u64)> = exported(&log.conn, None)
432            .iter()
433            .map(|r| {
434                (
435                    r["at"].as_str().expect("at").to_owned(),
436                    r["actor"].as_str().expect("actor").to_owned(),
437                    r["action"].as_str().expect("action").to_owned(),
438                    r["status"].as_u64().expect("status"),
439                    r["count"].as_u64().expect("count"),
440                )
441            })
442            .collect();
443        let row = |at: &str, actor: &str, action: &str, status, count| {
444            (
445                at.to_owned(),
446                actor.to_owned(),
447                action.to_owned(),
448                status,
449                count,
450            )
451        };
452        assert_eq!(
453            rows,
454            [
455                row("2026-10-23T12:34:00Z", "", "health", 200, 3),
456                row("2026-10-23T12:35:00Z", "", "health", 200, 1),
457                row("2026-10-23T12:34:00Z", "", "health", 503, 1),
458                row("2026-10-23T12:34:00Z", "", "auth_failed", 503, 1),
459                row("2026-10-23T12:34:00Z", "cli", "health", 200, 1),
460            ]
461        );
462    }
463
464    #[test]
465    fn a_record_in_a_transaction_that_does_not_commit_is_not_kept() {
466        let mut log = log();
467        let tx = log.conn.transaction().expect("tx");
468        let key = Key::generate().expect("key").id();
469        record(&tx, &read_by(key, 200), datetime!(2026-10-23 12:00:00 UTC)).expect("record");
470        drop(tx);
471        assert_eq!(exported(&log.conn, None), Vec::<Value>::new());
472    }
473
474    #[test]
475    fn every_action_has_its_name_in_the_log() {
476        // The name each action has in the logs, which exports and the audit
477        // routes give to operators. The match has no catch-all arm, so a
478        // new action fails to compile here until its name is written down.
479        let named = |action: Action| match action {
480            Action::Create => "create",
481            Action::Replace => "replace",
482            Action::Read => "read",
483            Action::Delete => "delete",
484            Action::Purge => "purge",
485            Action::LegalRead => "legal_read",
486            Action::LegalVersions => "legal_versions",
487            Action::Restore => "restore",
488            Action::Hold => "hold",
489            Action::Release => "release",
490            Action::KeyCreate => "key_create",
491            Action::KeyRevoke => "key_revoke",
492            Action::KeyList => "key_list",
493            Action::KeyInfo => "key_info",
494            Action::Retention => "retention",
495            Action::Health => "health",
496            Action::AuthFailed => "auth_failed",
497            Action::TenantCreate => "tenant_create",
498            Action::TenantList => "tenant_list",
499            Action::TenantRetention => "tenant_retention",
500            Action::TenantRemove => "tenant_remove",
501            Action::RekeyMaster => "rekey_master",
502            Action::RekeyTenant => "rekey_tenant",
503            Action::List => "list",
504            Action::AuditRead => "audit_read",
505            Action::Openapi => "openapi",
506            Action::Backup => "backup",
507            Action::Export => "export",
508            Action::TenantMetadata => "tenant_metadata",
509            Action::Metadata => "metadata",
510            Action::Mcp => "mcp",
511        };
512        let every = [
513            Action::Create,
514            Action::Replace,
515            Action::Read,
516            Action::Delete,
517            Action::Purge,
518            Action::LegalRead,
519            Action::LegalVersions,
520            Action::Restore,
521            Action::Hold,
522            Action::Release,
523            Action::KeyCreate,
524            Action::KeyRevoke,
525            Action::KeyList,
526            Action::KeyInfo,
527            Action::Retention,
528            Action::Health,
529            Action::AuthFailed,
530            Action::TenantCreate,
531            Action::TenantList,
532            Action::TenantRetention,
533            Action::TenantRemove,
534            Action::RekeyMaster,
535            Action::RekeyTenant,
536            Action::List,
537            Action::AuditRead,
538            Action::Openapi,
539            Action::Backup,
540            Action::Export,
541            Action::TenantMetadata,
542            Action::Metadata,
543            Action::Mcp,
544        ];
545        let mut seen = std::collections::BTreeSet::new();
546        for action in every {
547            assert_eq!(action.as_str(), named(action), "{action:?}");
548            assert!(
549                seen.insert(action.as_str()),
550                "{action:?}: a name of its own"
551            );
552        }
553    }
554
555    /// Five read records, at whole seconds.
556    fn five(log: &mut Log) {
557        let key = Key::generate().expect("key").id();
558        let tx = log.conn.transaction().expect("tx");
559        for second in 0..5 {
560            let now = datetime!(2026-10-23 12:00:00 UTC) + time::Duration::seconds(second);
561            record(&tx, &read_by(key, 200), now).expect("record");
562        }
563        tx.commit().expect("commit");
564    }
565
566    fn exported_text(log: &Connection, since: Option<i64>) -> String {
567        let mut out = Vec::new();
568        export(log, since, &mut out).expect("export");
569        String::from_utf8(out).expect("UTF-8")
570    }
571
572    fn paged(log: &Connection, after: i64, limit: u32) -> String {
573        let mut out = Vec::new();
574        page(log, &AuditQuery { after, limit }, &mut out).expect("page");
575        String::from_utf8(out).expect("UTF-8")
576    }
577
578    #[test]
579    fn a_page_is_the_first_lines_export_writes_from_the_same_place() {
580        let mut log = log();
581        five(&mut log);
582        for after in 0..=6 {
583            let lines: Vec<String> = exported_text(&log.conn, Some(after))
584                .lines()
585                .map(|l| format!("{l}\n"))
586                .collect();
587            for limit in 1..=6 {
588                let take = lines.len().min(usize::try_from(limit).expect("small"));
589                assert_eq!(
590                    paged(&log.conn, after, limit),
591                    lines[..take].concat(),
592                    "after {after}, limit {limit}"
593                );
594            }
595        }
596        assert_eq!(paged(&log.conn, 0, 10), exported_text(&log.conn, None));
597    }
598
599    #[test]
600    fn an_audit_query_takes_a_seq_and_a_limit_of_1_to_10000() {
601        let q = |after, limit| AuditQuery { after, limit };
602        assert_eq!(AuditQuery::new(None, None), Ok(q(0, 1000)));
603        assert_eq!(AuditQuery::new(Some(7), Some(1)), Ok(q(7, 1)));
604        assert_eq!(AuditQuery::new(None, Some(10000)), Ok(q(0, 10000)));
605        assert_eq!(
606            AuditQuery::new(Some(i64::MAX as u64), None),
607            Ok(q(i64::MAX, 1000))
608        );
609        for (after, limit) in [
610            (None, Some(0)),
611            (None, Some(10001)),
612            (None, Some(u64::MAX)),
613            (Some(i64::MAX as u64 + 1), None),
614        ] {
615            assert_eq!(
616                AuditQuery::new(after, limit),
617                Err(BadQuery),
618                "{after:?} {limit:?}"
619            );
620        }
621        assert_eq!(AuditQuery::from_query(None), Ok(q(0, 1000)));
622        assert_eq!(
623            AuditQuery::from_query(Some("after=12&limit=3")),
624            Ok(q(12, 3))
625        );
626        for bad in [
627            "after=-1",
628            "after=x",
629            "after=",
630            "limit=0",
631            "limit=1&limit=1",
632            "since=3",
633            "after=1.0",
634        ] {
635            assert_eq!(AuditQuery::from_query(Some(bad)), Err(BadQuery), "{bad}");
636        }
637    }
638}
vcon_server/audit.rs

vcon_server/
audit.rs
