manta_server/server/audit/
mod.rs

1//! Audit trail of every mutating operation, produced to Kafka.
2//!
3//! manta-server is the single chokepoint for every backend call, so it
4//! is where auditing lives — the CLI emits nothing. The pieces:
5//!
6//! - [`audit_requests`] — router middleware that turns every
7//!   mutating `/v2/*` request (any method except GET/HEAD/OPTIONS) and
8//!   every console attach into one [`event::AuditEvent`], emitted
9//!   after the response is known. New routes are covered without any
10//!   per-handler code.
11//! - [`AuditNote`] — extractor through which a handler adds what only
12//!   it knows: a stable `action` name, the targets, the dry-run flag.
13//! - [`AuditLog`] — fans each event out to its [`AuditSink`]s: always
14//!   a `tracing` line on the [`AUDIT_TARGET`] target, plus Kafka
15//!   ([`kafka::KafkaSink`]) when `[auditor.kafka]` is configured.
16//! - [`send_auth_audit`] — the `/v2/auth/token` attempt event.
17//!
18//! Auditing is fail-open: a broker outage is logged and never fails or
19//! slows the request.
20
21mod auth;
22pub mod event;
23pub mod kafka;
24mod middleware;
25
26use std::{sync::Arc, time::Duration};
27
28use serde::{Deserialize, Serialize};
29
30pub use auth::send_auth_audit;
31pub use middleware::{AuditNote, audit_requests};
32
33use event::{AuditEvent, Emitter, EventKind};
34
35/// `tracing` target of the always-on log copy of every audit event.
36pub const AUDIT_TARGET: &str = "manta::audit";
37
38/// `[auditor]` section of `server.toml`.
39#[derive(Serialize, Deserialize, Debug, Clone)]
40pub struct Auditor {
41  /// Kafka destination for audit events.
42  pub kafka: kafka::Kafka,
43}
44
45/// A destination for audit events.
46///
47/// `emit` is called on the request path and must not block: queue the
48/// event and return. Failures are the sink's to log; they never reach
49/// the request.
50pub trait AuditSink: Send + Sync {
51  /// Hand one event to the destination.
52  fn emit(&self, event: &AuditEvent);
53  /// Drain anything still queued, waiting at most `timeout`. Called
54  /// once at shutdown, off the async runtime (it may block).
55  fn flush(&self, _timeout: Duration) {}
56}
57
58/// Logs each event at INFO on [`AUDIT_TARGET`] — its summary as the
59/// message, the full JSON in the `event` field — so the container log
60/// keeps a readable copy even when Kafka is down or not configured.
61struct TracingSink;
62
63impl AuditSink for TracingSink {
64  fn emit(&self, event: &AuditEvent) {
65    match serde_json::to_string(event) {
66      Ok(json) => {
67        tracing::info!(target: AUDIT_TARGET, event = %json, "{}", event.summary);
68      }
69      Err(e) => tracing::warn!(
70        target: AUDIT_TARGET,
71        event_id = %event.event_id,
72        "audit event not serialisable: {e}"
73      ),
74    }
75  }
76}
77
78/// The server's audit trail: builds events and fans them out to sinks.
79/// Cheap to clone.
80#[derive(Clone)]
81pub struct AuditLog {
82  sinks: Vec<Arc<dyn AuditSink>>,
83  emitter: Emitter,
84}
85
86impl Default for AuditLog {
87  /// Log-only audit trail (no Kafka).
88  fn default() -> Self {
89    Self::with_sinks(vec![Arc::new(TracingSink)])
90  }
91}
92
93impl AuditLog {
94  /// The production trail: the log copy plus `kafka`, when configured.
95  pub fn new(kafka: Option<kafka::KafkaSink>) -> Self {
96    let mut sinks: Vec<Arc<dyn AuditSink>> = vec![Arc::new(TracingSink)];
97    if let Some(k) = kafka {
98      sinks.push(Arc::new(k));
99    }
100    Self::with_sinks(sinks)
101  }
102
103  /// A trail with exactly these sinks (tests plug a capturing sink).
104  pub fn with_sinks(sinks: Vec<Arc<dyn AuditSink>>) -> Self {
105    Self {
106      sinks,
107      emitter: Emitter {
108        service: "manta-server",
109        version: env!("CARGO_PKG_VERSION"),
110        host: host_name(),
111      },
112    }
113  }
114
115  /// A fresh event of `kind` with its id, timestamp and emitter
116  /// filled in; the caller sets the rest.
117  pub fn event(&self, kind: EventKind, action: String) -> AuditEvent {
118    AuditEvent {
119      schema_version: event::SCHEMA_VERSION,
120      event_id: uuid::Uuid::new_v4().to_string(),
121      timestamp: chrono::Utc::now()
122        .to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
123      kind,
124      action,
125      summary: String::new(),
126      actor: Default::default(),
127      client: Default::default(),
128      site: None,
129      request: Default::default(),
130      outcome: event::Outcome {
131        result: event::Verdict::Success,
132        status: 0,
133        duration_ms: 0,
134      },
135      dry_run: None,
136      targets: Default::default(),
137      nodes: Default::default(),
138      produced: Default::default(),
139      details: Default::default(),
140      emitter: self.emitter.clone(),
141    }
142  }
143
144  /// Fill in `event.summary` and hand the event to every sink. Never
145  /// blocks, never fails.
146  pub fn emit(&self, mut event: AuditEvent) {
147    event.summary = event.describe();
148    for sink in &self.sinks {
149      sink.emit(&event);
150    }
151  }
152
153  /// Drain every sink, each waiting at most `timeout`. Blocking.
154  pub fn flush(&self, timeout: Duration) {
155    for sink in &self.sinks {
156      sink.flush(timeout);
157    }
158  }
159}
160
161/// Host (in k8s: pod) name — `$HOSTNAME`, else `/etc/hostname`.
162fn host_name() -> Option<String> {
163  std::env::var("HOSTNAME")
164    .ok()
165    .or_else(|| std::fs::read_to_string("/etc/hostname").ok())
166    .map(|h| h.trim().to_string())
167    .filter(|h| !h.is_empty())
168}
169
170/// Test helper: a sink that keeps every event it is handed.
171#[cfg(test)]
172#[derive(Default)]
173pub(crate) struct CaptureSink(pub std::sync::Mutex<Vec<AuditEvent>>);
174
175#[cfg(test)]
176impl AuditSink for CaptureSink {
177  fn emit(&self, event: &AuditEvent) {
178    self.0.lock().unwrap().push(event.clone());
179  }
180}
181
182#[cfg(test)]
183mod tests {
184  use super::*;
185
186  #[test]
187  fn event_prefills_identity_fields() {
188    let log = AuditLog::default();
189    let a = log.event(EventKind::ApiRequest, "x.y".into());
190    let b = log.event(EventKind::ApiRequest, "x.y".into());
191    assert_eq!(a.schema_version, event::SCHEMA_VERSION);
192    assert_ne!(a.event_id, b.event_id, "event ids must be unique");
193    assert!(a.timestamp.ends_with('Z'), "UTC timestamp: {}", a.timestamp);
194    assert_eq!(a.emitter.service, "manta-server");
195    assert_eq!(a.emitter.version, env!("CARGO_PKG_VERSION"));
196  }
197
198  #[test]
199  fn emit_fans_out_to_every_sink() {
200    let first = Arc::new(CaptureSink::default());
201    let second = Arc::new(CaptureSink::default());
202    let log = AuditLog::with_sinks(vec![first.clone(), second.clone()]);
203    log.emit(log.event(EventKind::ApiRequest, "x.y".into()));
204    assert_eq!(first.0.lock().unwrap().len(), 1);
205    assert_eq!(
206      first.0.lock().unwrap()[0].summary,
207      "anonymous: x y → success"
208    );
209    assert_eq!(second.0.lock().unwrap().len(), 1);
210  }
211
212  #[test]
213  fn serialised_event_omits_empty_fields() {
214    let mut e = AuditLog::default().event(EventKind::ApiRequest, "x.y".into());
215    e.emitter.host = None;
216    let json = serde_json::to_string(&e).unwrap();
217    assert!(!json.contains("null"), "{json}");
218    for absent in ["\"dry_run\"", "\"site\"", "\"targets\"", "\"details\""] {
219      assert!(!json.contains(absent), "{absent} in {json}");
220    }
221  }
222
223  /// The schema has no field that could hold a credential. Every
224  /// optional field that exists today is populated, so every key the
225  /// schema can emit is checked. A new always-present `token`-like
226  /// field fails here; a new *optional* one does only once it is
227  /// populated below too.
228  #[test]
229  fn schema_has_no_credential_field() {
230    let mut e = AuditLog::default().event(EventKind::ApiRequest, "x.y".into());
231    e.actor = event::Actor {
232      username: Some("u".into()),
233      name: Some("n".into()),
234      subject: Some("s".into()),
235    };
236    e.client = event::Client {
237      ip: Some("i".into()),
238      forwarded_for: Some("f".into()),
239    };
240    e.site = Some("s".into());
241    e.request.query = Some("q".into());
242    e.dry_run = Some(true);
243    e.emitter.host = Some("h".into());
244
245    fn keys(v: &serde_json::Value, out: &mut Vec<String>) {
246      if let Some(map) = v.as_object() {
247        for (k, v) in map {
248          out.push(k.clone());
249          keys(v, out);
250        }
251      }
252    }
253    let mut all = Vec::new();
254    keys(&serde_json::to_value(&e).unwrap(), &mut all);
255    for key in ["forwarded_for", "query", "dry_run", "host", "subject"] {
256      assert!(
257        all.iter().any(|k| k == key),
258        "`{key}` not populated: {all:?}"
259      );
260    }
261    for key in &all {
262      for forbidden in ["token", "password", "authorization", "secret"] {
263        assert!(!key.contains(forbidden), "schema field `{key}`");
264      }
265    }
266  }
267}
manta_server/server/audit/mod.rs

manta_server/server/audit/
mod.rs
