diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index a37a687..e0c1414 100644 --- a/.cargo/mutants.toml +++ b/.cargo/mutants.toml @@ -24,6 +24,10 @@ exclude_re = [ # equivalence as Zeek/Journal above. 'EveParser>::extensions.*Vec::new', + # CloudTrailParser also claims no extension (returns `&[]`) — same empty-slice + # equivalence as Zeek/Journal/Eve above. + 'CloudTrailParser>::extensions.*Vec::new', + # Parser sniff confidences are registered in strictly descending order # (MAGIC > STRONG > TEXT > FALLBACK), so the highest-confidence match is # always unique — no two firing parsers ever tie. The `>` → `==` / `>=` diff --git a/crates/ax-normalize/src/parser.rs b/crates/ax-normalize/src/parser.rs index daae1ee..a7e7f0d 100644 --- a/crates/ax-normalize/src/parser.rs +++ b/crates/ax-normalize/src/parser.rs @@ -213,6 +213,7 @@ mod tests { "parquet", "arrow", "otlp", + "cloudtrail", "eve", "journal", "ndjson", @@ -233,6 +234,7 @@ mod tests { #[cfg(not(feature = "polars"))] let expected = vec![ "otlp", + "cloudtrail", "eve", "journal", "ndjson", diff --git a/crates/ax-normalize/src/parsers/cloudtrail.rs b/crates/ax-normalize/src/parsers/cloudtrail.rs new file mode 100644 index 0000000..66685aa --- /dev/null +++ b/crates/ax-normalize/src/parsers/cloudtrail.rs @@ -0,0 +1,253 @@ +//! AWS CloudTrail parser — the audit trail for AWS API activity. +//! +//! A CloudTrail file is a single JSON document with a top-level `Records` array; +//! each record is one API call. We flatten each record to **one row**, depth-1: +//! top-level scalars become columns (`eventName`, `eventSource`, +//! `sourceIPAddress`, …), and one level into nested objects +//! (`userIdentity.userName`, `userIdentity.type`); anything deeper or array- +//! valued (`requestParameters`, `responseElements`) is kept as canonical JSON, +//! so a varied per-API payload doesn't explode the schema. +//! +//! We also synthesize **`eventEpoch`** — the RFC 3339 `eventTime` parsed to Unix +//! seconds (deterministic; no wall clock) — so `--cadence eventEpoch` and the +//! `contextual` detector (`--period 24`) can read the call series for off-hours / +//! automated-pattern anomalies, while `eventName` feeds rare-API `dist` drift. +//! +//! Detected by a `Records` array whose entries carry `eventName`; claims no +//! extension (CloudTrail is delivered as `*.json`, owned by the JSON parser — +//! pipe it on stdin for CloudTrail-aware flattening). + +use crate::infer; +use crate::parser::{Confidence, FormatParser, STRONG}; +use crate::table::TableBuilder; +use ax_core::{AxError, Column, Value}; +use chrono::DateTime; +use serde_json::Value as J; +use std::collections::BTreeMap; + +#[derive(Debug, Default, Clone)] +pub struct CloudTrailParser; + +/// Flattens one record depth-1: top-level scalars keep their name, nested +/// objects contribute `parent.child` columns, and anything deeper (or an array) +/// is lowered to its canonical JSON string by [`infer::json_to_value`]. +fn flatten_record(record: &serde_json::Map, row: &mut BTreeMap) { + for (key, value) in record { + match value { + J::Object(inner) => { + for (child, child_value) in inner { + row.insert(format!("{key}.{child}"), infer::json_to_value(child_value)); + } + } + scalar_or_array => { + row.insert(key.clone(), infer::json_to_value(scalar_or_array)); + } + } + } +} + +impl CloudTrailParser { + fn err(&self, msg: impl std::fmt::Display) -> AxError { + AxError::Parse { + format: self.id().to_string(), + message: msg.to_string(), + } + } +} + +impl FormatParser for CloudTrailParser { + fn id(&self) -> &'static str { + "cloudtrail" + } + fn extensions(&self) -> &'static [&'static str] { + &[] + } + fn sniff(&self, bytes: &[u8]) -> Option { + let value: J = serde_json::from_slice(bytes).ok()?; + let records = value.get("Records").and_then(J::as_array)?; + // A `Records` array whose entries carry `eventName` is CloudTrail (and not + // just any JSON that happens to have a "Records" key). + records + .first() + .is_some_and(|r| r.get("eventName").is_some()) + .then_some(STRONG) + } + fn parse(&self, _source: &str, bytes: &[u8]) -> Result, AxError> { + let root: J = serde_json::from_slice(bytes).map_err(|e| self.err(e))?; + let records = root + .get("Records") + .and_then(J::as_array) + .ok_or_else(|| self.err("not CloudTrail: missing 'Records' array"))?; + + let mut builder = TableBuilder::new(); + for record in records { + let obj = record + .as_object() + .ok_or_else(|| self.err("CloudTrail record is not an object"))?; + let mut row: BTreeMap = BTreeMap::new(); + flatten_record(obj, &mut row); + // Numeric event time for cadence / contextual analysis. + if let Some(time) = obj.get("eventTime").and_then(J::as_str) { + if let Ok(dt) = DateTime::parse_from_rfc3339(time) { + row.insert("eventEpoch".into(), Value::Int(dt.timestamp())); + } + } + builder.push_row(row); + } + Ok(builder.finish()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ax_core::ColType; + + const TRAIL: &str = r#"{ + "Records": [ + { + "eventVersion": "1.08", + "eventTime": "2021-01-01T00:00:00Z", + "eventSource": "s3.amazonaws.com", + "eventName": "GetObject", + "awsRegion": "us-east-1", + "sourceIPAddress": "1.2.3.4", + "readOnly": true, + "userIdentity": {"type": "IAMUser", "userName": "alice", "accountId": "111"}, + "requestParameters": {"bucketName": "logs", "key": "a/b"}, + "responseElements": null + }, + { + "eventVersion": "1.08", + "eventTime": "2021-01-01T03:30:00Z", + "eventSource": "iam.amazonaws.com", + "eventName": "CreateUser", + "awsRegion": "us-east-1", + "sourceIPAddress": "9.9.9.9", + "readOnly": false, + "userIdentity": {"type": "AssumedRole", "sessionContext": {"mfa": true}} + } + ] + }"#; + + fn parse(s: &str) -> Vec { + CloudTrailParser.parse("-", s.as_bytes()).unwrap() + } + fn col<'a>(cols: &'a [Column], name: &str) -> &'a Column { + cols.iter() + .find(|c| c.name == name) + .unwrap_or_else(|| panic!("missing column {name}")) + } + + #[test] + fn one_row_per_record_with_typed_top_level_fields() { + let cols = parse(TRAIL); + let name = col(&cols, "eventName"); + assert_eq!(name.cells.len(), 2); + assert_eq!(name.cells[0], Value::Str("GetObject".into())); + assert_eq!(name.cells[1], Value::Str("CreateUser".into())); + assert_eq!( + col(&cols, "eventSource").cells[1], + Value::Str("iam.amazonaws.com".into()) + ); + assert_eq!(col(&cols, "readOnly").ty, ColType::Bool); + assert_eq!(col(&cols, "readOnly").cells[0], Value::Bool(true)); + } + + #[test] + fn event_time_is_parsed_to_epoch_seconds() { + let cols = parse(TRAIL); + let epoch = col(&cols, "eventEpoch"); + assert_eq!(epoch.ty, ColType::Int); + assert_eq!(epoch.cells[0], Value::Int(1_609_459_200)); // 2021-01-01T00:00:00Z + assert_eq!(epoch.cells[1], Value::Int(1_609_471_800)); // +3h30m (12600s) + } + + #[test] + fn nested_objects_flatten_one_level() { + let cols = parse(TRAIL); + assert_eq!( + col(&cols, "userIdentity.userName").cells[0], + Value::Str("alice".into()) + ); + assert_eq!( + col(&cols, "userIdentity.type").cells[1], + Value::Str("AssumedRole".into()) + ); + // requestParameters flattens one level too. + assert_eq!( + col(&cols, "requestParameters.bucketName").cells[0], + Value::Str("logs".into()) + ); + // userName absent on the second record → padded null. + assert_eq!(col(&cols, "userIdentity.userName").cells[1], Value::Null); + } + + #[test] + fn deeper_nesting_is_kept_as_canonical_json() { + let cols = parse(TRAIL); + // sessionContext is an object nested inside userIdentity → stringified. + assert_eq!( + col(&cols, "userIdentity.sessionContext").cells[1], + Value::Str("{\"mfa\":true}".into()) + ); + } + + #[test] + fn flatten_record_units() { + let mut row = BTreeMap::new(); + let serde_json::Value::Object(obj) = + serde_json::json!({"a": 1, "b": {"c": "x", "d": {"e": 2}}}) + else { + unreachable!() + }; + flatten_record(&obj, &mut row); + assert_eq!(row.get("a"), Some(&Value::Int(1))); // top-level scalar + assert_eq!(row.get("b.c"), Some(&Value::Str("x".into()))); // one level in + // b.d is an object → canonical JSON string, not b.d.e. + assert_eq!(row.get("b.d"), Some(&Value::Str("{\"e\":2}".into()))); + assert_eq!(row.get("b.d.e"), None); + } + + #[test] + fn malformed_and_non_cloudtrail_error() { + assert!(matches!( + CloudTrailParser.parse("-", b"{not json"), + Err(AxError::Parse { .. }) + )); + assert!(matches!( + CloudTrailParser.parse("-", br#"{"foo": 1}"#), + Err(AxError::Parse { .. }) + )); + // A record that is not an object. + assert!(matches!( + CloudTrailParser.parse("-", br#"{"Records": [1, 2]}"#), + Err(AxError::Parse { .. }) + )); + } + + #[test] + fn sniff_keys_on_records_with_event_name() { + assert_eq!(CloudTrailParser.sniff(TRAIL.as_bytes()), Some(STRONG)); + // A Records array without eventName entries is not CloudTrail. + assert_eq!(CloudTrailParser.sniff(br#"{"Records": [{"x": 1}]}"#), None); + assert_eq!(CloudTrailParser.sniff(br#"{"Records": []}"#), None); // empty + assert_eq!(CloudTrailParser.sniff(br#"{"Records": 5}"#), None); // not an array + assert_eq!(CloudTrailParser.sniff(br#"{"foo": 1}"#), None); + assert_eq!(CloudTrailParser.sniff(b"a,b,c\n1,2,3"), None); // not JSON + } + + #[test] + fn claims_no_extension() { + assert!(CloudTrailParser.extensions().is_empty()); + } + + #[test] + fn resolves_by_content() { + let reg = crate::parser::ParserRegistry::default(); + assert_eq!( + reg.resolve("-", TRAIL.as_bytes()).unwrap().id(), + "cloudtrail" + ); + } +} diff --git a/crates/ax-normalize/src/parsers/mod.rs b/crates/ax-normalize/src/parsers/mod.rs index 194934e..aa5913e 100644 --- a/crates/ax-normalize/src/parsers/mod.rs +++ b/crates/ax-normalize/src/parsers/mod.rs @@ -6,6 +6,7 @@ use crate::parser::ParserRegistry; pub mod accesslog; pub mod cef; +pub mod cloudtrail; pub mod delimited; pub mod eve; pub mod journal; @@ -24,6 +25,7 @@ pub mod columnar; pub use accesslog::AccessLogParser; pub use cef::{CefParser, LeefParser}; +pub use cloudtrail::CloudTrailParser; pub use delimited::{CsvParser, TsvParser}; pub use eve::EveParser; pub use journal::JournalParser; @@ -49,6 +51,7 @@ pub fn default_registry() -> ParserRegistry { // OTLP before NDJSON: a compact single-object OTLP doc must win the // `resourceSpans` signature before any JSON-line heuristic sees it. r.register(Box::new(OtlpParser)); + r.register(Box::new(CloudTrailParser)); // EVE and Journal before NDJSON: both are NDJSON dialects, so their content // signatures must claim them before the generic NDJSON shape does. r.register(Box::new(EveParser));