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

vcon_server/
audit.rs
