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

vcon_server/
audit.rs
