use serde_json::{Value, json};
use sqlx::Row;
use sqlx::sqlite::SqliteRow;
use tracing::debug;
use crate::audit::{Actor, ActorKind, AuditEvent, AuditRecord, ClientContext};
use crate::sqlite::db::Database;
use crate::sqlite::nonce::now_secs;
use crate::sqlite::order::rfc3339;
#[derive(Debug, Clone)]
pub struct AuditEntry {
pub id: i64,
pub created_at: i64,
pub event: String,
pub outcome: String,
pub profile: String,
pub actor_kind: String,
pub actor_id: Option<String>,
pub account_id: Option<String>,
pub order_id: Option<String>,
pub cert_serial: Option<String>,
pub identifiers: Vec<String>,
pub client_ip: Option<String>,
pub client_ptr: Option<String>,
pub user_agent: Option<String>,
pub request_id: Option<String>,
pub reason: Option<String>,
pub detail: Option<String>,
}
const COLUMNS: &str = "id, created_at, event, outcome, profile, actor_kind, actor_id, \
account_id, order_id, cert_serial, identifiers, client_ip, client_ptr, \
user_agent, request_id, reason, detail";
#[derive(Debug, Clone, Default)]
pub struct AuditQuery {
pub profile: Option<String>,
pub account_id: Option<String>,
pub order_id: Option<String>,
pub cert_serial: Option<String>,
pub event: Option<String>,
pub outcome: Option<String>,
pub since: Option<i64>,
pub limit: i64,
pub offset: i64,
}
impl AuditQuery {
fn push_predicates(&self, builder: &mut sqlx::QueryBuilder<sqlx::Sqlite>) {
let mut separator = " WHERE ";
for (column, value) in [
("profile = ", self.profile.as_ref()),
("account_id = ", self.account_id.as_ref()),
("order_id = ", self.order_id.as_ref()),
("cert_serial = ", self.cert_serial.as_ref()),
("event = ", self.event.as_ref()),
("outcome = ", self.outcome.as_ref()),
] {
if let Some(value) = value {
builder
.push(separator)
.push(column)
.push_bind(value.clone());
separator = " AND ";
}
}
if let Some(since) = self.since {
builder
.push(separator)
.push("created_at >= ")
.push_bind(since);
}
}
}
impl AuditEntry {
fn from_row(row: SqliteRow) -> Result<Self, sqlx::Error> {
let identifiers_json: String = row.try_get("identifiers")?;
let identifiers: Vec<String> = serde_json::from_str(&identifiers_json)
.map_err(|error| sqlx::Error::Decode(Box::new(error)))?;
Ok(Self {
id: row.try_get("id")?,
created_at: row.try_get("created_at")?,
event: row.try_get("event")?,
outcome: row.try_get("outcome")?,
profile: row.try_get("profile")?,
actor_kind: row.try_get("actor_kind")?,
actor_id: row.try_get("actor_id")?,
account_id: row.try_get("account_id")?,
order_id: row.try_get("order_id")?,
cert_serial: row.try_get("cert_serial")?,
identifiers,
client_ip: row.try_get("client_ip")?,
client_ptr: row.try_get("client_ptr")?,
user_agent: row.try_get("user_agent")?,
request_id: row.try_get("request_id")?,
reason: row.try_get("reason")?,
detail: row.try_get("detail")?,
})
}
#[must_use]
pub fn event(&self) -> Option<AuditEvent> {
AuditEvent::parse(&self.event)
}
pub async fn insert(record: AuditRecord, database: &Database) -> Result<i64, sqlx::Error> {
let AuditRecord {
event,
profile,
actor,
account_id,
order_id,
cert_serial,
identifiers,
client,
reason,
detail,
} = record;
let Actor { kind, id: actor_id } = actor;
let ClientContext {
ip: client_ip,
ptr: client_ptr,
user_agent,
request_id,
} = client;
let identifiers_json = Value::from(identifiers).to_string();
let created_at = now_secs();
let id = sqlx::query(
"INSERT INTO audit_log (created_at, event, outcome, profile, actor_kind, actor_id, \
account_id, order_id, cert_serial, identifiers, client_ip, client_ptr, user_agent, \
request_id, reason, detail) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id;",
)
.bind(created_at)
.bind(event.as_str())
.bind(event.outcome())
.bind(&profile)
.bind(kind.as_str())
.bind(&actor_id)
.bind(&account_id)
.bind(&order_id)
.bind(&cert_serial)
.bind(identifiers_json)
.bind(&client_ip)
.bind(&client_ptr)
.bind(&user_agent)
.bind(&request_id)
.bind(&reason)
.bind(&detail)
.fetch_one(&database.pool)
.await?
.try_get::<i64, _>("id")?;
debug!(
event = "db_audit_row_written",
outcome = "success",
audit_id = id,
audit_event = event.as_str(),
profile = %profile,
actor_kind = kind.as_str(),
);
Ok(id)
}
pub async fn find_by_id(id: i64, database: &Database) -> Result<Option<Self>, sqlx::Error> {
let mut query =
sqlx::QueryBuilder::new(format!("SELECT {COLUMNS} FROM audit_log WHERE id = "));
query.push_bind(id);
let row = query.build().fetch_optional(&database.pool).await?;
row.map(Self::from_row).transpose()
}
pub async fn search(
query: &AuditQuery,
database: &Database,
) -> Result<(Vec<Self>, i64), sqlx::Error> {
debug!(
event = "db_audit_search_started",
outcome = "progress",
profile = ?query.profile,
account_id = ?query.account_id,
audit_event = ?query.event,
outcome = ?query.outcome,
limit = query.limit,
offset = query.offset,
);
let mut page = sqlx::QueryBuilder::new(format!("SELECT {COLUMNS} FROM audit_log"));
query.push_predicates(&mut page);
page.push(" ORDER BY created_at DESC, id DESC LIMIT ");
page.push_bind(query.limit);
page.push(" OFFSET ");
page.push_bind(query.offset);
let rows = page.build().fetch_all(&database.pool).await?;
let entries: Vec<Self> = rows
.into_iter()
.map(Self::from_row)
.collect::<Result<_, _>>()?;
let mut count = sqlx::QueryBuilder::new("SELECT COUNT(*) FROM audit_log");
query.push_predicates(&mut count);
let total: i64 = count
.build()
.fetch_one(&database.pool)
.await?
.try_get::<i64, _>(0)?;
Ok((entries, total))
}
pub async fn count_older_than(cutoff: i64, database: &Database) -> Result<i64, sqlx::Error> {
sqlx::query("SELECT COUNT(*) FROM audit_log WHERE created_at < ?;")
.bind(cutoff)
.fetch_one(&database.pool)
.await?
.try_get::<i64, _>(0)
}
pub async fn cleanup(cutoff: i64, database: &Database) -> Result<u64, sqlx::Error> {
let deleted = sqlx::query("DELETE FROM audit_log WHERE created_at < ?;")
.bind(cutoff)
.execute(&database.pool)
.await?
.rows_affected();
debug!(
event = "db_audit_cleanup",
outcome = "success",
rows_removed = deleted,
cutoff = cutoff
);
Ok(deleted)
}
#[must_use]
pub fn to_json(&self) -> Value {
let mut value = json!({
"id": self.id,
"createdAt": rfc3339(self.created_at),
"event": self.event,
"outcome": self.outcome,
"profile": self.profile,
"actorKind": self.actor_kind,
"identifiers": self.identifiers,
});
let map = value
.as_object_mut()
.expect("the literal above is an object");
for (key, field) in [
("actorId", self.actor_id.as_ref()),
("accountId", self.account_id.as_ref()),
("orderId", self.order_id.as_ref()),
("certSerial", self.cert_serial.as_ref()),
("clientIp", self.client_ip.as_ref()),
("clientPtr", self.client_ptr.as_ref()),
("userAgent", self.user_agent.as_ref()),
("requestId", self.request_id.as_ref()),
("reason", self.reason.as_ref()),
("detail", self.detail.as_ref()),
] {
if let Some(field) = field {
map.insert(key.to_string(), Value::from(field.clone()));
}
}
value
}
}
#[must_use]
pub fn actor_kinds() -> [&'static str; 4] {
[
ActorKind::Acme.as_str(),
ActorKind::Admin.as_str(),
ActorKind::Cli.as_str(),
ActorKind::System.as_str(),
]
}
#[cfg(test)]
mod tests {
use super::*;
use crate::audit::{Actor, AuditEvent, ClientContext};
use std::sync::Arc;
async fn db() -> Arc<Database> {
Arc::new(Database::connect_in_memory().await.unwrap())
}
fn client() -> ClientContext {
ClientContext {
ip: Some("203.0.113.7".to_string()),
ptr: Some("host.example.com".to_string()),
user_agent: Some("certbot/2.9.0".to_string()),
request_id: Some("req-1".to_string()),
}
}
fn record(event: AuditEvent, profile: &str) -> AuditRecord {
AuditRecord::new(event, profile, Actor::acme("acct-1"))
}
fn with_subject(mut record: AuditRecord, order_id: &str, identifiers: &[&str]) -> AuditRecord {
record.order_id = Some(order_id.to_string());
record.identifiers = identifiers.iter().map(|v| (*v).to_string()).collect();
record
}
#[tokio::test]
async fn a_row_round_trips_and_derives_its_own_outcome() {
let db = db().await;
let id = AuditEntry::insert(
with_subject(
record(AuditEvent::CertificateIssued, "le"),
"order-1",
&["a.example.com"],
)
.with_serial("0a0b")
.with_client(client()),
&db,
)
.await
.unwrap();
let entry = AuditEntry::find_by_id(id, &db).await.unwrap().unwrap();
assert_eq!(entry.event, "certificate_issued");
assert_eq!(entry.outcome, "success");
assert_eq!(entry.event(), Some(AuditEvent::CertificateIssued));
assert_eq!(entry.profile, "le");
assert_eq!(entry.actor_kind, "acme");
assert_eq!(entry.actor_id.as_deref(), Some("acct-1"));
assert_eq!(entry.order_id.as_deref(), Some("order-1"));
assert_eq!(entry.cert_serial.as_deref(), Some("0a0b"));
assert_eq!(entry.identifiers, vec!["a.example.com"]);
assert_eq!(entry.client_ip.as_deref(), Some("203.0.113.7"));
assert_eq!(entry.client_ptr.as_deref(), Some("host.example.com"));
assert_eq!(entry.user_agent.as_deref(), Some("certbot/2.9.0"));
assert_eq!(entry.request_id.as_deref(), Some("req-1"));
let failed = AuditEntry::insert(
record(AuditEvent::CertificateRevokeFailed, "le").with_reason("unauthorized"),
&db,
)
.await
.unwrap();
let entry = AuditEntry::find_by_id(failed, &db).await.unwrap().unwrap();
assert_eq!(entry.outcome, "failure");
assert!(entry.identifiers.is_empty());
assert_eq!(entry.client_ip, None);
assert!(AuditEntry::find_by_id(9_999, &db).await.unwrap().is_none());
}
#[tokio::test]
async fn a_row_survives_the_account_and_order_it_names_being_deleted() {
let db = db().await;
let account_id = crate::testutil::account_id(&db).await;
let order = crate::sqlite::order::Order::new(
"default",
&account_id,
vec![crate::sqlite::order::Identifier::dns("a.example.com")],
0,
None,
None,
);
order.insert(&db.pool).await.unwrap();
let id = AuditEntry::insert(
AuditRecord::new(
AuditEvent::CertificateIssued,
"default",
Actor::acme(&account_id),
)
.with_order(&order),
&db,
)
.await
.unwrap();
crate::sqlite::account::Account::delete(&account_id, &db)
.await
.unwrap();
assert!(
crate::sqlite::order::Order::find_by_id(&order.id, &db)
.await
.unwrap()
.is_none()
);
let entry = AuditEntry::find_by_id(id, &db).await.unwrap().unwrap();
assert_eq!(entry.account_id.as_deref(), Some(account_id.as_str()));
assert_eq!(entry.order_id.as_deref(), Some(order.id.as_str()));
assert_eq!(entry.identifiers, vec!["a.example.com"]);
}
#[tokio::test]
async fn every_filter_narrows_the_page_and_the_total_together() {
let db = db().await;
for (event, profile, account, serial) in [
(AuditEvent::CertificateIssued, "le", "acct-1", "aa"),
(AuditEvent::CertificateIssued, "le", "acct-2", "bb"),
(AuditEvent::CertificateIssueFailed, "le", "acct-1", "cc"),
(AuditEvent::CertificateRevoked, "internal", "acct-1", "dd"),
] {
AuditEntry::insert(
AuditRecord::new(event, profile, Actor::acme(account))
.with_account(account)
.with_serial(serial),
&db,
)
.await
.unwrap();
}
let count = async |query: AuditQuery| {
let (rows, total) = AuditEntry::search(&query, &db).await.unwrap();
assert_eq!(rows.len() as i64, total, "page and total must agree");
total
};
let base = AuditQuery {
limit: 50,
..AuditQuery::default()
};
assert_eq!(count(base.clone()).await, 4);
assert_eq!(
count(AuditQuery {
profile: Some("le".to_string()),
..base.clone()
})
.await,
3
);
assert_eq!(
count(AuditQuery {
account_id: Some("acct-1".to_string()),
..base.clone()
})
.await,
3
);
assert_eq!(
count(AuditQuery {
cert_serial: Some("dd".to_string()),
..base.clone()
})
.await,
1
);
assert_eq!(
count(AuditQuery {
event: Some("certificate_issued".to_string()),
..base.clone()
})
.await,
2
);
assert_eq!(
count(AuditQuery {
outcome: Some("failure".to_string()),
..base.clone()
})
.await,
1
);
assert_eq!(
count(AuditQuery {
profile: Some("le".to_string()),
outcome: Some("success".to_string()),
account_id: Some("acct-1".to_string()),
..base.clone()
})
.await,
1
);
assert_eq!(
count(AuditQuery {
event: Some("certificate_renewed".to_string()),
..base
})
.await,
0
);
}
#[tokio::test]
async fn paging_one_row_at_a_time_sees_every_row_exactly_once_newest_first() {
let db = db().await;
let mut inserted = Vec::new();
for index in 0..5 {
inserted.push(
AuditEntry::insert(
record(AuditEvent::CertificateIssued, "le")
.with_serial(format!("serial-{index}")),
&db,
)
.await
.unwrap(),
);
}
let mut seen = Vec::new();
for offset in 0..5 {
let (rows, total) = AuditEntry::search(
&AuditQuery {
limit: 1,
offset,
..AuditQuery::default()
},
&db,
)
.await
.unwrap();
assert_eq!(total, 5);
seen.push(rows[0].id);
}
inserted.reverse();
assert_eq!(seen, inserted);
}
#[tokio::test]
async fn cleanup_removes_only_what_is_strictly_older_than_the_cutoff() {
let db = db().await;
for _ in 0..3 {
AuditEntry::insert(record(AuditEvent::CertificateIssued, "le"), &db)
.await
.unwrap();
}
let now = crate::sqlite::nonce::now_secs();
let past = now - 3600;
assert_eq!(AuditEntry::count_older_than(past, &db).await.unwrap(), 0);
assert_eq!(AuditEntry::cleanup(past, &db).await.unwrap(), 0);
assert_eq!(AuditEntry::count_older_than(now, &db).await.unwrap(), 0);
let future = now + 3600;
assert_eq!(AuditEntry::count_older_than(future, &db).await.unwrap(), 3);
assert_eq!(AuditEntry::cleanup(future, &db).await.unwrap(), 3);
assert_eq!(AuditEntry::count_older_than(future, &db).await.unwrap(), 0);
}
#[tokio::test]
async fn to_json_omits_the_columns_that_have_no_value() {
let db = db().await;
let bare = AuditEntry::insert(
AuditRecord::new(AuditEvent::CertificateRevoked, "le", Actor::system()),
&db,
)
.await
.unwrap();
let json = AuditEntry::find_by_id(bare, &db)
.await
.unwrap()
.unwrap()
.to_json();
let object = json.as_object().unwrap();
assert_eq!(object["event"], "certificate_revoked");
assert_eq!(object["outcome"], "success");
assert_eq!(object["actorKind"], "system");
assert_eq!(object["identifiers"], serde_json::json!([]));
assert!(object["createdAt"].as_str().unwrap().contains('T'));
for absent in [
"actorId",
"accountId",
"orderId",
"certSerial",
"clientIp",
"clientPtr",
"userAgent",
"requestId",
"reason",
"detail",
] {
assert!(!object.contains_key(absent), "{absent} should be absent");
}
let full = AuditEntry::insert(
record(AuditEvent::CertificateIssueFailed, "le")
.with_client(client())
.with_reason("badCSR")
.with_detail("nope"),
&db,
)
.await
.unwrap();
let json = AuditEntry::find_by_id(full, &db)
.await
.unwrap()
.unwrap()
.to_json();
assert_eq!(json["clientIp"], "203.0.113.7");
assert_eq!(json["clientPtr"], "host.example.com");
assert_eq!(json["reason"], "badCSR");
assert_eq!(json["detail"], "nope");
}
#[tokio::test]
async fn an_unknown_event_string_still_loads_and_simply_does_not_parse() {
let db = db().await;
sqlx::query(
"INSERT INTO audit_log (created_at, event, outcome, profile, actor_kind) \
VALUES (0, 'certificate_issued', 'success', 'le', 'acme');",
)
.execute(&db.pool)
.await
.unwrap();
let mut entry = AuditEntry::search(
&AuditQuery {
limit: 1,
..AuditQuery::default()
},
&db,
)
.await
.unwrap()
.0
.remove(0);
entry.event = "certificate_renewed".to_string();
assert_eq!(entry.event(), None);
}
#[tokio::test]
async fn the_schema_refuses_an_event_outcome_or_actor_it_does_not_know() {
let db = db().await;
for (event, outcome, actor) in [
("certificate_renewed", "success", "acme"),
("certificate_issued", "maybe", "acme"),
("certificate_issued", "success", "robot"),
] {
let error = sqlx::query(
"INSERT INTO audit_log (created_at, event, outcome, profile, actor_kind) \
VALUES (0, ?, ?, 'le', ?);",
)
.bind(event)
.bind(outcome)
.bind(actor)
.execute(&db.pool)
.await
.unwrap_err();
assert!(
error.to_string().contains("CHECK constraint failed"),
"{event}/{outcome}/{actor} was accepted: {error}"
);
}
}
#[test]
fn the_actor_kinds_helper_lists_every_variant() {
assert_eq!(actor_kinds(), ["acme", "admin", "cli", "system"]);
}
}