server/lib/src/audit/query.rs
//! **Search the audit log** — filtered, paged records (actor, artifact, action,
//! time range, source address, status), and its oldest special case, "what did C
//! download" (stinky SB #19).
//!
//! Lane HOL-AUDIT (2026-10-30) brought the rich query back after the lean cut
//! (`9521bf06`) and put it on the Arrow index ([`super::arrow_ipc`]), which is
//! caught up with the one hash-chained log before every answer: an answer cannot
//! disagree with the log, and every row names its `seq` and chain `hash` so a
//! reader can check it against a verified chain (`AuditService.VerifyAuditChain`).
//!
//! Read by an Auditor or an Admin only ([`super::trail::authorize_read`]), over
//! gRPC `AuditService.QueryAudit` / `QueryDownloads`, HTTP `GET /-/audit/downloads`
//! and `holger audit query`.
use serde::Serialize;
use super::arrow_ipc::Row;
use super::{AuditAction, AuditError, QuerySource, StoredRecord};
/// The rows an answer carries when the caller names no limit.
pub const DEFAULT_LIMIT: usize = 1_000;
/// The most rows one answer carries, whatever the caller asks.
pub const MAX_LIMIT: usize = 10_000;
/// **Which records**: every field optional, the set ones AND together.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AuditFilter {
/// The actor — the authenticated principal (`anonymous` for open reads).
pub ident: Option<String>,
/// The repository.
pub repo: Option<String>,
/// A PREFIX of the artifact (`acme/` a namespace, `acme/widget@` every version).
pub artifact_prefix: Option<String>,
/// The action, as the log writes it (`download`, `upload`, …).
pub action: Option<String>,
/// The client address, with or without its port; a port-less filter matches
/// every port of that address.
pub source_ip: Option<String>,
/// The exact status.
pub status: Option<u16>,
/// Only statuses of this class (`2` = 2xx).
pub status_class: Option<u16>,
/// From this moment, inclusive (nanoseconds since the Unix epoch).
pub since_nanos: Option<i64>,
/// Up to this moment, exclusive (consecutive windows tile).
pub until_nanos: Option<i64>,
}
fn named(s: &str) -> Option<String> {
Some(s.trim().to_string()).filter(|s| !s.is_empty())
}
impl AuditFilter {
/// From wire values where empty / zero means "not named". An action the log
/// never writes is refused, never matched against nothing.
#[allow(clippy::too_many_arguments)]
pub fn of_wire(
ident: &str,
repo: &str,
artifact_prefix: &str,
action: &str,
source_ip: &str,
status: u32,
since_nanos: i64,
until_nanos: i64,
) -> Result<Self, AuditError> {
let action = named(action);
if let Some(a) = &action {
if AuditAction::from_wire(a).as_str() != a {
return Err(AuditError::Other(format!(
"`{a}` is not an action the audit log writes (download, upload, delete, list, promote, entitlement, purge, scan, token, other)"
)));
}
}
let status = match status {
0 => None,
n => Some(u16::try_from(n).map_err(|_| {
AuditError::Other(format!("status {n} is not an HTTP-like status"))
})?),
};
Ok(Self {
ident: named(ident),
repo: named(repo),
artifact_prefix: named(artifact_prefix),
action,
source_ip: named(source_ip),
status,
status_class: None,
since_nanos: (since_nanos > 0).then_some(since_nanos),
until_nanos: (until_nanos > 0).then_some(until_nanos),
})
}
/// An empty time range is refused rather than answered with nothing.
pub fn check(&self) -> Result<(), AuditError> {
if let (Some(a), Some(b)) = (self.since_nanos, self.until_nanos) {
if b <= a {
return Err(AuditError::Other(format!(
"the time range is empty: until ({b}) is not after since ({a})"
)));
}
}
Ok(())
}
/// Does `r` match?
pub fn matches(&self, r: &Row<'_>) -> bool {
self.ident.as_deref().is_none_or(|i| r.ident == i)
&& self.repo.as_deref().is_none_or(|p| r.repo == p)
&& self
.artifact_prefix
.as_deref()
.is_none_or(|p| r.artifact.starts_with(p))
&& self.action.as_deref().is_none_or(|a| r.action == a)
&& self
.source_ip
.as_deref()
.is_none_or(|ip| ip_matches(ip, r.source_ip))
&& self.status.is_none_or(|s| r.status == s)
&& self.status_class.is_none_or(|c| r.status / 100 == c)
&& self.since_nanos.is_none_or(|t| r.ts_nanos >= t)
&& self.until_nanos.is_none_or(|t| r.ts_nanos < t)
}
}
/// A filter address with a port matches that `ip:port` only; without one, every
/// port of the address.
fn ip_matches(want: &str, recorded: &str) -> bool {
let w = want.trim();
if w == recorded.trim() {
return true;
}
super::client_host(w) == w
&& super::client_host(recorded) == w.trim_start_matches('[').trim_end_matches(']')
}
/// **One page of records.**
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AuditQuery {
pub filter: AuditFilter,
/// Matching records to skip (pass back [`AuditPage::next_offset`]).
pub offset: u64,
/// Page size; 0 = [`PAGE_DEFAULT`], capped at [`PAGE_MAX`].
pub limit: usize,
}
/// The page size when the caller names none.
pub const PAGE_DEFAULT: usize = 100;
/// The largest page.
pub const PAGE_MAX: usize = 1_000;
/// The answer: the page, oldest first, how many matched in all, and where the next
/// page starts.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AuditPage {
pub records: Vec<StoredRecord>,
/// Every match, before paging: the number to show.
pub matched: u64,
pub next_offset: u64,
/// Matches exist beyond this page.
pub more: bool,
/// Rows read to answer.
pub scanned: u64,
/// The page size used.
pub limit_applied: usize,
}
/// **Filter and page the records** of `source`, oldest first.
pub fn query(source: &QuerySource, q: &AuditQuery) -> Result<AuditPage, AuditError> {
q.filter.check()?;
let table = source.table()?;
let limit = match q.limit {
0 => PAGE_DEFAULT,
n => n.min(PAGE_MAX),
};
let mut page = AuditPage {
limit_applied: limit,
..Default::default()
};
table.for_each_row(|r| {
page.scanned += 1;
if !q.filter.matches(&r) {
return;
}
if page.matched >= q.offset && page.records.len() < limit {
page.records.push(r.to_record());
}
page.matched += 1;
})?;
page.next_offset = q.offset + page.records.len() as u64;
page.more = page.next_offset < page.matched;
Ok(page)
}
/// One query. Every filter is optional; an empty one matches every download.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct DownloadQuery {
/// Who downloaded — the authenticated principal (`anonymous` for open reads).
pub ident: Option<String>,
/// The repository downloaded from.
pub repo: Option<String>,
/// From this moment, inclusive (nanoseconds since the Unix epoch).
pub since_nanos: Option<i64>,
/// Up to this moment, exclusive.
pub until_nanos: Option<i64>,
/// At most this many rows (the NEWEST that match); 0 = [`DEFAULT_LIMIT`],
/// capped at [`MAX_LIMIT`].
pub limit: usize,
}
impl DownloadQuery {
/// From wire values where empty / zero means "not named".
pub fn of_wire(
ident: &str,
repo: &str,
since_nanos: i64,
until_nanos: i64,
limit: u64,
) -> Self {
Self {
ident: named(ident),
repo: named(repo),
since_nanos: (since_nanos > 0).then_some(since_nanos),
until_nanos: (until_nanos > 0).then_some(until_nanos),
limit: usize::try_from(limit).unwrap_or(MAX_LIMIT),
}
}
fn effective_limit(&self) -> usize {
match self.limit {
0 => DEFAULT_LIMIT,
n => n.min(MAX_LIMIT),
}
}
fn filter(&self) -> AuditFilter {
AuditFilter {
ident: self.ident.clone(),
repo: self.repo.clone(),
action: Some(AuditAction::Download.as_str().to_string()),
since_nanos: self.since_nanos,
until_nanos: self.until_nanos,
..Default::default()
}
}
}
/// One download, as the log holds it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct DownloadRow {
/// The record's place in the audit chain and its chain hash.
pub seq: u64,
pub hash: String,
/// When, nanoseconds since the Unix epoch.
pub ts_nanos: i64,
/// Who.
pub ident: String,
/// From which repository, and what.
pub repo: String,
pub artifact: String,
/// The client's address, the HTTP-like status and the bytes served.
pub source_ip: String,
pub status: u16,
pub bytes: u64,
/// The sha256 of the bytes when the record carries it.
pub sha256: String,
}
/// The answer: the newest matching rows in log order, how many matched in all,
/// and whether rows were left out by the limit.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct Downloads {
pub rows: Vec<DownloadRow>,
pub matched: u64,
pub truncated: bool,
}
/// **Every download the log holds that `q` matches** — the newest
/// `q.limit` of them, in log order.
pub fn downloads(source: &QuerySource, q: &DownloadQuery) -> Result<Downloads, AuditError> {
let f = q.filter();
f.check()?;
let limit = q.effective_limit();
let mut ring: std::collections::VecDeque<DownloadRow> =
std::collections::VecDeque::with_capacity(limit.min(4096));
let mut total = 0u64;
source.table()?.for_each_row(|r| {
if !f.matches(&r) {
return;
}
total += 1;
if ring.len() == limit {
ring.pop_front();
}
ring.push_back(DownloadRow {
seq: r.seq,
hash: r.hash.to_string(),
ts_nanos: r.ts_nanos,
ident: r.ident.to_string(),
repo: r.repo.to_string(),
artifact: r.artifact.to_string(),
source_ip: r.source_ip.to_string(),
status: r.status,
bytes: r.bytes,
sha256: r.sha256.to_string(),
});
})?;
Ok(Downloads {
truncated: total > ring.len() as u64,
matched: total,
rows: ring.into(),
})
}
#[cfg(test)]
mod tests {
use super::super::{AuditEvent, AuditLog, MemoryAuditLog};
use super::*;
fn at(ev: AuditEvent, ts: i64) -> AuditEvent {
AuditEvent { ts_nanos: ts, ..ev }
}
/// ★ **What did C download**: only C's DOWNLOAD rows, filtered by
/// repository and by time, the newest kept under a limit.
#[test]
fn what_c_downloaded_is_its_download_rows_by_repository_and_time() {
let log = MemoryAuditLog::new();
log.record(at(
AuditEvent::new(
"c",
AuditAction::Download,
"crates",
"a-1.tar",
"10.0.0.9:1",
200,
10,
),
100,
))
.unwrap();
log.record(at(
AuditEvent::new("c", AuditAction::Upload, "crates", "b-1.tar", "", 200, 5),
200,
))
.unwrap();
log.record(at(
AuditEvent::new("d", AuditAction::Download, "crates", "a-1.tar", "", 200, 10),
300,
))
.unwrap();
log.record(at(
AuditEvent::new("c", AuditAction::Download, "pypi", "w.whl", "", 404, 0),
400,
))
.unwrap();
log.record(at(
AuditEvent::new("c", AuditAction::Download, "crates", "c-2.tar", "", 200, 7),
500,
))
.unwrap();
let src = log.query_source().unwrap();
let q = |ident: &str, repo: &str, since: i64, until: i64, limit: u64| {
downloads(
&src,
&DownloadQuery::of_wire(ident, repo, since, until, limit),
)
.unwrap()
};
let c = q("c", "", 0, 0, 0);
assert_eq!(
c.rows
.iter()
.map(|r| r.artifact.as_str())
.collect::<Vec<_>>(),
["a-1.tar", "w.whl", "c-2.tar"],
"C's downloads, no upload, nobody else's"
);
assert_eq!((c.matched, c.truncated), (3, false));
assert_eq!(
(
c.rows[0].source_ip.as_str(),
c.rows[0].bytes,
c.rows[1].status
),
("10.0.0.9:1", 10, 404)
);
let crates = q("", "crates", 0, 0, 0);
assert_eq!(
crates
.rows
.iter()
.map(|r| r.ident.as_str())
.collect::<Vec<_>>(),
["c", "d", "c"],
"every download from the repository"
);
let window = q("c", "crates", 150, 600, 0);
assert_eq!(
window
.rows
.iter()
.map(|r| r.artifact.as_str())
.collect::<Vec<_>>(),
["c-2.tar"],
"since inclusive, until exclusive"
);
let newest = q("c", "", 0, 0, 2);
assert_eq!(
newest
.rows
.iter()
.map(|r| r.artifact.as_str())
.collect::<Vec<_>>(),
["w.whl", "c-2.tar"],
"the newest under the limit"
);
assert_eq!((newest.matched, newest.truncated), (3, true));
assert!(
downloads(&src, &DownloadQuery::of_wire("c", "", 500, 100, 0)).is_err(),
"an empty range is refused"
);
}
/// ★ The rich query: by actor, artifact prefix, action, source address (with
/// or without the port), status and time, paged; the index answers what the
/// in-memory snapshot does.
#[test]
fn the_query_filters_by_actor_artifact_action_address_and_time_and_pages() {
let mem = MemoryAuditLog::new();
let dir = std::env::temp_dir().join(format!(
"holger-q-{}-{}",
std::process::id(),
super::super::now_nanos()
));
let file = super::super::FileAuditLog::new(&dir)
.unwrap()
.with_checkpoint_every(2);
let rows = [
(
"c",
AuditAction::Download,
"crates",
"acme/widget@1",
"10.0.0.9:1",
200,
100,
),
(
"c",
AuditAction::Upload,
"crates",
"acme/widget@2",
"10.0.0.9:2",
200,
200,
),
(
"d",
AuditAction::Download,
"crates",
"acme/gadget@1",
"10.0.0.7:1",
200,
300,
),
(
"c",
AuditAction::Download,
"pypi",
"acme/widget@1",
"[2001:db8::1]:443",
404,
400,
),
(
"c",
AuditAction::Download,
"crates",
"other/x@1",
"10.0.0.9:3",
200,
500,
),
];
for log in [&mem as &dyn AuditLog, &file] {
for (who, a, repo, art, ip, st, ts) in rows {
log.record(at(AuditEvent::new(who, a, repo, art, ip, st, 1), ts))
.unwrap();
}
}
for src in [mem.query_source().unwrap(), file.query_source().unwrap()] {
let q = |f: AuditFilter, offset: u64, limit: usize| {
query(
&src,
&AuditQuery {
filter: f,
offset,
limit,
},
)
.unwrap()
};
let arts = |p: &AuditPage| {
p.records
.iter()
.map(|r| r.event.artifact.clone())
.collect::<Vec<_>>()
};
let by_c = q(
AuditFilter::of_wire("c", "", "acme/widget", "download", "", 0, 0, 0).unwrap(),
0,
0,
);
assert_eq!(arts(&by_c), ["acme/widget@1", "acme/widget@1"]);
let by_ip = q(
AuditFilter::of_wire("", "", "", "", "10.0.0.9", 0, 0, 0).unwrap(),
0,
0,
);
assert_eq!(by_ip.matched, 3, "every port of the address");
let by_ip_port = q(
AuditFilter::of_wire("", "", "", "", "10.0.0.9:2", 0, 0, 0).unwrap(),
0,
0,
);
assert_eq!(arts(&by_ip_port), ["acme/widget@2"]);
let v6 = q(
AuditFilter::of_wire("", "", "", "", "2001:db8::1", 404, 0, 0).unwrap(),
0,
0,
);
assert_eq!(v6.matched, 1);
let window = q(
AuditFilter::of_wire("", "crates", "", "", "", 0, 200, 500).unwrap(),
0,
0,
);
assert_eq!(
arts(&window),
["acme/widget@2", "acme/gadget@1"],
"since inclusive, until exclusive"
);
let p1 = q(AuditFilter::default(), 0, 2);
let p2 = q(AuditFilter::default(), p1.next_offset, 2);
let p3 = q(AuditFilter::default(), p2.next_offset, 2);
assert_eq!(
(p1.matched, p1.more, p2.more, p3.more, p3.records.len()),
(5, true, true, false, 1)
);
assert!(p1.records[0].seq < p2.records[0].seq, "oldest first");
}
assert!(
AuditFilter::of_wire("", "", "", "teleport", "", 0, 0, 0).is_err(),
"an action the log never writes is refused"
);
let _ = std::fs::remove_dir_all(&dir);
}
}