manta_server/server/audit/
event.rs

1//! The audit event wire shape (`schema_version` 1).
2//!
3//! One [`AuditEvent`] is emitted per audited request, after the
4//! response is known, so every event carries both *who asked for
5//! what* and *how it ended*. The JSON produced by `serde` is the
6//! contract with whoever consumes the Kafka topic: add fields freely,
7//! but rename or remove one only together with a `SCHEMA_VERSION`
8//! bump. A field with no value (`None`, an empty map) is left out of
9//! the JSON rather than written as `null` / `{}`.
10//!
11//! Nothing secret ever lands here by construction: there is no field
12//! for a token, a password or a raw request body. Handlers describe
13//! their operation through [`super::AuditNote`] with values they pick
14//! explicitly.
15
16use std::collections::BTreeMap;
17
18use axum::http::StatusCode;
19use serde::Serialize;
20
21/// Version of the event shape. Bump on any rename/removal.
22pub const SCHEMA_VERSION: u32 = 1;
23
24/// What kind of activity the event records.
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
26#[serde(rename_all = "snake_case")]
27pub enum EventKind {
28  /// A mutating `/v2/*` API call (any method other than GET/HEAD/OPTIONS).
29  ApiRequest,
30  /// An interactive console attach (node or CFS session) over WebSocket.
31  ConsoleSession,
32  /// A `/v2/auth/token` credential exchange.
33  AuthAttempt,
34}
35
36/// How the request ended, derived from the response status.
37#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
38#[serde(rename_all = "snake_case")]
39pub enum Verdict {
40  /// 1xx–3xx: the operation went through (101 = console attached).
41  Success,
42  /// 401/403: the caller was not allowed to do it.
43  Denied,
44  /// 408: the server gave up on the request (`TimeoutLayer`).
45  TimedOut,
46  /// Any other 4xx: the request was refused as invalid.
47  Rejected,
48  /// 5xx: the server or a backend failed while doing it.
49  Failed,
50}
51
52impl Verdict {
53  /// Classify a response status.
54  pub fn from_status(status: StatusCode) -> Self {
55    match status.as_u16() {
56      401 | 403 => Self::Denied,
57      408 => Self::TimedOut,
58      400..=499 => Self::Rejected,
59      500.. => Self::Failed,
60      _ => Self::Success,
61    }
62  }
63}
64
65/// Who performed the operation, as claimed by the inbound bearer token.
66///
67/// The token signature is not verified locally (see
68/// `manta_shared::common::jwt_ops`); the backend verifies it on use.
69/// The identity is therefore authoritative when `outcome.result` is
70/// `success`, and a *claim* otherwise.
71#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
72pub struct Actor {
73  /// `preferred_username` claim — the login name.
74  #[serde(skip_serializing_if = "Option::is_none")]
75  pub username: Option<String>,
76  /// `name` claim — the display name.
77  #[serde(skip_serializing_if = "Option::is_none")]
78  pub name: Option<String>,
79  /// `sub` claim — the stable IdP subject id.
80  #[serde(skip_serializing_if = "Option::is_none")]
81  pub subject: Option<String>,
82}
83
84/// Where the request came from.
85#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
86pub struct Client {
87  /// TCP peer address of the connection (a reverse proxy, if any).
88  #[serde(skip_serializing_if = "Option::is_none")]
89  pub ip: Option<String>,
90  /// Raw `X-Forwarded-For` header, recorded as-is and **not**
91  /// trusted: any client can set it.
92  #[serde(skip_serializing_if = "Option::is_none")]
93  pub forwarded_for: Option<String>,
94}
95
96/// The HTTP request as received.
97#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
98pub struct Request {
99  /// HTTP method, e.g. `DELETE`.
100  pub method: String,
101  /// Route template, e.g. `/v2/groups/{label}`.
102  pub route: String,
103  /// Concrete path, e.g. `/v2/groups/zinal`.
104  pub path: String,
105  /// Raw query string, when present.
106  #[serde(skip_serializing_if = "Option::is_none")]
107  pub query: Option<String>,
108}
109
110/// How the request ended.
111#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
112pub struct Outcome {
113  /// Coarse classification of [`Self::status`].
114  pub result: Verdict,
115  /// HTTP response status code.
116  pub status: u16,
117  /// Wall time from the request reaching the audit layer to the
118  /// response leaving the handler. For a console session, the time to
119  /// attach (the session itself outlives the response).
120  pub duration_ms: u64,
121}
122
123/// Which manta-server instance emitted the event.
124#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
125pub struct Emitter {
126  /// Always `manta-server`.
127  pub service: &'static str,
128  /// `CARGO_PKG_VERSION` of the running binary.
129  pub version: &'static str,
130  /// Host (pod) name, when it can be determined.
131  #[serde(skip_serializing_if = "Option::is_none")]
132  pub host: Option<String>,
133}
134
135/// One audit record. See the module docs for the stability contract.
136#[derive(Debug, Clone, PartialEq, Serialize)]
137pub struct AuditEvent {
138  /// Always [`SCHEMA_VERSION`].
139  pub schema_version: u32,
140  /// Random v4 UUID; also the Kafka message key.
141  pub event_id: String,
142  /// RFC 3339 UTC timestamp (millisecond precision) of emission.
143  pub timestamp: String,
144  /// What kind of activity this is.
145  pub kind: EventKind,
146  /// Stable operation name chosen by the handler, e.g.
147  /// `group.delete`; `"<METHOD> <route>"` when the handler did not
148  /// name it (e.g. the request was rejected before the handler ran).
149  pub action: String,
150  /// One human-readable line saying who did what, where, and how it
151  /// ended — see [`AuditEvent::describe`]. Derived from the other
152  /// fields when the event is emitted; for reading, not for parsing.
153  pub summary: String,
154  /// Who did it.
155  pub actor: Actor,
156  /// From where.
157  pub client: Client,
158  /// Target site (`X-Manta-Site`), when supplied.
159  #[serde(skip_serializing_if = "Option::is_none")]
160  pub site: Option<String>,
161  /// The request as received.
162  pub request: Request,
163  /// How it ended.
164  pub outcome: Outcome,
165  /// `true` when the caller asked for a dry run (nothing changed),
166  /// `false` for a real run of an operation that has a dry-run mode;
167  /// absent when the operation has none or the handler never ran.
168  #[serde(skip_serializing_if = "Option::is_none")]
169  pub dry_run: Option<bool>,
170  /// Objects the operation acts on, keyed by object kind (`group`,
171  /// `node`, `hosts`, `image`, …). Values are as the caller supplied
172  /// them (a hosts expression is not expanded; see [`Self::nodes`]),
173  /// except where the caller selected by pattern: `configuration.delete`
174  /// records the configurations, sessions, templates and images it
175  /// matched.
176  #[serde(skip_serializing_if = "BTreeMap::is_empty")]
177  pub targets: BTreeMap<String, Vec<String>>,
178  /// Concrete node xnames the server resolved the request to, when it
179  /// expanded a hosts expression or a group itself (power, boot,
180  /// kernel parameters, runtime configuration, migrate). Power, boot
181  /// and kernel parameters record it before the operation runs, so it
182  /// is present on dry runs and failures too; runtime configuration
183  /// and migrate resolve and apply in one service call, so there it
184  /// is recorded only once that call succeeds.
185  #[serde(skip_serializing_if = "Vec::is_empty")]
186  pub nodes: Vec<String>,
187  /// What the operation created or started, by kind (`transition`,
188  /// `session`, `configuration`, `image`, `hostname`) — the ids to
189  /// look the operation up by in the backend. Only on success, never
190  /// on a dry run.
191  #[serde(skip_serializing_if = "BTreeMap::is_empty")]
192  pub produced: BTreeMap<String, String>,
193  /// Further operation parameters a handler chose to record (e.g.
194  /// `force`, the power action, the kernel parameters).
195  #[serde(skip_serializing_if = "serde_json::Map::is_empty")]
196  pub details: serde_json::Map<String, serde_json::Value>,
197  /// Which server emitted it.
198  pub emitter: Emitter,
199}
200
201/// Most values of one target kind spelled out in a summary.
202const SUMMARY_MAX_VALUES: usize = 5;
203
204impl AuditEvent {
205  /// One line for humans, built from the structured fields:
206  ///
207  /// ```text
208  /// rjanalik @ alps: power off nid001288 (1 node, force) → success, transition 4c1e…
209  /// alice @ alps: image delete a b (dry run) → success
210  /// anonymous: DELETE /v2/groups/zinal → denied (401)
211  /// ```
212  ///
213  /// Control characters are escaped (`\n` → `\\n`): the line is
214  /// logged as plain text and its values are caller-controlled.
215  ///
216  /// Every route gets one with no per-handler code: the verb is the
217  /// handler's `action` with its separators turned into spaces, or
218  /// method and path when no handler named it.
219  pub fn describe(&self) -> String {
220    use std::fmt::Write as _;
221
222    let who = self
223      .actor
224      .username
225      .as_deref()
226      .or(self.actor.subject.as_deref())
227      .unwrap_or("anonymous");
228    let mut line = match &self.site {
229      Some(site) => format!("{who} @ {site}:"),
230      None => format!("{who}:"),
231    };
232
233    if self.action.contains(' ') {
234      // Fallback `"<METHOD> <route>"`: show the concrete path instead.
235      let _ = write!(line, " {} {}", self.request.method, self.request.path);
236    } else {
237      let _ = write!(line, " {}", self.action.replace(['.', '_'], " "));
238      for values in self.targets.values() {
239        for v in values.iter().take(SUMMARY_MAX_VALUES) {
240          let _ = write!(line, " {v}");
241        }
242        if values.len() > SUMMARY_MAX_VALUES {
243          let _ =
244            write!(line, " (+{} more)", values.len() - SUMMARY_MAX_VALUES);
245        }
246      }
247    }
248
249    let mut notes = Vec::new();
250    match self.nodes.len() {
251      0 => {}
252      1 => notes.push("1 node".to_string()),
253      n => notes.push(format!("{n} nodes")),
254    }
255    if self.dry_run == Some(true) {
256      notes.push("dry run".to_string());
257    }
258    if self.details.get("force") == Some(&serde_json::Value::Bool(true)) {
259      notes.push("force".to_string());
260    }
261    if !notes.is_empty() {
262      let _ = write!(line, " ({})", notes.join(", "));
263    }
264
265    let result = serde_json::to_value(self.outcome.result)
266      .ok()
267      .and_then(|v| v.as_str().map(str::to_string))
268      .unwrap_or_default()
269      .replace('_', " ");
270    let _ = write!(line, " → {result}");
271    if self.outcome.result != Verdict::Success {
272      let _ = write!(line, " ({})", self.outcome.status);
273    }
274    for (kind, id) in &self.produced {
275      let _ = write!(line, ", {kind} {id}");
276    }
277    // Most of the line is caller-controlled (unverified token claims,
278    // path segments, body strings), and the summary is a plain-text
279    // log message: escape control characters so a `\n` cannot forge
280    // a second log line.
281    line
282      .chars()
283      .flat_map(|c| {
284        let escaped: Vec<char> = if c.is_control() {
285          c.escape_default().collect()
286        } else {
287          vec![c]
288        };
289        escaped
290      })
291      .collect()
292  }
293}
294
295#[cfg(test)]
296mod tests {
297  use super::*;
298
299  fn event(action: &str) -> AuditEvent {
300    AuditEvent {
301      schema_version: SCHEMA_VERSION,
302      event_id: "e".into(),
303      timestamp: "t".into(),
304      kind: EventKind::ApiRequest,
305      action: action.into(),
306      summary: String::new(),
307      actor: Actor {
308        username: Some("rjanalik".into()),
309        ..Actor::default()
310      },
311      client: Client::default(),
312      site: Some("alps".into()),
313      request: Request {
314        method: "POST".into(),
315        route: "/v2/power".into(),
316        path: "/v2/power".into(),
317        query: None,
318      },
319      outcome: Outcome {
320        result: Verdict::Success,
321        status: 200,
322        duration_ms: 1,
323      },
324      dry_run: None,
325      targets: BTreeMap::new(),
326      nodes: Vec::new(),
327      produced: BTreeMap::new(),
328      details: serde_json::Map::new(),
329      emitter: Emitter {
330        service: "manta-server",
331        version: "0",
332        host: None,
333      },
334    }
335  }
336
337  #[test]
338  fn describe_names_who_what_where_and_how_it_ended() {
339    let mut e = event("power.off");
340    e.targets.insert("hosts".into(), vec!["nid001288".into()]);
341    e.nodes = vec!["x1000c0s0b0n0".into()];
342    e.details.insert("force".into(), serde_json::json!(true));
343    e.produced.insert("transition".into(), "abc-123".into());
344    assert_eq!(
345      e.describe(),
346      "rjanalik @ alps: power off nid001288 (1 node, force) → success, \
347       transition abc-123"
348    );
349  }
350
351  #[test]
352  fn describe_marks_dry_runs_and_failures() {
353    let mut e = event("boot_parameters.delete");
354    e.dry_run = Some(true);
355    e.outcome = Outcome {
356      result: Verdict::TimedOut,
357      status: 408,
358      duration_ms: 1,
359    };
360    assert_eq!(
361      e.describe(),
362      "rjanalik @ alps: boot parameters delete (dry run) → timed out (408)"
363    );
364  }
365
366  #[test]
367  fn describe_escapes_control_characters() {
368    let mut e = event("group.delete");
369    e.actor.username = Some("x\nINFO forged".into());
370    e.targets
371      .insert("group".into(), vec!["g\r\u{1b}[31m".into()]);
372    let line = e.describe();
373    assert!(!line.chars().any(char::is_control), "{line:?}");
374    assert_eq!(
375      line,
376      "x\\nINFO forged @ alps: group delete g\\r\\u{1b}[31m → success"
377    );
378  }
379
380  #[test]
381  fn describe_truncates_long_target_lists() {
382    let mut e = event("image.delete");
383    e.targets
384      .insert("image".into(), (1..=7).map(|i| format!("i{i}")).collect());
385    assert_eq!(
386      e.describe(),
387      "rjanalik @ alps: image delete i1 i2 i3 i4 i5 (+2 more) → success"
388    );
389  }
390
391  #[test]
392  fn describe_falls_back_to_method_and_path() {
393    let mut e = event("DELETE /v2/groups/{label}");
394    e.actor = Actor::default();
395    e.site = None;
396    e.request.method = "DELETE".into();
397    e.request.path = "/v2/groups/zinal".into();
398    e.outcome = Outcome {
399      result: Verdict::Denied,
400      status: 401,
401      duration_ms: 1,
402    };
403    assert_eq!(
404      e.describe(),
405      "anonymous: DELETE /v2/groups/zinal → denied (401)"
406    );
407  }
408
409  #[test]
410  fn result_classifies_statuses() {
411    let r = |s: u16| Verdict::from_status(StatusCode::from_u16(s).unwrap());
412    assert_eq!(r(101), Verdict::Success);
413    assert_eq!(r(200), Verdict::Success);
414    assert_eq!(r(204), Verdict::Success);
415    assert_eq!(r(400), Verdict::Rejected);
416    assert_eq!(r(401), Verdict::Denied);
417    assert_eq!(r(403), Verdict::Denied);
418    assert_eq!(r(404), Verdict::Rejected);
419    assert_eq!(r(408), Verdict::TimedOut);
420    assert_eq!(r(409), Verdict::Rejected);
421    assert_eq!(r(500), Verdict::Failed);
422    assert_eq!(r(502), Verdict::Failed);
423  }
424
425  #[test]
426  fn result_serialises_snake_case() {
427    assert_eq!(
428      serde_json::to_value(Verdict::TimedOut).unwrap(),
429      serde_json::json!("timed_out")
430    );
431    assert_eq!(
432      serde_json::to_value(EventKind::ConsoleSession).unwrap(),
433      serde_json::json!("console_session")
434    );
435  }
436}
manta_server/server/audit/event.rs

manta_server/server/audit/
event.rs
