manta_server/server/audit/
middleware.rs

1//! The audit middleware and the [`AuditNote`] extractor.
2
3use std::{
4  collections::BTreeMap,
5  convert::Infallible,
6  net::SocketAddr,
7  sync::{Arc, Mutex},
8  time::Instant,
9};
10
11use axum::{
12  extract::{
13    ConnectInfo, FromRequestParts, MatchedPath, OriginalUri, Request, State,
14  },
15  http::{HeaderMap, Method, StatusCode, header, request::Parts},
16  middleware::Next,
17  response::Response,
18};
19use manta_shared::common::jwt_ops;
20use serde::Serialize;
21
22use super::event::{self, Actor, Client, EventKind, Verdict};
23use crate::server::ServerState;
24
25/// What a handler adds to its request's audit event.
26#[derive(Default)]
27struct Note {
28  action: Option<&'static str>,
29  dry_run: Option<bool>,
30  targets: BTreeMap<String, Vec<String>>,
31  nodes: Vec<String>,
32  produced: BTreeMap<String, String>,
33  details: serde_json::Map<String, serde_json::Value>,
34}
35
36/// Handler-side handle on the current request's audit event.
37///
38/// Extract it next to `RequestCtx` and describe the operation **before**
39/// the first fallible call, so the description is kept on the error
40/// path too:
41///
42/// ```ignore
43/// note.action("group.delete").target("group", &label).detail("force", q.force);
44/// ```
45///
46/// Outside an audited request (GET routes, unit tests) the handle is
47/// inert and every call is a no-op. Never put a credential in it.
48#[derive(Clone, Default)]
49pub struct AuditNote(Option<Arc<Mutex<Note>>>);
50
51impl AuditNote {
52  fn with(&self, f: impl FnOnce(&mut Note)) -> &Self {
53    if let Some(note) = &self.0 {
54      // A poisoned lock means a handler panicked mid-note; the
55      // half-written note is still the best record we have.
56      f(&mut note.lock().unwrap_or_else(|p| p.into_inner()));
57    }
58    self
59  }
60
61  /// Stable operation name, `<resource>.<verb>` (e.g. `group.delete`).
62  pub fn action(&self, action: &'static str) -> &Self {
63    self.with(|n| n.action = Some(action))
64  }
65
66  /// Whether the caller asked for a dry run.
67  pub fn dry_run(&self, dry_run: bool) -> &Self {
68    self.with(|n| n.dry_run = Some(dry_run))
69  }
70
71  /// Add one object of `kind` (`group`, `node`, `hosts`, …) that the
72  /// operation acts on.
73  pub fn target(&self, kind: &str, value: impl Into<String>) -> &Self {
74    self.with(|n| {
75      n.targets
76        .entry(kind.to_string())
77        .or_default()
78        .push(value.into());
79    })
80  }
81
82  /// Add several objects of `kind`. No values, no entry.
83  pub fn targets<I, S>(&self, kind: &str, values: I) -> &Self
84  where
85    I: IntoIterator<Item = S>,
86    S: Into<String>,
87  {
88    self.with(|n| {
89      let mut values = values.into_iter().map(Into::into).peekable();
90      if values.peek().is_some() {
91        n.targets
92          .entry(kind.to_string())
93          .or_default()
94          .extend(values);
95      }
96    })
97  }
98
99  /// Record the node xnames the server resolved the targets to.
100  pub fn nodes<I, S>(&self, values: I) -> &Self
101  where
102    I: IntoIterator<Item = S>,
103    S: Into<String>,
104  {
105    self.with(|n| n.nodes.extend(values.into_iter().map(Into::into)))
106  }
107
108  /// Record the id of something the operation created or started
109  /// (`transition`, `session`, …). Call it only once that happened.
110  pub fn produced(&self, kind: &str, id: impl Into<String>) -> &Self {
111    self.with(|n| {
112      n.produced.insert(kind.to_string(), id.into());
113    })
114  }
115
116  /// Record a further operation parameter. Silently skipped if it
117  /// does not serialise.
118  pub fn detail(&self, key: &str, value: impl Serialize) -> &Self {
119    self.with(|n| {
120      if let Ok(v) = serde_json::to_value(value) {
121        n.details.insert(key.to_string(), v);
122      }
123    })
124  }
125}
126
127impl<S: Send + Sync> FromRequestParts<S> for AuditNote {
128  type Rejection = Infallible;
129
130  async fn from_request_parts(
131    parts: &mut Parts,
132    _state: &S,
133  ) -> Result<Self, Self::Rejection> {
134    Ok(
135      parts
136        .extensions
137        .get::<AuditNote>()
138        .cloned()
139        .unwrap_or_default(),
140    )
141  }
142}
143
144/// What the middleware audits: mutating methods, plus console attaches
145/// (a GET upgrade, but an interactive root shell on a node).
146fn audited_kind(method: &Method, route: &str) -> Option<EventKind> {
147  if route.ends_with("/console") {
148    Some(EventKind::ConsoleSession)
149  } else if matches!(*method, Method::GET | Method::HEAD | Method::OPTIONS) {
150    None
151  } else {
152    Some(EventKind::ApiRequest)
153  }
154}
155
156/// Identity claims of the inbound bearer token. Reads the
157/// `Authorization` header itself rather than any processed token, so
158/// the event names the caller as they authenticated.
159fn actor(headers: &HeaderMap) -> Actor {
160  let Some(token) = headers
161    .get(header::AUTHORIZATION)
162    .and_then(|v| v.to_str().ok())
163  else {
164    return Actor::default();
165  };
166  let claim = |name| jwt_ops::get_string_claim(token, name).ok().flatten();
167  Actor {
168    username: claim("preferred_username"),
169    name: claim("name"),
170    subject: claim("sub"),
171  }
172}
173
174fn header_str(headers: &HeaderMap, name: &str) -> Option<String> {
175  headers
176    .get(name)
177    .and_then(|v| v.to_str().ok())
178    .map(str::to_string)
179}
180
181/// Router middleware: emit one [`event::AuditEvent`] per audited
182/// request once its response is known. Layered outside the request
183/// timeout, so timed-out requests are recorded too. Requests that
184/// matched no route (404) or no method (405) are not audited — nothing
185/// was attempted.
186pub async fn audit_requests(
187  State(state): State<Arc<ServerState>>,
188  mut request: Request,
189  next: Next,
190) -> Response {
191  let Some(route) = request
192    .extensions()
193    .get::<MatchedPath>()
194    .map(|p| p.as_str().to_string())
195  else {
196    return next.run(request).await;
197  };
198  let Some(kind) = audited_kind(request.method(), &route) else {
199    return next.run(request).await;
200  };
201
202  let note = AuditNote(Some(Arc::default()));
203  request.extensions_mut().insert(note.clone());
204
205  let headers = request.headers();
206  let actor = actor(headers);
207  let site = header_str(headers, "X-Manta-Site");
208  let client = Client {
209    ip: request
210      .extensions()
211      .get::<ConnectInfo<SocketAddr>>()
212      .map(|ConnectInfo(peer)| peer.ip().to_string()),
213    forwarded_for: header_str(headers, "X-Forwarded-For"),
214  };
215  // The nested `/v2` router sees its URI with the prefix stripped;
216  // record the path as the client sent it.
217  let uri = request
218    .extensions()
219    .get::<OriginalUri>()
220    .map_or_else(|| request.uri().clone(), |OriginalUri(u)| u.clone());
221  let req = event::Request {
222    method: request.method().to_string(),
223    route,
224    path: uri.path().to_string(),
225    query: uri.query().map(str::to_string),
226  };
227
228  let started = Instant::now();
229  let response = next.run(request).await;
230  let status = response.status();
231  if status == StatusCode::METHOD_NOT_ALLOWED {
232    return response;
233  }
234
235  // Take rather than unwrap the Arc: a clone may outlive the response
236  // (e.g. moved into a WebSocket upgrade callback).
237  let note = note
238    .0
239    .map(|n| std::mem::take(&mut *n.lock().unwrap_or_else(|p| p.into_inner())))
240    .unwrap_or_default();
241  let action = note
242    .action
243    .map_or_else(|| format!("{} {}", req.method, req.route), str::to_string);
244  let mut event = state.audit.event(kind, action);
245  event.actor = actor;
246  event.client = client;
247  event.site = site;
248  event.request = req;
249  event.outcome = event::Outcome {
250    result: Verdict::from_status(status),
251    status: status.as_u16(),
252    duration_ms: u64::try_from(started.elapsed().as_millis())
253      .unwrap_or(u64::MAX),
254  };
255  event.dry_run = note.dry_run;
256  event.targets = note.targets;
257  event.nodes = note.nodes;
258  event.produced = note.produced;
259  event.details = note.details;
260  state.audit.emit(event);
261
262  response
263}
264
265#[cfg(test)]
266mod tests {
267  use axum::http::HeaderValue;
268  use base64::prelude::*;
269
270  use super::*;
271
272  #[test]
273  fn audits_mutations_and_consoles_only() {
274    let r = "/v2/groups/{label}";
275    assert_eq!(audited_kind(&Method::GET, r), None);
276    assert_eq!(audited_kind(&Method::HEAD, r), None);
277    assert_eq!(audited_kind(&Method::OPTIONS, r), None);
278    for m in [Method::POST, Method::PUT, Method::PATCH, Method::DELETE] {
279      assert_eq!(audited_kind(&m, r), Some(EventKind::ApiRequest), "{m}");
280    }
281    assert_eq!(
282      audited_kind(&Method::GET, "/v2/nodes/{xname}/console"),
283      Some(EventKind::ConsoleSession)
284    );
285  }
286
287  #[test]
288  fn actor_reads_claims_from_bearer_header() {
289    let claims = serde_json::json!({
290      "preferred_username": "alice", "name": "Alice A", "sub": "u-1"
291    });
292    let token =
293      format!("h.{}.s", BASE64_URL_SAFE_NO_PAD.encode(claims.to_string()));
294    let mut headers = HeaderMap::new();
295    headers.insert(
296      header::AUTHORIZATION,
297      HeaderValue::from_str(&format!("Bearer {token}")).unwrap(),
298    );
299    assert_eq!(
300      actor(&headers),
301      Actor {
302        username: Some("alice".into()),
303        name: Some("Alice A".into()),
304        subject: Some("u-1".into()),
305      }
306    );
307  }
308
309  #[test]
310  fn actor_is_empty_without_or_with_garbage_token() {
311    assert_eq!(actor(&HeaderMap::new()), Actor::default());
312    let mut headers = HeaderMap::new();
313    headers.insert(header::AUTHORIZATION, HeaderValue::from_static("Bearer x"));
314    assert_eq!(actor(&headers), Actor::default());
315  }
316
317  #[test]
318  fn inert_note_ignores_calls() {
319    let note = AuditNote::default();
320    note.action("x.y").target("group", "g").dry_run(true);
321    assert!(note.0.is_none());
322  }
323
324  #[test]
325  fn note_accumulates_targets_and_details() {
326    let note = AuditNote(Some(Arc::default()));
327    note
328      .action("group.delete")
329      .target("group", "g1")
330      .targets("node", ["x1", "x2"])
331      .targets("image", Vec::<String>::new())
332      .dry_run(false)
333      .detail("force", true)
334      .nodes(["x1000c0s0b0n0", "x1000c0s0b0n1"])
335      .produced("transition", "t-1");
336    let n = note.0.unwrap();
337    let n = n.lock().unwrap();
338    assert_eq!(n.action, Some("group.delete"));
339    assert_eq!(n.targets["group"], vec!["g1"]);
340    assert_eq!(n.targets["node"], vec!["x1", "x2"]);
341    assert!(!n.targets.contains_key("image"), "empty list left an entry");
342    assert_eq!(n.dry_run, Some(false));
343    assert_eq!(n.details["force"], serde_json::json!(true));
344    assert_eq!(n.nodes, vec!["x1000c0s0b0n0", "x1000c0s0b0n1"]);
345    assert_eq!(n.produced["transition"], "t-1");
346  }
347}
manta_server/server/audit/middleware.rs

manta_server/server/audit/
middleware.rs
