diff --git a/Cargo.lock b/Cargo.lock index ed472b8..68d0aba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1686,6 +1686,7 @@ dependencies = [ "axum-extra", "axum-streams", "bytes", + "chrono", "config", "dicom", "dicom-json", diff --git a/Cargo.toml b/Cargo.toml index 6a45930..ff424da 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,6 +30,8 @@ serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" # Logging tracing = "0.1.41" +# Timestamps for the audit log; already transitive via dicom-core. +chrono = { version = "0.4.42", default-features = false, features = ["clock"] } tracing-subscriber = { version = "0.3.20", features = ["env-filter"] } # Convenient error handling thiserror = "2.0.17" diff --git a/docs/topics/configuration.md b/docs/topics/configuration.md index 547443f..c23c467 100644 --- a/docs/topics/configuration.md +++ b/docs/topics/configuration.md @@ -81,6 +81,8 @@ aets: telemetry: sentry: https://sentry.local/dsn level: INFO + audit: + enabled: false ``` @@ -98,6 +100,20 @@ telemetry:
  • TRACE
    • + + Structured access-audit logging (default false). + When enabled, every HTTP request emits one self-contained JSON line + on stdout: timestamp, caller identity (read from the + X-Auth-Request-User/X-Auth-Request-Email + headers an authenticating reverse proxy such as oauth2-proxy + injects), source IP, method, path, the DICOM coordinates + (aet, study, series, + instance), response status and duration. Intended for + healthcare access-audit requirements; delivery is fail-open + (bounded buffer — a slow log consumer never blocks requests, drops + are counted and logged). The identity headers are trustworthy only + when the proxy is the sole ingress to DICOM-RST. +     ## Global Server Config diff --git a/src/audit.rs b/src/audit.rs new file mode 100644 index 0000000..351c74b --- /dev/null +++ b/src/audit.rs @@ -0,0 +1,230 @@ +//! Structured access-audit logging. +//! +//! When enabled (`telemetry.audit.enabled: true`), every HTTP request emits +//! one self-contained JSON line on stdout describing WHO accessed WHAT: +//! +//! ```json +//! {"audit":"http-access","ts":"2026-08-17T17:16:55Z","user":"alex@example.com", +//! "subject":"5939ae08-…","source":"10.244.9.49","method":"GET", +//! "path":"/app/dicom-rst/aets/GEPACS/studies/1.2.840…","aet":"GEPACS", +//! "study":"1.2.840…","status":200,"duration_ms":4886} +//! ``` +//! +//! Identity is read from the `X-Auth-Request-User` / `X-Auth-Request-Email` +//! headers that an authenticating reverse proxy (e.g. oauth2-proxy with +//! `set_xauthrequest`) injects. DICOM-RST itself performs no authentication +//! (see #15/#42): these fields are TRUSTWORTHY ONLY when the deployment +//! guarantees that the proxy is the sole ingress. The record is emitted +//! regardless — an absent identity is itself audit-relevant. +//! +//! Delivery is FAIL-OPEN by design: records flow through a bounded channel +//! to a writer task; when the buffer is full the record is dropped, a +//! counter increments and a warning is logged — a slow disk or collector +//! never blocks request handling. Deployments with stricter requirements +//! should alert on the drop warnings. + +use std::io::Write; +use std::sync::atomic::{AtomicU64, Ordering}; + +use axum::extract::{RawPathParams, Request, State}; +use axum::middleware::Next; +use axum::response::Response; +use chrono::{SecondsFormat, Utc}; +use serde::Serialize; +use tokio::sync::mpsc; +use tracing::warn; + +/// One audit record per HTTP request. +#[derive(Debug, Serialize)] +pub struct AuditRecord { + /// Discriminator for log pipelines; always `"http-access"` for now. + pub audit: &'static str, + /// Wall-clock request completion time (UTC, RFC 3339, second precision). + pub ts: String, + /// `X-Auth-Request-Email` from the authenticating proxy, if present. + #[serde(skip_serializing_if = "Option::is_none")] + pub user: Option, + /// `X-Auth-Request-User` (the OIDC subject), if present. + #[serde(skip_serializing_if = "Option::is_none")] + pub subject: Option, + /// First `X-Forwarded-For` entry, if present. + #[serde(skip_serializing_if = "Option::is_none")] + pub source: Option, + pub method: String, + /// Full request path and query. QIDO match parameters are part of + /// "which data was accessed" and are deliberately included. + pub path: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub aet: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub study: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub series: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub instance: Option, + pub status: u16, + pub duration_ms: u128, + #[serde(skip_serializing_if = "Option::is_none")] + pub user_agent: Option, +} + +/// Cloneable handle to the audit writer. `None` inside means auditing is +/// disabled and the middleware is a no-op. +#[derive(Clone)] +pub struct AuditSink { + tx: Option>, +} + +/// Records dropped because the buffer was full (fail-open pressure valve). +static DROPPED: AtomicU64 = AtomicU64::new(0); + +const BUFFER: usize = 1024; + +impl AuditSink { + /// Create the sink and, when enabled, spawn the stdout writer task. + pub fn new(enabled: bool) -> Self { + if !enabled { + return Self { tx: None }; + } + let (tx, mut rx) = mpsc::channel::(BUFFER); + tokio::spawn(async move { + // One locked write per record keeps lines atomic alongside the + // regular tracing output on the same stream. + while let Some(record) = rx.recv().await { + match serde_json::to_string(&record) { + Ok(mut line) => { + line.push('\n'); + let mut stdout = std::io::stdout().lock(); + let _ = stdout.write_all(line.as_bytes()); + } + Err(err) => warn!("failed to serialize audit record: {err}"), + } + } + }); + Self { tx: Some(tx) } + } + + fn emit(&self, record: AuditRecord) { + let Some(tx) = &self.tx else { return }; + if tx.try_send(record).is_err() { + let dropped = DROPPED.fetch_add(1, Ordering::Relaxed) + 1; + // Every drop is a warning-worthy event, but do not spam a + // saturated system: log the first and then every 100th. + if dropped == 1 || dropped.is_multiple_of(100) { + warn!("audit buffer full: {dropped} record(s) dropped so far (fail-open)"); + } + } + } +} + +/// Axum middleware producing one [`AuditRecord`] per request. +/// +/// Attach with `axum::middleware::from_fn_with_state(sink, audit::middleware)` +/// OUTSIDE the timeout layer, so timed-out requests are recorded with their +/// 408 as well. +pub async fn middleware( + State(sink): State, + params: RawPathParams, + request: Request, + next: Next, +) -> Response { + if sink.tx.is_none() { + return next.run(request).await; + } + + let mut aet = None; + let mut study = None; + let mut series = None; + let mut instance = None; + for (name, value) in ¶ms { + match name { + "aet" => aet = Some(value.to_owned()), + "study" => study = Some(value.to_owned()), + "series" => series = Some(value.to_owned()), + "instance" => instance = Some(value.to_owned()), + _ => {} + } + } + + // Extract everything BEFORE the await, inside a block that ends first: + // a closure borrowing `&Request` held across `next.run().await` makes + // the future `!Send` (axum's `Body` is `!Sync`), failing the middleware + // `Service` bound with a famously opaque error. + let (user, subject, source, user_agent) = { + let headers = request.headers(); + let get = |name: &str| { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned) + }; + ( + get("x-auth-request-email"), + get("x-auth-request-user"), + get("x-forwarded-for").map(|forwarded| { + forwarded + .split(',') + .next() + .unwrap_or_default() + .trim() + .to_owned() + }), + get("user-agent"), + ) + }; + let method = request.method().to_string(); + let path = request.uri().to_string(); + + let started = std::time::Instant::now(); + let response = next.run(request).await; + + sink.emit(AuditRecord { + audit: "http-access", + ts: Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), + user, + subject, + source, + method, + path, + aet, + study, + series, + instance, + status: response.status().as_u16(), + duration_ms: started.elapsed().as_millis(), + user_agent, + }); + + response +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn record_serializes_without_absent_fields() { + let record = AuditRecord { + audit: "http-access", + ts: "2026-08-17T00:00:00Z".to_owned(), + user: None, + subject: None, + source: None, + method: "GET".to_owned(), + path: "/aets".to_owned(), + aet: None, + study: None, + series: None, + instance: None, + status: 200, + duration_ms: 3, + user_agent: None, + }; + let json = serde_json::to_string(&record).expect("serialize"); + assert!( + !json.contains("user"), + "absent fields must be omitted: {json}" + ); + assert!(json.contains("\"audit\":\"http-access\"")); + } +} diff --git a/src/backend/dimse/cmove/movescu.rs b/src/backend/dimse/cmove/movescu.rs index c99f611..1cc087a 100644 --- a/src/backend/dimse/cmove/movescu.rs +++ b/src/backend/dimse/cmove/movescu.rs @@ -25,6 +25,15 @@ impl MoveServiceClassUser { #[instrument(skip_all, name = "MOVE-SCU")] #[allow(clippy::significant_drop_tightening)] pub async fn invoke(&self, request: CompositeMoveRequest) -> Result<(), MoveError> { + // Surface WHICH study the C-MOVE concerns — the audit trail needs + // more than "a move happened". + let study_uid = request + .identifier + .element(tags::STUDY_INSTANCE_UID) + .ok() + .and_then(|element| element.to_str().ok()) + .map(|uid| uid.trim_end_matches('\0').to_owned()) + .unwrap_or_default(); let association = self .pool .get(PresentationParameter { @@ -54,7 +63,7 @@ impl MoveServiceClassUser { match status_type { StatusType::Success => { - info!("C-MOVE completed successfully"); + info!(study_uid, "C-MOVE completed successfully"); break; } StatusType::Pending => { diff --git a/src/config/mod.rs b/src/config/mod.rs index d7047ef..7986c52 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -340,6 +340,9 @@ pub struct TelemetryConfig { pub sentry: Option, #[serde(deserialize_with = "deserialize_log_level")] pub level: tracing::Level, + /// Structured access-audit logging — see [`crate::audit`]. + #[serde(default)] + pub audit: AuditConfig, } impl Default for TelemetryConfig { @@ -347,10 +350,24 @@ impl Default for TelemetryConfig { Self { sentry: None, level: tracing::Level::INFO, + audit: AuditConfig::default(), } } } +/// Configuration for the structured access-audit log ([`crate::audit`]). +/// +/// Disabled by default: enabling it emits one JSON line per HTTP request on +/// stdout, carrying the identity injected by an authenticating reverse proxy +/// (`X-Auth-Request-*`) plus the DICOM resource coordinates. Delivery is +/// fail-open (bounded buffer, drops are counted and logged). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct AuditConfig { + #[serde(default)] + pub enabled: bool, +} + /// Deserializer for [`tracing::Level`] as it does not implement [Deserialize] fn deserialize_log_level<'de, D>(deserializer: D) -> Result where diff --git a/src/main.rs b/src/main.rs index 6c23b68..7922b76 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,7 @@ #![allow(clippy::multiple_crate_versions)] pub(crate) mod api; +pub(crate) mod audit; pub(crate) mod backend; pub(crate) mod config; pub(crate) mod rendering; @@ -17,6 +18,7 @@ use axum::extract::{DefaultBodyLimit, Request}; use axum::http::StatusCode; use axum::response::Response; use axum::ServiceExt; +use std::io::IsTerminal; use std::net::SocketAddr; use std::time::Duration; use tokio::net::TcpListener; @@ -46,6 +48,9 @@ fn init_logger(level: tracing::Level) { .with( tracing_subscriber::fmt::layer() .compact() + // ANSI escapes belong on terminals, not in collected pod + // logs (they garble downstream log pipelines). + .with_ansi(std::io::stdout().is_terminal()) .with_file(false) .with_line_number(false) .with_target(false), @@ -134,6 +139,8 @@ async fn run(config: AppConfig) -> anyhow::Result<()> { }); } + let audit_sink = audit::AuditSink::new(config.telemetry.audit.enabled); + let app = api::routes(&config.server.http.base_path) .layer(CorsLayer::permissive()) .layer(axum::middleware::from_fn(add_common_headers)) @@ -148,6 +155,12 @@ async fn run(config: AppConfig) -> anyhow::Result<()> { StatusCode::REQUEST_TIMEOUT, Duration::from_secs(config.server.http.request_timeout), )) + // Outside the timeout layer, so timed-out requests are audited + // with their 408 as well. No-op unless telemetry.audit.enabled. + .layer(axum::middleware::from_fn_with_state( + audit_sink, + audit::middleware, + )) .with_state(app_state); let app = NormalizePathLayer::trim_trailing_slash().layer(app);