From c7a23fc455998ded178a5b10f7b0b716f5bc611b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 22:13:14 +0000 Subject: [PATCH] =?UTF-8?q?feat(integrations):=20decision-audit=20sink=20?= =?UTF-8?q?=E2=80=94=20cpex-ocsf-audit=20records=20denials=20off=20the=20c?= =?UTF-8?q?pex#166=20seam?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Delivers the WS-A / P1 step the guinea-pig report (#456) left open. With no hooks: listed, the plugin auto-attaches as a cpex AuditHandler (same registration contract as the upstream audit-logger; hook-listed instances stay post-hook observers and never double-emit) and consumes the finalized DecisionLog at every pipeline verdict. Deny maps to action 2/Blocked with the violation surfaced at status_code/status_detail (plugin_panic stays distinguishable by code), allow-after-modification to Modified, plain allow to Allowed; ordered per-plugin steps (incl. deny_ignored, flagged flat, and aborted, distinct from error), span, entry taint, content hashes and the (epoch, stream_id, stream_seq, emission_seq) stamps ride under unmapped.cpex.* inside the hashed bytes, tamper-evident in the attestation chain. Non-CMF dispatches and zero-plugin invocations emit honest records. 11 new tests (32 total, green against seam head 386710a); on_effect lifecycle mapping tracked as follow-up. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TKAGmGAp5ECmcjFuHei5q8 --- CHANGELOG.md | 1 + integrations/cpex-ocsf-audit/README.md | 24 ++ .../cpex-ocsf-audit/SEAM-PORT-RESULTS.md | 34 +- integrations/cpex-ocsf-audit/src/emitter.rs | 389 +++++++++++++++++- integrations/cpex-ocsf-audit/src/factory.rs | 42 +- integrations/cpex-ocsf-audit/src/lib.rs | 23 +- integrations/cpex-ocsf-audit/src/ocsf.rs | 219 +++++++++- 7 files changed, 694 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 604f5bc7..c68be98e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - **Compliance cost model verified — SOC 2 readiness is ~$15–17K first-year, not $20–50K.** The compliance-readiness tracker's cost table is filled from real quotes (2026-08-07 compliance-automation platform conversation): ~$12K/yr platform covering both SOC 2 and ISO 42001, ~$3K Type I, ~$4–5K Type II, pen test and ISO 42001 bundled at no increment. Cost is no longer the gating variable — the trigger remains the first enterprise contract requiring certification. The three-horizon map's P5 profile carries the same economics. Numbers rounded and vendor-neutral in the public tree; attributed quote record internal. Docs only — no product impact. (#454) ### Added +- **Decision-sink port delivered: cpex-ocsf-audit now records denials off the cpex#166 audit seam.** The WS-A / P1 step the guinea-pig report left open is implemented: with no `hooks:` listed the plugin auto-attaches as a cpex `AuditHandler` (same registration contract as the upstream audit-logger — hook-listed instances stay post-hook observers and never double-emit) and consumes the executor's finalized `DecisionLog` at every pipeline verdict. The mapping the passive stream could never produce: Deny → `action_id` 2 / `disposition` Blocked with the violation at `status_code`/`status_detail` (a fail-closed panic arrives as `plugin_panic`, distinguishable by code), Allow-after-modification → Modified, plain Allow → Allowed; the ordered per-plugin steps (full vocabulary incl. `deny_ignored` — flagged flat so "every suppressed transform-deny" is one SIEM query — and `aborted`, distinct from error), invocation span, entry taint, content hashes, and the seam's `(epoch, stream_id, stream_seq, emission_seq)` completeness/ordering stamps ride under `unmapped.cpex.*` inside the hashed bytes, so decision facts are tamper-evident in the attestation chain. Non-CMF dispatches and zero-plugin invocations emit honest records (activity Unknown / empty steps). 11 new tests (32 total, green against seam head `386710a`); follow-up tracked: `on_effect` lifecycle events want a richer class than 6003. (#457) - **Guinea-pig port verified: cpex-ocsf-audit runs clean against the cpex#166 audit seam.** Delivers the commitment made on the cpex PR thread (2026-08-14): the OCSF audit plugin built and ran its full suite (21/21 tests, `cargo check --all-targets` clean) against `feat/audit-seam` @ `386710a` — the post-hardening head — with zero source changes, confirming the seam's "all changes are opt-in" claim for a real out-of-tree consumer. `integrations/cpex-ocsf-audit/SEAM-PORT-RESULTS.md` records scope (the `AuditHandler`/`DecisionLog` sink port with the `DenyIgnored`/`Aborted` mapping remains the tracked WS-A/P1 next step), findings for upstream (cpex-core gains `futures` + `sha2` transitive deps; the 1.96 MSRV pin fails fast), and repro steps; the committed `Cargo.lock` pins the exact resolution used. Docs/test evidence only — no product impact. (#456) - **Full agent test suite now runs in CI on `agent/**` changes.** The Ada Evals workflow ran only the static citation gate (deliberately no ADK install), so the other ~150 agent tests ran nowhere — how a test sat red on `main` for weeks after #235 and how the ADK 0.4→2.7 drift stayed invisible until #452. New `agent-tests` job installs from the exact-pinned `agent/requirements.txt` on Python 3.13 and runs the whole suite; separate job keeps the citation gate fast, the existing path filter keeps main CI untouched for non-agent changes, and it is deliberately not a required check (path-filtered required checks block merges when skipped). CI only — no product impact. (#453) diff --git a/integrations/cpex-ocsf-audit/README.md b/integrations/cpex-ocsf-audit/README.md index 268d26a8..f4043412 100644 --- a/integrations/cpex-ocsf-audit/README.md +++ b/integrations/cpex-ocsf-audit/README.md @@ -31,6 +31,30 @@ open OCSF/WS4 gaps self-documenting in the wire output. ## Wiring (APL) +**Audit-only sink mode (recommended on cpex with the audit seam, PR #166):** omit +`hooks:` entirely. The plugin then auto-attaches as a **decision-audit sink** +(`Plugin::as_audit_handler`) and fires at every pipeline verdict — **denials +included** — with the executor's `DecisionLog`: verdict → `action_id`/`disposition` +(Denied/Blocked with the violation at `status_code`/`status_detail`, Modified, +Allowed), and the ordered per-plugin steps (incl. `deny_ignored` / `aborted`), +span, entry taint, content hashes and the `(epoch, stream_id, stream_seq, +emission_seq)` stream stamps under `unmapped.cpex.*`, inside the hashed bytes. + +```yaml +plugins: + - name: ocsf-audit + kind: audit/ocsf + # no `hooks:` -> decision-audit sink mode (sees denials) + config: + destination: stderr + chain: true +``` + +**Post-hook observer mode (legacy; pre-seam cpex):** list the CMF POST hooks to +observe. This path sees allowed traffic only — it structurally cannot record a +denial — and a hook-listed instance deliberately does **not** also attach as a +sink, so one invocation never emits twice. + ```yaml routes: - tool: get_compensation diff --git a/integrations/cpex-ocsf-audit/SEAM-PORT-RESULTS.md b/integrations/cpex-ocsf-audit/SEAM-PORT-RESULTS.md index 76c2a8a5..60ded6a1 100644 --- a/integrations/cpex-ocsf-audit/SEAM-PORT-RESULTS.md +++ b/integrations/cpex-ocsf-audit/SEAM-PORT-RESULTS.md @@ -31,15 +31,31 @@ The PR's claim that "all changes are opt-in; no behavior changes without explicit configuration" holds for a real out-of-tree consumer, not just the in-tree `audit-logger`. -**Not yet exercised (next step of the port):** registering as a -decision-audit sink (`AuditHandler` / `DecisionLog`) and mapping the -finalized decision vocabulary — `Allowed` / `ModifiedPayload` / -`ModifiedExtensions` / `DenyIgnored` / `Aborted` / `Error`, terminal -verdicts, `plugin_panic` coding, and the `(epoch, stream_seq, emission_seq)` -stamps — into OCSF records (deny/modify records, action_id 2/4). That is the -WS-A / P1 work item already tracked in `src/lib.rs`; the vocabulary review -on the PR thread (2026-08-18) confirmed the contract carries everything the -mapping needs. +**Not yet exercised at the time of the first run (delivered later the same +day — see below):** registering as a decision-audit sink (`AuditHandler` / +`DecisionLog`) and mapping the finalized decision vocabulary — `Allowed` / +`ModifiedPayload` / `ModifiedExtensions` / `DenyIgnored` / `Aborted` / +`Error`, terminal verdicts, `plugin_panic` coding, and the +`(epoch, stream_seq, emission_seq)` stamps — into OCSF records (deny/modify +records, action_id 2/4). + +## Addendum (2026-08-18, later): the decision-sink port is implemented + +The WS-A / P1 step above landed the same day: with no `hooks:` listed the +plugin now auto-attaches as an `AuditHandler` (mirroring the upstream +audit-logger's registration contract) and consumes the full `DecisionLog` +at every verdict. Verdict → `security_control` (Denied/Blocked with the +violation surfaced at `status_code`/`status_detail` — so `plugin_panic` +arrives distinguishable by code; Allow-after-modification → Modified; +plain Allow → Allowed); steps, span, entry taint, content hashes and the +stream stamps ride under `unmapped.cpex.*` **inside the hashed bytes**, so +the decision facts are tamper-evident in the attestation chain. 11 new +tests (32 total, all green against the same seam head `386710a`) cover the +registration contract, every verdict mapping, the `deny_ignored` and +`aborted` renderings, zero-step and non-CMF dispatches, and that the +stream stamps are bound into the fingerprint. Remaining follow-up: +`AuditHandler::on_effect` (effect-lifecycle events want a richer OCSF +class than 6003 — e.g. Authentication for a token mint). ## Observations for upstream (the "anything that doesn't match intent" list) diff --git a/integrations/cpex-ocsf-audit/src/emitter.rs b/integrations/cpex-ocsf-audit/src/emitter.rs index 2a3d8cd5..c1b781a8 100644 --- a/integrations/cpex-ocsf-audit/src/emitter.rs +++ b/integrations/cpex-ocsf-audit/src/emitter.rs @@ -161,7 +161,31 @@ impl OcsfAuditEmitter { /// the stderr/tracing emit path. pub fn build(&self, payload: &MessagePayload, ext: &Extensions, now_rfc3339: &str) -> Value { let event = ocsf::build_ai_operation(payload, ext, &self.typed, now_rfc3339); + self.wrap_in_chain(event) + } + /// Build a decision-audit event: the same OCSF shape as `build`, with + /// the pipeline's ruling overlaid (verdict → action/disposition/status, + /// per-plugin steps, span, taint, content hashes and stream stamps + /// under `unmapped.cpex.*`) — then chained like any other record. + /// `payload` is `None` for a non-CMF dispatch (delegation, identity): + /// the record still emits, from the extensions alone. + pub fn build_decision( + &self, + payload: Option<&MessagePayload>, + ext: &Extensions, + decisions: &cpex_core::decision::DecisionLog, + now_rfc3339: &str, + ) -> Value { + let mut event = ocsf::build_event(payload, ext, &self.typed, now_rfc3339); + ocsf::apply_decision(&mut event, payload, decisions); + self.wrap_in_chain(event) + } + + /// Wrap `event` in the attestation chain (no-op when `chain: false`). + /// Decision records and post-hook observations share one chain: the + /// chain orders *emissions of this emitter*, whichever path built them. + fn wrap_in_chain(&self, event: Value) -> Value { if !self.typed.chain { return event; } @@ -280,6 +304,52 @@ impl Plugin for OcsfAuditEmitter { fn config(&self) -> &PluginConfig { &self.cfg } + + /// Auto-attach as a decision-audit sink when run in audit-only mode + /// (no `hooks:` listed) — the manager then invokes the `AuditHandler` + /// impl below at every pipeline verdict, denials included. If the + /// operator listed hooks, this runs as a CMF post-hook observer + /// instead and does not also auto-attach, so records aren't emitted + /// twice for one invocation. (Same contract as the upstream + /// audit-logger builtin.) + fn as_audit_handler( + self: std::sync::Arc, + ) -> Option> { + if self.cfg.hooks.is_empty() { + Some(self) + } else { + None + } + } +} + +/// Decision-audit consumer — the first-class path off the PR #166 audit +/// seam. Fires at the verdict of every pipeline run with the finalized +/// [`DecisionLog`](cpex_core::decision::DecisionLog); this is what makes +/// denials, suppressed transform-denies, panics and modifications visible +/// to the OCSF stream (a post-hook observer only ever saw allowed +/// traffic). Awaited on the request path by contract — `handle` stays +/// serialize-and-emit cheap. +#[async_trait] +impl cpex_core::audit::AuditHandler for OcsfAuditEmitter { + async fn handle( + &self, + payload: &dyn cpex_core::hooks::payload::PluginPayload, + ext: &Extensions, + decisions: &cpex_core::decision::DecisionLog, + ) { + // Downcast to the CMF payload when this dispatch carried one; a + // non-CMF dispatch (delegation, identity) records without the + // message-derived fields. + let msg = payload.as_any().downcast_ref::(); + let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + let event = self.build_decision(msg, ext, decisions, &now); + self.emit(&event); + } + + fn name(&self) -> &str { + &self.cfg.name + } } impl HookHandler for OcsfAuditEmitter { @@ -619,11 +689,328 @@ mod tests { async fn handler_is_observation_only() { let e = OcsfAuditEmitter::new(cfg(json!({}))).unwrap(); let mut ctx = PluginContext::default(); - let r = e.handle(&tool_payload(), &subject_ext(), &mut ctx).await; + let r = HookHandler::::handle(&e, &tool_payload(), &subject_ext(), &mut ctx).await; assert!(r.continue_processing); assert!(r.violation.is_none()); } + // --- decision-audit sink (PR #166 seam; WS-A / P1) ------------------- + + use cpex_core::decision::{DecisionLog, PluginAction, Span, Verdict}; + use cpex_core::error::PluginViolation; + + /// Sink-mode config: no `hooks:` — the factory registers no post-hook + /// handlers and the plugin auto-attaches as a decision-audit sink. + fn sink_cfg(extra: serde_json::Value) -> PluginConfig { + PluginConfig { + hooks: vec![], + ..cfg(extra) + } + } + + fn finalized(steps: Vec<(&str, PluginMode, PluginAction)>, verdict: Verdict) -> DecisionLog { + let mut log = DecisionLog::new(); + for (name, mode, action) in steps { + log.record(name, mode, action); + } + log.finalize(verdict); + log + } + + /// Registration contract: audit-only mode (no hooks) attaches as a + /// sink; a hook-listed observer does NOT also attach, so one + /// invocation never emits twice. + #[test] + fn audit_handler_attaches_only_in_sink_mode() { + use cpex_core::plugin::Plugin; + let sink = Arc::new(OcsfAuditEmitter::new(sink_cfg(json!({}))).unwrap()); + assert!(sink.as_audit_handler().is_some(), "no hooks -> sink"); + + let observer = Arc::new(OcsfAuditEmitter::new(cfg(json!({}))).unwrap()); + assert!( + observer.as_audit_handler().is_none(), + "hooks listed -> post-hook observer only, no double emission" + ); + } + + #[test] + fn allow_verdict_maps_to_allowed() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let log = finalized( + vec![("cedar-pdp", PluginMode::Sequential, PluginAction::Allowed)], + Verdict::Allow, + ); + let ev = e.build_decision( + Some(&tool_payload()), + &subject_ext(), + &log, + "2026-08-18T12:00:00.000Z", + ); + + assert_eq!(ev["action_id"], 1); + assert_eq!(ev["action"], "Allowed"); + assert_eq!(ev["disposition_id"], 1); + assert_eq!(ev["disposition"], "Allowed"); + // The operation classification is untouched by the ruling. + assert_eq!(ev["class_uid"], 6003); + assert_eq!(ev["activity_name"], "Invoke Tool"); + let steps = ev["unmapped"]["cpex.decision"]["steps"].as_array().unwrap(); + assert_eq!(steps.len(), 1); + assert_eq!(steps[0]["plugin"], "cedar-pdp"); + assert_eq!(steps[0]["phase"], "sequential"); + assert_eq!(steps[0]["action"], "allowed"); + assert_eq!(ev["unmapped"]["cpex.decision"]["verdict"], "allow"); + } + + /// The record a post-hook observer could never produce: a denial — + /// including the fail-closed panic contract from the hardening round + /// (`eda9821`): the violation code (`plugin_panic`) must survive to + /// `status_code`, distinguishable from an ordinary `plugin_error`. + #[test] + fn deny_verdict_maps_to_denied_with_violation_status() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let mut violation = PluginViolation::new( + "plugin_panic", + "Plugin 'minter' failed: task panicked: simulated", + ); + violation.plugin_name = Some("minter".into()); + let log = finalized( + vec![( + "minter", + PluginMode::Sequential, + PluginAction::Error("task panicked: simulated".into()), + )], + Verdict::Deny(violation), + ); + let ev = e.build_decision( + Some(&tool_payload()), + &subject_ext(), + &log, + "2026-08-18T12:00:00.000Z", + ); + + assert_eq!(ev["action_id"], 2); + assert_eq!(ev["action"], "Denied"); + assert_eq!(ev["disposition_id"], 2); + assert_eq!(ev["disposition"], "Blocked"); + assert_eq!(ev["status_id"], 2); + assert_eq!(ev["status_code"], "plugin_panic"); + assert!(ev["status_detail"] + .as_str() + .unwrap() + .contains("task panicked")); + let d = &ev["unmapped"]["cpex.decision"]; + assert_eq!(d["verdict"]["deny"]["code"], "plugin_panic"); + assert_eq!(d["steps"][0]["action"], "error"); + assert!(d["steps"][0]["error"] + .as_str() + .unwrap() + .contains("panicked")); + } + + #[test] + fn modified_allow_maps_to_modified() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let log = finalized( + vec![ + ( + "pii-scrubber", + PluginMode::Transform, + PluginAction::ModifiedPayload, + ), + ("cedar-pdp", PluginMode::Sequential, PluginAction::Allowed), + ], + Verdict::Allow, + ); + let ev = e.build_decision( + Some(&tool_payload()), + &subject_ext(), + &log, + "2026-08-18T12:00:00.000Z", + ); + + assert_eq!(ev["action_id"], 4); + assert_eq!(ev["action"], "Modified"); + // The request still proceeded. + assert_eq!(ev["disposition_id"], 1); + assert_eq!( + ev["unmapped"]["cpex.decision"]["steps"][0]["action"], + "modified_payload" + ); + } + + /// Seam contract on `PluginAction::DenyIgnored`: a suppressed + /// Transform-phase block must never read as a plain allow. The event + /// stays Allowed (enforcement DID allow it) but the step carries the + /// plugin's actual decision and the block is flagged flat for SIEM + /// queries. + #[test] + fn deny_ignored_never_reads_as_plain_allow() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let log = finalized( + vec![( + "strict-transform", + PluginMode::Transform, + PluginAction::DenyIgnored, + )], + Verdict::Allow, + ); + let ev = e.build_decision( + Some(&tool_payload()), + &subject_ext(), + &log, + "2026-08-18T12:00:00.000Z", + ); + + assert_eq!(ev["action_id"], 1, "enforcement outcome was allow"); + let d = &ev["unmapped"]["cpex.decision"]; + assert_eq!(d["steps"][0]["action"], "deny_ignored"); + assert_eq!(d["deny_ignored"], true, "flat flag for SIEM queries"); + } + + /// `Aborted` (a concurrent sibling short-circuited the phase) is an + /// intentional cancellation — it must not render as an error. + #[test] + fn aborted_step_is_not_an_error() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let log = finalized( + vec![ + ("scanner-b", PluginMode::Concurrent, PluginAction::Aborted), + ("scanner-a", PluginMode::Concurrent, PluginAction::Denied), + ], + Verdict::Deny(PluginViolation::new("policy_deny", "blocked")), + ); + let ev = e.build_decision( + Some(&tool_payload()), + &subject_ext(), + &log, + "2026-08-18T12:00:00.000Z", + ); + + let steps = ev["unmapped"]["cpex.decision"]["steps"].as_array().unwrap(); + assert_eq!(steps[0]["action"], "aborted"); + assert!(steps[0].get("error").is_none()); + assert_eq!(steps[1]["action"], "denied"); + } + + /// Zero-plugin invocations emit one allow record on the seam (dense + /// stream); our sink renders it with an empty steps array, not a gap. + #[test] + fn zero_step_invocation_emits_allow_record() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let log = finalized(vec![], Verdict::Allow); + let ev = e.build_decision( + Some(&tool_payload()), + &subject_ext(), + &log, + "2026-08-18T12:00:00.000Z", + ); + + assert_eq!(ev["action_id"], 1); + assert_eq!( + ev["unmapped"]["cpex.decision"]["steps"], + json!([]), + "explicit empty steps, not absence" + ); + } + + /// Audit sinks fire for every hook family; a non-CMF dispatch carries + /// no MessagePayload and must still produce a record. + #[test] + fn non_cmf_dispatch_still_emits() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let log = finalized( + vec![("cedar-pdp", PluginMode::Sequential, PluginAction::Denied)], + Verdict::Deny(PluginViolation::new("missing_permission", "no")), + ); + let ev = e.build_decision(None, &subject_ext(), &log, "2026-08-18T12:00:00.000Z"); + + assert_eq!(ev["class_uid"], 6003); + assert_eq!(ev["activity_id"], 0, "no payload -> honest Unknown"); + assert_eq!(ev["action_id"], 2); + assert_eq!(ev["status_code"], "missing_permission"); + assert!(ev.get("tool").is_none(), "no payload -> no tool coords"); + // Extension-derived context still populates. + assert_eq!(ev["actor"]["user"]["uid"], "alice@corp.com"); + } + + /// Span, entry taint, content provenance and the stream stamps land + /// under unmapped.cpex.* — the seam's counters verbatim. + #[test] + fn provenance_and_stream_stamps_land_in_unmapped() { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let mut log = finalized( + vec![("cedar-pdp", PluginMode::Sequential, PluginAction::Allowed)], + Verdict::Allow, + ); + log.set_span(Span::for_request(Some("trace-1"), Some("parent-1"))); + log.set_input_labels(vec!["PII".into()]); + log.set_input_hash(Some("in-hash".into())); + log.set_stream(1_755_000_000_000_000_000, "decision".into(), 7, 42); + + let payload = tool_payload(); + let ev = e.build_decision(Some(&payload), &subject_ext(), &log, "2026-08-18T12:00:00.000Z"); + + let un = &ev["unmapped"]; + assert_eq!(un["cpex.span"]["trace_id"], "trace-1"); + assert_eq!(un["cpex.span"]["parent_span_id"], "parent-1"); + assert_eq!(un["cpex.taint.input_labels"], json!(["PII"])); + assert_eq!(un["cpex.content"]["input_hash"], "in-hash"); + // Output hash present iff the payload yields audit bytes; either + // way the key exists so the claim is explicit. + assert!(un["cpex.content"] + .as_object() + .unwrap() + .contains_key("output_hash")); + assert_eq!(un["cpex.stream"]["epoch"], 1_755_000_000_000_000_000u64); + assert_eq!(un["cpex.stream"]["stream_id"], "decision"); + assert_eq!(un["cpex.stream"]["stream_seq"], 7); + assert_eq!(un["cpex.stream"]["emission_seq"], 42); + } + + /// The decision facts sit INSIDE the hashed bytes: two otherwise + /// identical genesis records with different stream stamps must + /// fingerprint differently — renumbering the stream post-hoc breaks + /// the chain. + #[test] + fn decision_facts_are_bound_into_the_fingerprint() { + let build = |stream_seq: u64| { + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": true }))).unwrap(); + let mut log = finalized( + vec![("cedar-pdp", PluginMode::Sequential, PluginAction::Allowed)], + Verdict::Allow, + ); + log.set_stream(1, "decision".into(), stream_seq, stream_seq); + e.build_decision( + Some(&tool_payload()), + &subject_ext(), + &log, + "2026-08-18T12:00:00.000Z", + ) + }; + let a = build(7); + let b = build(8); + assert_ne!( + a["attestation_list"][0]["fingerprint"]["value"], + b["attestation_list"][0]["fingerprint"]["value"] + ); + } + + /// End-to-end through the trait object, as the executor calls it: the + /// dyn payload downcasts to CMF and the handler completes. + #[tokio::test] + async fn audit_handler_handles_dyn_payload() { + use cpex_core::audit::AuditHandler; + let e = OcsfAuditEmitter::new(sink_cfg(json!({ "chain": false }))).unwrap(); + let payload = tool_payload(); + let log = finalized( + vec![("cedar-pdp", PluginMode::Sequential, PluginAction::Allowed)], + Verdict::Allow, + ); + AuditHandler::handle(&e, &payload, &subject_ext(), &log).await; + assert_eq!(AuditHandler::name(&e), "ocsf-audit"); + } + // --- gap-branch coverage -------------------------------------------- // The happy-path test above only exercises a tool call + subject. These // build a fully-populated Extensions set and assert every gap field diff --git a/integrations/cpex-ocsf-audit/src/factory.rs b/integrations/cpex-ocsf-audit/src/factory.rs index e9ab6fc7..16ddbe53 100644 --- a/integrations/cpex-ocsf-audit/src/factory.rs +++ b/integrations/cpex-ocsf-audit/src/factory.rs @@ -28,20 +28,38 @@ impl PluginFactory for OcsfAuditFactory { fn create(&self, config: &PluginConfig) -> Result> { let emitter = Arc::new(OcsfAuditEmitter::new(config.clone())?); + // Make the inferred mode explicit in the startup log (mirrors the + // upstream audit-logger factory): audit-only sink is the + // recommended mode, but an operator who *meant* to list hooks and + // lost them to a YAML slip should be able to catch it here. if config.hooks.is_empty() { - return Err(Box::new(PluginError::Config { - message: format!( - "plugin '{}' (cpex-plugin-ocsf-audit): `hooks:` must list at least \ - one CMF hook to emit on. For audit, prefer the POST hooks: \ - cmf.tool_post_invoke, cmf.llm_output, cmf.resource_post_fetch, \ - cmf.prompt_post_invoke. (NOT cmf.prompt_post_fetch — that name \ - exists in hooks/types.rs but the Rust CMF/APL runtime dispatches \ - the cmf/constants.rs name, cmf.prompt_post_invoke; a handler on \ - the _fetch name silently never fires.)", - config.name - ), - })); + // Audit-only sink mode: no CMF post-hook handlers; the plugin + // auto-attaches as a decision-audit sink instead (see + // `Plugin::as_audit_handler` in emitter.rs) and fires at every + // pipeline verdict — denials included. This is the recommended + // mode on cpex with the audit seam (PR #166). + tracing::info!( + plugin = %config.name, + "ocsf-audit '{}' running in audit-only sink mode (no `hooks:` listed) — \ + auto-attaches to the executor verdict path; if you meant to observe \ + specific CMF hooks, list them under `hooks:`", + config.name, + ); + return Ok(PluginInstance { + plugin: emitter, + handlers: Vec::new(), + }); } + tracing::info!( + plugin = %config.name, + hooks = ?config.hooks, + "ocsf-audit '{}' running as a CMF post-hook observer on {:?} — this path \ + sees allowed traffic only; audit-only sink mode (no `hooks:`) also \ + records denials. (Avoid cmf.prompt_post_fetch — the runtime dispatches \ + cmf.prompt_post_invoke; a handler on the _fetch name silently never fires.)", + config.name, + config.hooks, + ); let handlers: Vec<_> = config .hooks diff --git a/integrations/cpex-ocsf-audit/src/lib.rs b/integrations/cpex-ocsf-audit/src/lib.rs index d7d02686..3b946ec3 100644 --- a/integrations/cpex-ocsf-audit/src/lib.rs +++ b/integrations/cpex-ocsf-audit/src/lib.rs @@ -37,8 +37,8 @@ // record_integrity when chained) and the passive stream carries // action_id 3 (Observed) / disposition_id 17 (Logged); and the hash // commits to the record's chain position — predecessor binding, not a -// back-pointer. Remaining by design: deny/modify records (action_id -// 2/4) wait on the cpex-core decision event (WS-A / P1). +// back-pointer. (The deny/modify records this note deferred to WS-A / +// P1 landed 2026-08-18 — see the decision-sink revision below.) // // Revision 2026-07-31 — MERGED #1661 SHAPE. PR #1661 merged upstream // 2026-07-17 (`2a244bc9`), and the emitted attestation now matches it: @@ -64,6 +64,25 @@ // `sign::dsse_pae`. Key custody (HSM/KMS, rotation epochs, JWKS // publication) is deliberately out of plugin scope — it belongs to the // operating authority. +// +// Revision 2026-08-18 — DECISION-AUDIT SINK (WS-A / P1 delivered). The +// plugin now consumes the first-class audit seam from cpex PR #166 +// (verified against feat/audit-seam @ 386710a, post-hardening): with no +// `hooks:` listed it auto-attaches as an AuditHandler +// (Plugin::as_audit_handler) and fires at every pipeline verdict — +// denials included, which the post-hook path structurally never saw. +// The DecisionLog maps as: verdict -> security_control (Deny -> action 2 +// Denied / disposition 2 Blocked, violation at status_code/status_detail +// so `plugin_panic` survives by code; Allow-after-modification -> +// action 4 Modified; plain Allow -> 1/1), and the ordered per-plugin +// steps (full vocabulary incl. deny_ignored / aborted), span, entry +// taint, content hashes and the (epoch, stream_id, stream_seq, +// emission_seq) stamps ride under unmapped.cpex.* — inside the hashed +// bytes, so the decision facts are tamper-evident in the attestation +// chain. Listing hooks still runs the legacy post-hook observer (and +// then deliberately does NOT also attach, so one invocation never emits +// twice). Effect-lifecycle events (AuditHandler::on_effect) are the +// next tracked step — a token mint wants a richer OCSF class than 6003. pub mod config; pub mod emitter; diff --git a/integrations/cpex-ocsf-audit/src/ocsf.rs b/integrations/cpex-ocsf-audit/src/ocsf.rs index cec66e99..f5c84284 100644 --- a/integrations/cpex-ocsf-audit/src/ocsf.rs +++ b/integrations/cpex-ocsf-audit/src/ocsf.rs @@ -29,7 +29,8 @@ use serde_json::{json, Map, Value}; use cpex_core::cmf::{ContentPart, MessagePayload}; -use cpex_core::hooks::payload::Extensions; +use cpex_core::decision::{DecisionLog, PluginAction, Verdict}; +use cpex_core::hooks::payload::{Extensions, PluginPayload}; use crate::config::OcsfAuditConfig; @@ -150,7 +151,24 @@ pub fn build_ai_operation( cfg: &OcsfAuditConfig, now_rfc3339: &str, ) -> Value { - let activity = activity_of(payload, ext); + build_event(Some(payload), ext, cfg, now_rfc3339) +} + +/// Payload-optional form of [`build_ai_operation`]. The decision-audit +/// sink fires for every hook family, and a non-CMF dispatch (delegation, +/// identity) carries no `MessagePayload` — the event is then built from +/// the extensions alone, with `activity_id` 0 (Unknown) and no +/// tool/status coordinates. The extension-derived blocks (actor, +/// ai_agent, ai_model, delegation, gap fields) are identical either way. +pub fn build_event( + payload: Option<&MessagePayload>, + ext: &Extensions, + cfg: &OcsfAuditConfig, + now_rfc3339: &str, +) -> Value { + let activity = payload + .map(|p| activity_of(p, ext)) + .unwrap_or(Activity::Unknown); let mut ev = Map::new(); @@ -166,11 +184,12 @@ pub fn build_ai_operation( ev.insert("severity_id".into(), json!(SEVERITY_INFORMATIONAL)); ev.insert("time".into(), json!(now_rfc3339)); - // security_control profile: this passive post-hook stream is - // action_id 3 (Observed) / disposition_id 17 (Logged). The deny and - // modify mappings (action_id 2 / 4) arrive with the cpex-core - // decision event (WS-A / P1) — the plugin structurally cannot see a - // denial from a post hook. + // security_control profile defaults: a passive post-hook observation + // is action_id 3 (Observed) / disposition_id 17 (Logged). When this + // event is built by the decision-audit sink, `apply_decision` + // overwrites these with the pipeline's actual ruling (Denied / + // Modified / Allowed) — a post-hook observer structurally cannot see + // a denial, which is exactly what the sink path fixes. ev.insert("action_id".into(), json!(3)); ev.insert("action".into(), json!("Observed")); ev.insert("disposition_id".into(), json!(17)); @@ -205,7 +224,7 @@ pub fn build_ai_operation( ev.insert("metadata".into(), metadata); // status (field map: ToolResult.is_error -> status) - if let Some(is_err) = first_tool_error(payload) { + if let Some(is_err) = payload.and_then(first_tool_error) { ev.insert("status_id".into(), json!(if is_err { 2 } else { 1 })); // 1=Success 2=Failure } @@ -296,11 +315,13 @@ pub fn build_ai_operation( } // --- tool/prompt/resource coordinates from content ---------------- - attach_capability_coords(&mut ev, payload); + if let Some(p) = payload { + attach_capability_coords(&mut ev, p); + } // --- the five gaps -> unmapped (field map §5) --------------------- if cfg.include_gap_fields { - let unmapped = build_unmapped_gaps(payload, ext); + let unmapped = build_unmapped_gaps(ext); if let Value::Object(m) = &unmapped { if !m.is_empty() { ev.insert("unmapped".into(), unmapped); @@ -313,7 +334,7 @@ pub fn build_ai_operation( /// Gap fields with no native OCSF home yet. Emitting them under /// `unmapped` keeps the evidence complete and documents the gaps. -fn build_unmapped_gaps(payload: &MessagePayload, ext: &Extensions) -> Value { +fn build_unmapped_gaps(ext: &Extensions) -> Value { let mut g = Map::new(); // gap 3: completion.stop_reason @@ -361,12 +382,182 @@ fn build_unmapped_gaps(payload: &MessagePayload, ext: &Extensions) -> Value { g.insert("cmf.workload_identity".into(), wl); } - // multimodal content kinds present (lightweight provenance of shape) - let _ = payload; - Value::Object(g) } +// --------------------------------------------------------------------- +// Decision overlay — the WS-A / P1 mapping. Applies the pipeline's +// ruling (cpex-core DecisionLog, from the PR #166 audit seam) onto an +// event built by `build_event`, replacing the passive Observed/Logged +// defaults with what enforcement actually did. +// --------------------------------------------------------------------- + +/// The stable, queryable rendering of one [`PluginAction`]. Deliberately +/// a fixed snake_case vocabulary (not `Debug` formatting) so SIEM +/// queries survive upstream enum renames; `error` carries its message +/// beside the action, not inside it. +fn action_str(a: &PluginAction) -> &'static str { + match a { + PluginAction::Allowed => "allowed", + PluginAction::Denied => "denied", + PluginAction::ModifiedPayload => "modified_payload", + PluginAction::ModifiedExtensions => "modified_extensions", + // Never rendered as an allow — the step reflects the plugin's + // actual decision (a suppressed Transform-phase block), per the + // seam's contract on `PluginAction::DenyIgnored`. + PluginAction::DenyIgnored => "deny_ignored", + // Intentional cancellation (a concurrent sibling short-circuited + // the phase) — distinct from `error` so it doesn't read as a crash. + PluginAction::Aborted => "aborted", + PluginAction::Error(_) => "error", + } +} + +/// Overlay one finalized [`DecisionLog`] onto an event from +/// [`build_event`], turning a passive observation into a decision record: +/// +/// * **Verdict → security_control.** Deny → `action_id` 2 (Denied) / +/// `disposition_id` 2 (Blocked), with the violation surfaced at +/// `status_code` / `status_detail` (`status_id` 2) — so a fail-closed +/// panic arrives as `status_code: "plugin_panic"`, distinguishable +/// from an ordinary `plugin_error` by code. Allow after a payload or +/// extension modification → `action_id` 4 (Modified) / +/// `disposition_id` 1 (Allowed). Plain allow → 1 / 1. `activity_*` / +/// `type_uid` are untouched: they describe the operation observed, the +/// action describes what the control did about it. +/// * **Everything else → `unmapped.cpex.*`**, inside the hashed bytes +/// when chaining is on, so the decision facts are tamper-evident: +/// the ordered per-plugin steps (full vocabulary incl. `deny_ignored` +/// and `aborted`), the invocation span, entry-taint labels, content +/// provenance (input/output hashes), and the audit-stream stamps +/// (`epoch` / `stream_id` / `stream_seq` / `emission_seq` — the +/// completeness and ordering claims from the seam). +pub fn apply_decision(ev: &mut Value, payload: Option<&MessagePayload>, decisions: &DecisionLog) { + let Some(map) = ev.as_object_mut() else { return }; + + let modified = decisions.steps().iter().any(|s| { + matches!( + s.action, + PluginAction::ModifiedPayload | PluginAction::ModifiedExtensions + ) + }); + + let (action_id, action, disposition_id, disposition, verdict_json) = match decisions.verdict() { + Some(Verdict::Deny(v)) => { + // The violation is the forensic core of a deny — surface it + // on the base-event status fields where OCSF consumers + // already look, not only inside the unmapped block. + map.insert("status_id".into(), json!(2)); // Failure + map.insert("status_code".into(), json!(v.code)); + map.insert("status_detail".into(), json!(v.reason)); + ( + 2, + "Denied", + 2, + "Blocked", + json!({ "deny": { "code": v.code, "reason": v.reason } }), + ) + } + Some(Verdict::Allow) if modified => (4, "Modified", 1, "Allowed", json!("allow")), + Some(Verdict::Allow) => (1, "Allowed", 1, "Allowed", json!("allow")), + // The seam finalizes before invoking sinks; `None` would mean a + // contract break upstream. Keep the Observed/Logged defaults and + // say so rather than claim a ruling that never happened. + None => (3, "Observed", 17, "Logged", json!("pending")), + }; + map.insert("action_id".into(), json!(action_id)); + map.insert("action".into(), json!(action)); + map.insert("disposition_id".into(), json!(disposition_id)); + map.insert("disposition".into(), json!(disposition)); + + // --- unmapped.cpex.* — merged into any existing gap fields -------- + let un = map + .entry("unmapped") + .or_insert_with(|| Value::Object(Map::new())); + let Some(un) = un.as_object_mut() else { return }; + + let steps: Vec = decisions + .steps() + .iter() + .map(|s| { + let mut step = json!({ + "plugin": s.plugin_name, + "phase": s.phase.to_string(), + "action": action_str(&s.action), + }); + if let PluginAction::Error(e) = &s.action { + step["error"] = json!(e); + } + step + }) + .collect(); + let mut decision = json!({ "verdict": verdict_json, "steps": steps }); + // Flagged at the top level of the block (not only discoverable by + // scanning the steps array) so "every suppressed deny" is a flat + // SIEM query — the seam's contract is that this must never read as + // a plain allow. + if decisions + .steps() + .iter() + .any(|s| s.action == PluginAction::DenyIgnored) + { + decision["deny_ignored"] = json!(true); + } + un.insert("cpex.decision".into(), decision); + + // The invocation's node identity in the decision graph (W3C ids; + // child-span model — parent is the causal edge). + if let Some(span) = decisions.span() { + un.insert( + "cpex.span".into(), + json!({ + "trace_id": span.trace_id, + "span_id": span.span_id, + "parent_span_id": span.parent_span_id, + }), + ); + } + + // Entry-side taint. The final labels already ride at + // `cmf.security.labels` (gap 4); the difference is what the + // pipeline added. + if !decisions.input_labels().is_empty() { + un.insert( + "cpex.taint.input_labels".into(), + json!(decisions.input_labels()), + ); + } + + // Content provenance — gated on the executor having captured an + // input hash (i.e. capture_content_provenance on). Digests only. + if let Some(input_hash) = decisions.input_hash() { + let output_hash = payload + .and_then(|p| p.audit_bytes()) + .map(|b| cpex_core::hooks::payload::content_hash(&b)); + un.insert( + "cpex.content".into(), + json!({ "input_hash": input_hash, "output_hash": output_hash }), + ); + } + + // Audit-stream identity + counters, verbatim from the seam: + // `stream_seq` is the completeness claim (dense within + // (epoch, stream_id)); `emission_seq` is ordering-only (sparse for a + // single-stream consumer by design). Inside the hashed bytes, so a + // post-hoc renumbering breaks the fingerprint chain. + if decisions.stream_seq().is_some() { + un.insert( + "cpex.stream".into(), + json!({ + "epoch": decisions.epoch(), + "stream_id": decisions.stream_id(), + "stream_seq": decisions.stream_seq(), + "emission_seq": decisions.emission_seq(), + }), + ); + } +} + // --------------------------------------------------------------------- // Helpers — small, content-shape-dependent extractors. CMF accessor and // variant shapes confirmed against cpex@feat/hil_apl ad666ba (2026-07-06).