diff --git a/CHANGELOG.md b/CHANGELOG.md index bee9ddcb..80125422 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/). ### Added - **Out-of-process host for existing Python CPEX plugins.** A new `cpex-hosts-python` crate registers `kind: isolated_venv`, running an unmodified Python CPEX plugin in its own cached virtualenv as a subprocess instead of in-process through the PyO3 bindings. Each plugin gets a venv keyed by a SHA-256 fingerprint of its requirements + manifest (rebuilt when either changes, `rmtree`d rather than upgraded in place so a removed dependency actually disappears), and the host drives the Python framework's `worker.py` over a newline-delimited JSON stdio protocol. Hook payloads, `context`, and the capability-filtered `Extensions` view cross as JSON; returns come back as a serialized `PluginResult`, with `modified_extensions` merged through the executor's existing copy-on-write tier validation — the host implements no tier logic of its own. Failure modes the executor cannot otherwise distinguish (venv build failure, worker death mid-flight, a task over `max_content_size`, per-invocation timeout) map to distinct `PluginError`s carrying a stable `code` and structured `details`, so the executor's configured `on_error` policy applies unchanged. Pure Rust plus a subprocess — no libpython link, so the crate is in `default-members` and a plain `cargo build` does not require a Python dev install. The wire contract is pinned in `docs/specs/extensions-wire-contract.md`; CMF §3 remains normative for the extension slots themselves. (#149) +- **First-class decision and effect auditing.** CPEX can now audit its own enforcement — every allow, deny, and modify — instead of only the allowed post-hook traffic an observation plugin happened to see. A new `AuditHook` family, auto-attached by the `PluginManager`, fires at the executor's verdict return points (not a pipeline phase), so a blocked call, a scope narrowing, and a clean allow all produce a record. Each carries a `DecisionLog` — executor-owned and handed only to audit sinks, never placed on `PluginContext` — with the ordered plugin steps, the terminal verdict, the invocation's W3C trace span (`trace_id` / `span_id` / `parent_span_id`, child-span model: a fresh span whose parent is the request's span, for causal-DAG reconstruction), the taint labels the request arrived with, and, opt-in, a content hash of the payload at entry. Irreversible external effects (a token mint, an approval grant) are audited as their own events through a capability-gated, write-ahead protocol: a plugin holding `emit_effect` calls `ext.begin_effect` to durably record intent *before* the act (fail-closed — no durable record, no act) and `ext.complete_effect` to record the outcome (`confirmed` / `rejected` / `unknown`). A durable `FileEffectLog` write-ahead log (append + `fsync`, serialized against concurrent writers, self-compacting at a configurable threshold) makes the intent crash-safe; startup recovery (`PluginManager::recover_effects`) compacts completed effects and reconciles crash-orphaned ones against the issuing participant through an `EffectReconciler` seam (the default logs and leaves them `unknown`). `Extensions::perform_effect` brackets the two-phase protocol so a caller cannot skip, reorder, or forget it. Opt-in throughout: no effect WAL and no content hashing unless `plugin_settings.effect_log_path` / `plugin_settings.capture_content_provenance` are set. (#XXX) +- **The OAuth delegator emits write-ahead audit for the tokens it mints.** `cpex-plugin-delegator-oauth` now brackets both mint legs — the workload `client_assertion` base-token mint and the RFC 8693 exchange — with `begin_effect` / `complete_effect`, mapping a successful exchange to `confirmed`, a definitive IdP rejection to `rejected`, and a timeout or unreachable IdP to `unknown` (reconciled later, never assumed minted). Effects are emitted only when the operator grants the plugin `emit_effect` and configures an effect WAL; otherwise the mint path is unchanged. There is deliberately no OAuth-specific reconciler — an IdP exposes no lookup by mint key, so the core default (log and leave `unknown`) is the honest behavior. (#XXX) +- **The reference `audit-logger` renders the new provenance.** Decision records now include the invocation `span`, a `taint` object (the labels the request arrived with vs. the labels after the pipeline — their difference is the taint this node added), and, when content provenance is enabled, a `content` object with the input and output payload hashes (`sha256:…`, digests only, never the content itself). (#XXX) ### Fixed diff --git a/Cargo.lock b/Cargo.lock index 3082bddd..2e50f591 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -798,10 +798,12 @@ dependencies = [ "async-trait", "chrono", "cpex-orchestration", + "futures", "hashbrown 0.17.1", "serde", "serde_json", "serde_yaml", + "sha2 0.10.9", "thiserror 2.0.18", "tokio", "tokio-util", @@ -954,6 +956,7 @@ dependencies = [ "serde_json", "tokio", "tracing", + "uuid", "zeroize", ] diff --git a/builtins/plugins/audit-logger/src/factory.rs b/builtins/plugins/audit-logger/src/factory.rs index 2ef785c5..0388628d 100644 --- a/builtins/plugins/audit-logger/src/factory.rs +++ b/builtins/plugins/audit-logger/src/factory.rs @@ -25,16 +25,10 @@ impl PluginFactory for AuditLoggerFactory { fn create(&self, config: &PluginConfig) -> Result> { let logger = Arc::new(AuditLogger::new(config.clone())?); - if config.hooks.is_empty() { - return Err(Box::new(PluginError::Config { - message: format!( - "plugin '{}' (cpex-plugin-audit-logger): `hooks:` must list at \ - least one CMF hook to audit (e.g. cmf.tool_pre_invoke)", - config.name - ), - })); - } - + // With no `hooks:` listed the logger runs in audit-only mode — it + // registers no CMF post-hook handlers and instead auto-attaches as a + // decision-audit sink (see `Plugin::as_audit_handler`). Listing hooks + // keeps the legacy per-hook observation behavior. let handlers: Vec<_> = config .hooks .iter() diff --git a/builtins/plugins/audit-logger/src/logger.rs b/builtins/plugins/audit-logger/src/logger.rs index aade654c..cb405481 100644 --- a/builtins/plugins/audit-logger/src/logger.rs +++ b/builtins/plugins/audit-logger/src/logger.rs @@ -8,10 +8,13 @@ use std::sync::Arc; use async_trait::async_trait; use serde_json::{json, Map, Value}; +use cpex_core::audit::AuditHandler; use cpex_core::cmf::{CmfHook, ContentPart, MessagePayload}; use cpex_core::context::PluginContext; +use cpex_core::decision::{DecisionLog, Verdict}; +use cpex_core::effect::EffectRecord; use cpex_core::error::PluginError; -use cpex_core::hooks::payload::Extensions; +use cpex_core::hooks::payload::{Extensions, PluginPayload}; use cpex_core::hooks::trait_def::{HookHandler, PluginResult}; use cpex_core::plugin::{Plugin, PluginConfig}; @@ -42,7 +45,7 @@ impl AuditLogger { Ok(Self { cfg, typed }) } - fn build_record(&self, payload: &MessagePayload, ext: &Extensions) -> Value { + fn build_record(&self, payload: Option<&MessagePayload>, ext: &Extensions) -> Value { let mut record = Map::new(); record.insert( "ts".into(), @@ -93,7 +96,7 @@ impl AuditLogger { // content part's args, if any. Mirrors what the gateway // would actually forward (so audit reflects post-redact // state if a PII scanner ran ahead of us). - for part in &payload.message.content { + for part in payload.iter().flat_map(|p| p.message.content.iter()) { match part { ContentPart::ToolCall { content } => { record.insert( @@ -166,6 +169,18 @@ impl Plugin for AuditLogger { fn config(&self) -> &PluginConfig { &self.cfg } + + /// Auto-attach as a decision-audit sink when run in audit-only mode (no + /// `hooks:` listed). If the operator listed hooks, this runs as a legacy + /// CMF post-hook handler instead and does not also auto-attach, so + /// records aren't emitted twice. + fn as_audit_handler(self: Arc) -> Option> { + if self.cfg.hooks.is_empty() { + Some(self) + } else { + None + } + } } impl HookHandler for AuditLogger { @@ -175,12 +190,161 @@ impl HookHandler for AuditLogger { ext: &Extensions, _ctx: &mut PluginContext, ) -> PluginResult { - let record = self.build_record(payload, ext); + let record = self.build_record(Some(payload), ext); self.emit(&record); PluginResult::allow() } } +impl AuditLogger { + /// Build the decision-audit record: the same fields as the CMF + /// observation record, plus the pipeline's verdict and the ordered + /// plugin actions. `payload` is present only when this dispatch carried + /// a CMF `MessagePayload` (audit sinks fire for every hook family). + fn build_decision_record( + &self, + payload: Option<&MessagePayload>, + ext: &Extensions, + decisions: &DecisionLog, + ) -> Value { + let mut record = self.build_record(payload, ext); + if let Value::Object(map) = &mut record { + let verdict = match decisions.verdict() { + Some(Verdict::Allow) => json!("allow"), + Some(Verdict::Deny(v)) => json!({ + "deny": { "code": v.code, "reason": v.reason } + }), + None => json!("pending"), + }; + map.insert("verdict".into(), verdict); + + let steps: Vec = decisions + .steps() + .iter() + .map(|s| { + json!({ + "plugin": s.plugin_name, + "phase": format!("{:?}", s.phase), + "action": format!("{:?}", s.action), + }) + }) + .collect(); + map.insert("decision_steps".into(), json!(steps)); + + // The invocation's node identity in the decision graph: its own + // span, the upstream call that triggered it (causal parent), and + // the trace they share. Downstream joins these into a causal DAG. + if let Some(span) = decisions.span() { + map.insert( + "span".into(), + json!({ + "trace_id": span.trace_id, + "span_id": span.span_id, + "parent_span_id": span.parent_span_id, + }), + ); + } + + // Taint provenance: the labels the request arrived with vs. the + // labels after the pipeline. Their difference is the taint this + // node added — a taint edge in the decision graph. + let input_labels: Vec<&String> = decisions.input_labels().iter().collect(); + let final_labels: Vec = ext + .security + .as_ref() + .map(|s| { + let mut l: Vec = s.labels.iter().cloned().collect(); + l.sort_unstable(); + l + }) + .unwrap_or_default(); + if !input_labels.is_empty() || !final_labels.is_empty() { + map.insert( + "taint".into(), + json!({ "input": input_labels, "final": final_labels }), + ); + } + + // Content-addressed provenance: the input hash (captured at entry + // when enabled) plus this node's output hash. Gated on input_hash + // presence — when provenance is off it is `None` and we emit + // neither. Only digests, never content. + 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)); + map.insert( + "content".into(), + json!({ "input_hash": input_hash, "output_hash": output_hash }), + ); + } + + // Stream identity + sequences. `stream_seq` is gap-free within the + // decision stream (a consumer proves none was dropped); the global + // `emission_seq` orders this record against the effect records a + // consumer merges into the same chain. + if let Some(stream_seq) = decisions.stream_seq() { + map.insert("epoch".into(), json!(decisions.epoch())); + map.insert("stream_id".into(), json!(decisions.stream_id())); + map.insert("stream_seq".into(), json!(stream_seq)); + map.insert("emission_seq".into(), json!(decisions.emission_seq())); + } + } + record + } + + /// Build an audit record for an irreversible effect: the ambient identity + /// / delegation context from `ext` (reusing `build_record`) plus the + /// effect's own facts and lifecycle state. + fn build_effect_record(&self, effect: &EffectRecord, ext: &Extensions) -> Value { + // No payload — an effect is a side-effect, not a message. The ambient + // context (subject, delegation) still comes from `ext`. + let mut record = self.build_record(None, ext); + if let Value::Object(map) = &mut record { + map.insert( + "effect".into(), + json!({ + "kind": effect.kind, + "description": effect.description, + "key": effect.key, + "state": format!("{:?}", effect.state), + "caused_by": effect.plugin_name, + "details": effect.details, + "epoch": effect.epoch, + "stream_id": effect.stream_id, + "stream_seq": effect.stream_seq, + "emission_seq": effect.emission_seq, + }), + ); + } + record + } +} + +/// Decision-audit consumer: fires at the verdict of every pipeline run — +/// including denials — with the decision log. This is the first-class path; +/// the `HookHandler` impl above remains for the legacy post-hook +/// registration. +#[async_trait] +impl AuditHandler for AuditLogger { + async fn handle(&self, payload: &dyn PluginPayload, ext: &Extensions, decisions: &DecisionLog) { + // Downcast to the CMF payload when present; a non-CMF dispatch + // (delegation, identity) records without the message summary. + let msg = payload.as_any().downcast_ref::(); + let record = self.build_decision_record(msg, ext, decisions); + self.emit(&record); + } + + async fn on_effect(&self, effect: &EffectRecord, ext: &Extensions) { + let record = self.build_effect_record(effect, ext); + self.emit(&record); + } + + fn name(&self) -> &str { + &self.cfg.name + } +} + // Silence import-unused warning if Arc isn't used elsewhere. #[allow(dead_code)] fn _force_link_arc(_: Arc<()>) {} @@ -240,15 +404,145 @@ mod tests { ..Default::default() }; - let record = plugin.build_record(&payload, &ext); + let record = plugin.build_record(Some(&payload), &ext); assert_eq!(record["subject"]["id"], "alice@corp.com"); assert_eq!(record["entity"]["name"], "get_compensation"); assert_eq!(record["tool_call"]["name"], "get_compensation"); assert_eq!(record["tool_call"]["args"]["employee_id"], "EMP-001234"); // Always-allow contract: handler returns continue_processing. let mut ctx = PluginContext::default(); - let r = plugin.handle(&payload, &ext, &mut ctx).await; + let r = + >::handle(&plugin, &payload, &ext, &mut ctx).await; assert!(r.continue_processing); assert!(r.violation.is_none()); } + + #[test] + fn decision_record_includes_verdict_and_steps() { + use cpex_core::decision::PluginAction; + use cpex_core::error::PluginViolation; + + let plugin = AuditLogger::new(cfg()).unwrap(); + let mut log = DecisionLog::new(); + log.record("cedar-pdp", PluginMode::Sequential, PluginAction::Denied); + log.finalize(Verdict::Deny(PluginViolation::new( + "missing_permission", + "not allowed", + ))); + + // No CMF payload on this dispatch — the record still carries the verdict. + let record = plugin.build_decision_record(None, &Extensions::default(), &log); + assert_eq!(record["verdict"]["deny"]["code"], "missing_permission"); + assert_eq!(record["decision_steps"][0]["plugin"], "cedar-pdp"); + assert_eq!(record["decision_steps"][0]["action"], "Denied"); + // No span was set on this log → no span field emitted. + assert!(record.get("span").is_none()); + } + + #[test] + fn decision_record_includes_span_when_set() { + use cpex_core::decision::Span; + + let plugin = AuditLogger::new(cfg()).unwrap(); + let mut log = DecisionLog::new(); + log.set_span(Span::for_request(Some("trace-abc"), Some("upstream-span"))); + log.finalize(Verdict::Allow); + + let record = plugin.build_decision_record(None, &Extensions::default(), &log); + assert_eq!(record["span"]["trace_id"], "trace-abc"); + assert_eq!(record["span"]["parent_span_id"], "upstream-span"); + assert!(record["span"]["span_id"] + .as_str() + .is_some_and(|s| !s.is_empty())); + } + + #[test] + fn decision_record_includes_taint_delta() { + let plugin = AuditLogger::new(cfg()).unwrap(); + let mut log = DecisionLog::new(); + log.set_input_labels(vec!["PII".into()]); // the request arrived carrying PII + log.finalize(Verdict::Allow); + + // Final state: the pipeline added `secret`. + let mut sec = SecurityExtension::default(); + sec.labels.insert("PII".into()); + sec.labels.insert("secret".into()); + let ext = Extensions { + security: Some(Arc::new(sec)), + ..Default::default() + }; + + let record = plugin.build_decision_record(None, &ext, &log); + assert_eq!(record["taint"]["input"], serde_json::json!(["PII"])); + assert_eq!( + record["taint"]["final"], + serde_json::json!(["PII", "secret"]) + ); + } + + #[test] + fn decision_record_includes_content_hashes_when_captured() { + let plugin = AuditLogger::new(cfg()).unwrap(); + let mut log = DecisionLog::new(); + log.set_input_hash(Some("sha256:deadbeef".into())); + log.finalize(Verdict::Allow); + + // No payload on this dispatch → output_hash is null, input present. + let record = plugin.build_decision_record(None, &Extensions::default(), &log); + assert_eq!(record["content"]["input_hash"], "sha256:deadbeef"); + assert!(record["content"]["output_hash"].is_null()); + } + + #[test] + fn decision_record_includes_stream_and_sequences() { + let plugin = AuditLogger::new(cfg()).unwrap(); + let mut log = DecisionLog::new(); + log.set_stream(1_700_000_000, "decision".into(), 7, 42); + log.finalize(Verdict::Allow); + + let record = plugin.build_decision_record(None, &Extensions::default(), &log); + assert_eq!(record["epoch"], 1_700_000_000u64); + assert_eq!(record["stream_id"], "decision"); + assert_eq!(record["stream_seq"], 7); + assert_eq!(record["emission_seq"], 42); + } + + #[test] + fn no_content_field_without_input_hash() { + // Provenance off (input_hash None) → no content field at all. + let plugin = AuditLogger::new(cfg()).unwrap(); + let mut log = DecisionLog::new(); + log.finalize(Verdict::Allow); + let record = plugin.build_decision_record(None, &Extensions::default(), &log); + assert!(record.get("content").is_none()); + } + + #[test] + fn effect_record_carries_effect_and_ambient_identity() { + use cpex_core::effect::EffectRecord; + + let plugin = AuditLogger::new(cfg()).unwrap(); + + // Ambient context: a subject in extensions. + let mut sec = SecurityExtension::default(); + sec.subject = Some(SubjectExtension { + id: Some("alice@corp.com".into()), + ..Default::default() + }); + let ext = Extensions { + security: Some(Arc::new(sec)), + ..Default::default() + }; + + let effect = EffectRecord::prepared("token_mint", "exchange for workday-api", "k-1") + .with_detail("audience", "workday-api"); + + let record = plugin.build_effect_record(&effect, &ext); + // The effect's own facts… + assert_eq!(record["effect"]["kind"], "token_mint"); + assert_eq!(record["effect"]["state"], "Prepared"); + assert_eq!(record["effect"]["details"]["audience"], "workday-api"); + // …alongside ambient identity from ext (the reason we pass ext): + assert_eq!(record["subject"]["id"], "alice@corp.com"); + } } diff --git a/builtins/plugins/delegator-oauth/Cargo.toml b/builtins/plugins/delegator-oauth/Cargo.toml index d21ea22f..f8f2a715 100644 --- a/builtins/plugins/delegator-oauth/Cargo.toml +++ b/builtins/plugins/delegator-oauth/Cargo.toml @@ -59,6 +59,8 @@ serde_json = { workspace = true } tokio = { workspace = true } chrono = { workspace = true } tracing = { workspace = true } +# `uuid` for the per-mint effect key (a unique attempt id in the effect WAL). +uuid = { workspace = true } # `base64` decodes the minted token's JWT payload for a best-effort, # read-only interop check (did the IdP honor the RFC 8693 `actor_token` diff --git a/builtins/plugins/delegator-oauth/src/delegator.rs b/builtins/plugins/delegator-oauth/src/delegator.rs index 1de0374b..a987a822 100644 --- a/builtins/plugins/delegator-oauth/src/delegator.rs +++ b/builtins/plugins/delegator-oauth/src/delegator.rs @@ -51,6 +51,7 @@ use zeroize::Zeroizing; use cpex_core::context::PluginContext; use cpex_core::delegation::{DelegationPayload, DelegationSubject, TokenDelegateHook}; +use cpex_core::effect::{EffectRecord, EffectState}; use cpex_core::error::{PluginError, PluginViolation}; use cpex_core::extensions::raw_credentials::RawDelegatedToken; use cpex_core::hooks::payload::Extensions; @@ -275,6 +276,80 @@ impl OAuthDelegator { )), } } + + /// Bracket a token-mint I/O with write-ahead effect audit. Durably records + /// the mint *intent* (fail-closed — no durable record, no mint) before + /// `mint` runs, then records the outcome: + /// + /// - `Confirmed` on success, or on a 2xx we couldn't parse (the token *was* + /// minted; we just couldn't read it); + /// - `Rejected` on a definitive IdP rejection (a non-2xx response — + /// provably not minted); + /// - `Unknown` on an ambiguous failure (timeout / unreachable — the mint + /// may or may not have landed), left for the recovery sweep. + /// + /// A no-op unless the operator granted this plugin the `emit_effect` + /// capability *and* configured an effect WAL; otherwise `begin_effect` / + /// `complete_effect` do nothing and this just runs `mint`. + async fn audit_mint( + &self, + ext: &Extensions, + description: &str, + audience: &str, + scope: &str, + mint: F, + ) -> Result + where + F: FnOnce() -> Fut, + Fut: std::future::Future>, + { + let intent = + EffectRecord::prepared("token_mint", description.to_string(), new_effect_key()) + .with_detail("audience", audience.to_string()) + .with_detail("scope", scope.to_string()) + .with_detail("token_endpoint", self.typed.token_endpoint.clone()); + + // Write-ahead, fail-closed: if the intent can't be durably recorded, + // do not mint. + if let Err(e) = ext.begin_effect(&intent).await { + return Err(PluginViolation::new( + "delegation.effect_wal_failed", + format!("could not durably record token-mint intent: {e}"), + )); + } + + let outcome = mint().await; + + let state = match &outcome { + Ok(_) => EffectState::Confirmed, + Err(v) => match v.code.as_str() { + // A non-2xx IdP response — provably not minted. + "delegation.idp_rejected" => EffectState::Rejected, + // 2xx but unparseable: the token WAS minted; we just couldn't + // read it. The effect happened, even though the delegation fails. + "delegation.bad_response" => EffectState::Confirmed, + // Timeout / unreachable — the mint may have landed; reconcile. + _ => EffectState::Unknown, + }, + }; + // Best-effort completion: the act already happened, so a completion + // write failure is not fatal (recovery reconciles by the intent's key). + let _ = ext.complete_effect(&intent, state).await; + + outcome + } +} + +/// A fresh per-mint effect key — a unique attempt id in the effect WAL. OAuth +/// token exchange has no idempotency key, so this identifies the attempt for +/// the WAL rather than enabling IdP reconciliation. +/// +/// There is deliberately no OAuth-specific `EffectReconciler`: an IdP exposes +/// no lookup by mint key, so an `unknown` mint cannot be resolved against it — +/// the core `LogUnknownsReconciler` default (log + leave `unknown`) is exactly +/// the honest behavior, and nothing here is plugin-specific. +fn new_effect_key() -> String { + uuid::Uuid::new_v4().to_string() } /// Subset of the RFC 8693 response we care about. @@ -313,7 +388,7 @@ impl HookHandler for OAuthDelegator { async fn handle( &self, payload: &DelegationPayload, - _ext: &Extensions, + ext: &Extensions, _ctx: &mut PluginContext, ) -> PluginResult { // `subject: this_workload` means *we* are the principal. There @@ -356,7 +431,16 @@ impl HookHandler for OAuthDelegator { // own `bearer` directly. `Cow` avoids cloning the (already // borrowed) bearer on the non-workload path. let subject_token: Cow = if is_workload { - match self.mint_base_token(bearer).await { + match self + .audit_mint( + ext, + "base-token mint (client_credentials)", + audience, + &scope, + || self.mint_base_token(bearer), + ) + .await + { Ok(token) => Cow::Owned(token), Err(violation) => return PluginResult::deny(violation), } @@ -412,63 +496,80 @@ impl HookHandler for OAuthDelegator { form.push(("actor_token_type", &self.typed.actor_token_type)); } - // POST to the IdP. Basic auth carries our client credentials. - let response = match self - .http - .post(&self.typed.token_endpoint) - .basic_auth(&self.typed.client_id, Some(self.client_secret.as_str())) - .form(&form) - .send() - .await - { - Ok(r) => r, - Err(e) if e.is_timeout() => { - return PluginResult::deny(PluginViolation::new( - "delegation.idp_timeout", - format!("token-exchange to {} timed out", self.typed.token_endpoint), - )); - }, - Err(e) => { - return PluginResult::deny(PluginViolation::new( - "delegation.idp_unreachable", - format!( - "token-exchange POST to {} failed: {e}", - self.typed.token_endpoint, - ), - )); - }, - }; + // POST to the IdP, bracketed by write-ahead effect audit. Basic auth + // carries our client credentials. + let parsed = match self + .audit_mint( + ext, + "token exchange (RFC 8693)", + audience, + &scope, + || async { + let response = match self + .http + .post(&self.typed.token_endpoint) + .basic_auth(&self.typed.client_id, Some(self.client_secret.as_str())) + .form(&form) + .send() + .await + { + Ok(r) => r, + Err(e) if e.is_timeout() => { + return Err(PluginViolation::new( + "delegation.idp_timeout", + format!( + "token-exchange to {} timed out", + self.typed.token_endpoint + ), + )); + }, + Err(e) => { + return Err(PluginViolation::new( + "delegation.idp_unreachable", + format!( + "token-exchange POST to {} failed: {e}", + self.typed.token_endpoint, + ), + )); + }, + }; + + let status = response.status(); + if !status.is_success() { + // Surface the standard `error` / `error_description` fields + // from the IdP; fall back to the status code. + let body = response.text().await.unwrap_or_default(); + let (code, reason) = match serde_json::from_str::(&body) + { + Ok(err) => { + let mut reason = err.error.clone(); + if let Some(desc) = err.error_description { + reason.push_str(": "); + reason.push_str(&desc); + } + ("delegation.idp_rejected", reason) + }, + Err(_) => ( + "delegation.idp_rejected", + format!("IdP returned {status}: {body}"), + ), + }; + return Err(PluginViolation::new(code, reason)); + } - let status = response.status(); - if !status.is_success() { - // Try to surface the standard `error` / `error_description` - // fields from the IdP. Fall back to status code. - let body = response.text().await.unwrap_or_default(); - let (code, reason) = match serde_json::from_str::(&body) { - Ok(err) => { - let mut reason = err.error.clone(); - if let Some(desc) = err.error_description { - reason.push_str(": "); - reason.push_str(&desc); + match response.json::().await { + Ok(p) => Ok(p), + Err(e) => Err(PluginViolation::new( + "delegation.bad_response", + format!("IdP response wasn't valid token-exchange JSON: {e}"), + )), } - ("delegation.idp_rejected", reason) }, - Err(_) => ( - "delegation.idp_rejected", - format!("IdP returned {status}: {body}"), - ), - }; - return PluginResult::deny(PluginViolation::new(code, reason)); - } - - let parsed = match response.json::().await { + ) + .await + { Ok(p) => p, - Err(e) => { - return PluginResult::deny(PluginViolation::new( - "delegation.bad_response", - format!("IdP response wasn't valid token-exchange JSON: {e}"), - )); - }, + Err(v) => return PluginResult::deny(v), }; // Compute effective scopes. IdP's `scope` field wins (it diff --git a/builtins/plugins/delegator-oauth/tests/oauth_e2e.rs b/builtins/plugins/delegator-oauth/tests/oauth_e2e.rs index 23b9ffe4..7fa55932 100644 --- a/builtins/plugins/delegator-oauth/tests/oauth_e2e.rs +++ b/builtins/plugins/delegator-oauth/tests/oauth_e2e.rs @@ -728,3 +728,137 @@ async fn leg1_rejection_does_not_leak_the_client_assertion() { violation.reason, ); } + +// ===================================================================== +// Effect audit — the token mint is recorded as a write-ahead effect. +// ===================================================================== + +/// Build a manager whose delegator holds the `emit_effect` capability, with a +/// capturing audit sink attached. Returns the manager and the shared record of +/// (effect kind, state) the sink observed. +async fn build_manager_with_effect_audit( + token_endpoint: &str, +) -> ( + Arc, + Arc>>, +) { + use cpex_core::audit::AuditHandler; + use cpex_core::decision::DecisionLog; + use cpex_core::effect::EffectRecord; + use cpex_core::hooks::payload::PluginPayload; + use std::sync::Mutex; + + struct CapturingEffectAudit { + seen: Arc>>, + } + #[async_trait::async_trait] + impl AuditHandler for CapturingEffectAudit { + async fn handle(&self, _p: &dyn PluginPayload, _e: &Extensions, _d: &DecisionLog) {} + async fn on_effect(&self, effect: &EffectRecord, _ext: &Extensions) { + self.seen + .lock() + .unwrap() + .push((effect.kind.clone(), format!("{:?}", effect.state))); + } + } + + let mut cfg = plugin_config(token_endpoint); + cfg.capabilities.insert("emit_effect".into()); + let delegator = OAuthDelegator::new(cfg.clone()).expect("delegator constructs"); + let mgr = Arc::new(PluginManager::default()); + mgr.register_handler_for_names::( + Arc::new(delegator), + cfg, + &[HOOK_TOKEN_DELEGATE], + ) + .unwrap(); + let seen = Arc::new(Mutex::new(Vec::new())); + mgr.register_audit_handler(Arc::new(CapturingEffectAudit { seen: seen.clone() })); + mgr.initialize().await.unwrap(); + (mgr, seen) +} + +/// A successful exchange records the mint as prepared → confirmed. +#[tokio::test] +async fn successful_mint_emits_prepared_then_confirmed() { + let mut server = Server::new_async().await; + let _mock = server + .mock("POST", "/oauth/token") + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + json!({ + "access_token": "minted-downstream-jwt", + "expires_in": 300, + "scope": "read:compensation", + }) + .to_string(), + ) + .create_async() + .await; + + let (mgr, seen) = + build_manager_with_effect_audit(&format!("{}/oauth/token", server.url())).await; + let payload = build_payload( + "get_compensation", + "https://hr.example.com", + &["read:compensation"], + ); + let result = invoke(&mgr, payload).await; + assert!( + result.continue_processing, + "mint should succeed: {:?}", + result.violation + ); + + let seen = seen.lock().unwrap(); + assert!( + seen.iter() + .any(|(k, s)| k == "token_mint" && s == "Prepared"), + "write-ahead prepared intent emitted; got {seen:?}", + ); + assert!( + seen.iter() + .any(|(k, s)| k == "token_mint" && s == "Confirmed"), + "confirmed outcome emitted; got {seen:?}", + ); +} + +/// A definitive IdP rejection (HTTP 400) records the mint as prepared → +/// rejected, and denies the delegation. +#[tokio::test] +async fn rejected_mint_emits_prepared_then_rejected() { + let mut server = Server::new_async().await; + let _mock = server + .mock("POST", "/oauth/token") + .with_status(400) + .with_header("content-type", "application/json") + .with_body(json!({ "error": "invalid_grant" }).to_string()) + .create_async() + .await; + + let (mgr, seen) = + build_manager_with_effect_audit(&format!("{}/oauth/token", server.url())).await; + let payload = build_payload( + "get_compensation", + "https://hr.example.com", + &["read:compensation"], + ); + let result = invoke(&mgr, payload).await; + assert!( + !result.continue_processing, + "a 400 must deny the delegation" + ); + + let seen = seen.lock().unwrap(); + assert!( + seen.iter() + .any(|(k, s)| k == "token_mint" && s == "Prepared"), + "prepared intent emitted even on rejection; got {seen:?}", + ); + assert!( + seen.iter() + .any(|(k, s)| k == "token_mint" && s == "Rejected"), + "rejected outcome emitted; got {seen:?}", + ); +} diff --git a/crates/cpex-core/Cargo.toml b/crates/cpex-core/Cargo.toml index 1f0d90aa..dc3f7737 100644 --- a/crates/cpex-core/Cargo.toml +++ b/crates/cpex-core/Cargo.toml @@ -27,11 +27,14 @@ serde = { workspace = true } serde_yaml = { workspace = true } serde_json = { workspace = true } async-trait = { workspace = true } +futures = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } uuid = { workspace = true } hashbrown = { workspace = true } arc-swap = { workspace = true } +# `sha2` for content-addressed audit provenance (input/output payload hashes). +sha2 = "0.10" wildmatch = { workspace = true } chrono = { workspace = true } # Zeroizing wrapper for raw credential material in RawCredentialsExtension. diff --git a/crates/cpex-core/src/audit.rs b/crates/cpex-core/src/audit.rs new file mode 100644 index 00000000..6f662e34 --- /dev/null +++ b/crates/cpex-core/src/audit.rs @@ -0,0 +1,79 @@ +// Location: ./crates/cpex-core/src/audit.rs +// Copyright 2026 +// SPDX-License-Identifier: Apache-2.0 +// Authors: Teryl Taylor +// +// The audit-hook consumer: an observation-only sink invoked at the +// pipeline's verdict with the decision log. +// +// This is deliberately NOT a `HookHandler` (whose `handle` returns a +// `PluginResult` and so can allow/deny/modify). An audit sink returns +// nothing — the type is the contract: it *sees* the verdict and every +// plugin's action, but it cannot influence them. The manager auto-attaches +// these; the executor invokes them once per pipeline run, at the verdict, +// with the final payload, extensions, and the decision log. The decision +// log is passed directly here and never placed on `PluginContext`, so no +// ordinary plugin can read what the audit sink reads. + +use async_trait::async_trait; + +use crate::decision::DecisionLog; +use crate::effect::EffectRecord; +use crate::hooks::payload::{Extensions, PluginPayload}; + +/// An observation-only consumer of pipeline decisions. +/// +/// Implemented by audit plugins (e.g. `audit-logger`, `ocsf-audit`). The +/// executor calls [`AuditHandler::handle`] once per pipeline invocation, +/// after the verdict is decided, for both allowed and denied requests. +#[async_trait] +pub trait AuditHandler: Send + Sync { + /// Observe one finished pipeline invocation. Must not block or mutate + /// anything the pipeline depends on — its return is `()` by design. + /// + /// **Awaited at the verdict return point — a stable contract, not + /// fire-and-forget.** The executor `await`s this call *before* it returns + /// the pipeline result. That is deliberate: a crash cannot lose a verdict + /// that was emitted, so downstream evidence chains need no drop-detection + /// for the steady state. Consumers rely on this — a future change to + /// fire-and-forget would be a silent semantics break, so it must not be + /// made lightly. + /// + /// The cost of that guarantee is that **sink latency sits on the request + /// path** (bounded per sink by the plugin timeout with panic containment, + /// and sinks run sequentially). Keep `handle` cheap — serialize / hash / + /// append. A slower sink (a network destination, say) should hand off to + /// an internal queue on its own side of this boundary rather than block + /// here. + /// + /// * `payload` — the message as it stood at the verdict. + /// * `extensions` — the final extensions (identity, delegation, labels…). + /// * `decisions` — what each plugin did and how the pipeline ruled. + async fn handle( + &self, + payload: &dyn PluginPayload, + extensions: &Extensions, + decisions: &DecisionLog, + ); + + /// Observe an irreversible external effect a plugin *caused* — a token + /// mint, an approval grant — as its own event, separate from the + /// per-invocation decision. Fired at each lifecycle transition + /// (`prepared` → `confirmed` | `rejected` | `unknown`). + /// + /// `extensions` carries the same ambient context a decision sink gets — + /// identity, delegation, correlation (conversation / span) — so a sink can + /// build a correlatable, richly-typed event (e.g. an OCSF Authentication + /// event for a token mint, in the same attestation chain) rather than + /// working from the effect alone. `EffectRecord` stays effect-specific. + /// + /// Default: ignore. A sink that only cares about decisions need not + /// implement this; a sink that cares about effects overrides it. + async fn on_effect(&self, _effect: &EffectRecord, _extensions: &Extensions) {} + + /// A short identifier used in error logs when a sink panics or times + /// out. Defaults to `"audit"`; override to distinguish sinks. + fn name(&self) -> &str { + "audit" + } +} diff --git a/crates/cpex-core/src/cmf/message.rs b/crates/cpex-core/src/cmf/message.rs index a8818b4d..3e837cf3 100644 --- a/crates/cpex-core/src/cmf/message.rs +++ b/crates/cpex-core/src/cmf/message.rs @@ -224,7 +224,7 @@ pub struct MessagePayload { pub message: Message, } -crate::impl_plugin_payload!(MessagePayload); +crate::impl_plugin_payload!(MessagePayload, audit_serialize); crate::define_hook! { /// CMF message evaluation hook. diff --git a/crates/cpex-core/src/config.rs b/crates/cpex-core/src/config.rs index f758f430..cb9ccccb 100644 --- a/crates/cpex-core/src/config.rs +++ b/crates/cpex-core/src/config.rs @@ -115,6 +115,30 @@ pub struct PluginSettings { /// investigate the entity-name growth. #[serde(default = "default_route_cache_max_entries")] pub route_cache_max_entries: usize, + + /// Optional path to a durable write-ahead log for irreversible-effect + /// audit records (token mints, approval grants). When set, `begin_effect` + /// becomes crash-safe and fail-closed — the intent is fsync'd before the + /// act. When unset (default), effect auditing is ordering-only: records + /// still reach the audit sinks, but without durability. Opt-in — basic + /// logging leaves this empty. + #[serde(default)] + pub effect_log_path: Option, + + /// Override the effect WAL's auto-compaction threshold — the number of + /// appends between automatic compactions. Only meaningful with + /// `effect_log_path` set; `0` disables auto-compaction (compaction then + /// happens only on an explicit recovery). Unset uses the built-in default. + #[serde(default)] + pub effect_log_compaction_threshold: Option, + + /// Capture content-addressed provenance for audit: the executor hashes the + /// payload at pipeline entry (`sha256:`) so audit sinks can record an + /// input content ref without the raw content. Off by default — hashing is + /// on the request path, so it is opt-in. Only the digest is kept, never the + /// bytes. + #[serde(default)] + pub capture_content_provenance: bool, } impl Default for PluginSettings { @@ -126,6 +150,9 @@ impl Default for PluginSettings { parallel_execution_within_band: false, fail_on_plugin_error: false, route_cache_max_entries: default_route_cache_max_entries(), + effect_log_path: None, + effect_log_compaction_threshold: None, + capture_content_provenance: false, } } } @@ -1162,6 +1189,35 @@ plugins: .contains("duplicate plugin name")); } + #[test] + fn parses_effect_log_settings() { + let yaml = r#" +plugin_settings: + effect_log_path: /var/lib/cpex/effects.wal + effect_log_compaction_threshold: 256 +plugins: [] +"#; + let cfg = parse_config(yaml).unwrap(); + assert_eq!( + cfg.plugin_settings.effect_log_path.as_deref(), + Some("/var/lib/cpex/effects.wal") + ); + assert_eq!( + cfg.plugin_settings.effect_log_compaction_threshold, + Some(256) + ); + } + + #[test] + fn effect_log_settings_default_to_none() { + let cfg = parse_config("plugins: []\n").unwrap(); + assert!(cfg.plugin_settings.effect_log_path.is_none()); + assert!(cfg + .plugin_settings + .effect_log_compaction_threshold + .is_none()); + } + #[test] fn test_route_requires_one_entity_matcher() { let yaml = r#" diff --git a/crates/cpex-core/src/decision.rs b/crates/cpex-core/src/decision.rs new file mode 100644 index 00000000..c44d3d60 --- /dev/null +++ b/crates/cpex-core/src/decision.rs @@ -0,0 +1,383 @@ +// Location: ./crates/cpex-core/src/decision.rs +// Copyright 2026 +// SPDX-License-Identifier: Apache-2.0 +// Authors: Teryl Taylor +// +// The DecisionLog — the executor's private, append-only record of what +// each plugin did to a request and how the pipeline ruled on it. +// +// Why it exists: a plugin observing a request (audit-logger, ocsf-audit) +// cannot see the pipeline's verdict — allow/deny/modify lives in the +// executor's control flow (`PluginResult`, the short-circuit return), not +// in `Extensions`. The DecisionLog captures that control flow so an audit +// sink can serialize it. It is built by the executor and handed only to +// audit handlers; it is deliberately NOT placed on `PluginContext`, which +// every plugin can read — the component that records must not be readable +// (or writable) by the components it records. +// +// Kept cheap: it records what happened (which plugin, which phase, which +// action), not copies of payloads. + +use crate::error::PluginViolation; +use crate::plugin::PluginMode; + +/// What a single plugin did to the request, from the executor's point of +/// view. Derived from the plugin's `PluginResult`, not self-reported. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PluginAction { + /// Ran and let the request continue unchanged. + Allowed, + /// Blocked the request. The full violation rides on the terminal + /// [`Verdict::Deny`]; this marks *which* plugin, in order. + Denied, + /// Replaced the payload (accepted by the executor's modify path). + ModifiedPayload, + /// Wrote to an extension slot it was capable of writing. + ModifiedExtensions, + /// Failed. The string is the error rendered by the executor; whether + /// this halts the pipeline is decided by the plugin's `on_error`. + Error(String), +} + +/// One entry in the log: a plugin, the phase it ran in, and what it did. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DecisionStep { + /// The plugin instance name (`PluginConfig.name`). + pub plugin_name: String, + /// The phase this plugin ran in — Sequential / Transform / Audit / … + pub phase: PluginMode, + /// What it did. + pub action: PluginAction, +} + +/// The pipeline's terminal ruling on a request. +#[derive(Debug, Clone)] +pub enum Verdict { + /// The request was allowed through (possibly after modifications — + /// those are in [`DecisionLog::steps`]). + Allow, + /// The request was blocked. Carries the fully-formed violation the + /// executor stamped with the deciding plugin's name. + Deny(PluginViolation), +} + +impl Verdict { + /// True if this verdict blocked the request. + pub fn is_deny(&self) -> bool { + matches!(self, Verdict::Deny(_)) + } +} + +/// The W3C trace context for one pipeline invocation — the node identity in +/// the decision graph. `span_id` is this interception's own span, +/// `parent_span_id` is the upstream call that triggered it (the causal edge), +/// and `trace_id` correlates the whole run. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Span { + /// The trace this invocation belongs to (W3C trace-id: 32 hex chars). + pub trace_id: String, + /// This interception's own span (W3C span-id: 16 hex chars). + pub span_id: String, + /// The span of the upstream call that caused this one — the causal edge. + /// `None` when the request carried no trace context (a trace root). + pub parent_span_id: Option, +} + +impl Span { + /// Derive the span for an interception from the request's trace context, + /// child-span model: a fresh `span_id` for this interception, the request's + /// `span_id` as the causal parent, and the request's `trace_id` carried + /// through — or a freshly originated trace root when the request carries + /// none. Adopts W3C ids; it does not invent a bespoke scheme. + pub fn for_request(trace_id: Option<&str>, parent_span_id: Option<&str>) -> Self { + Self { + trace_id: trace_id.map(str::to_owned).unwrap_or_else(new_trace_id), + span_id: new_span_id(), + parent_span_id: parent_span_id.map(str::to_owned), + } + } +} + +/// A freshly originated W3C trace-id: 16 bytes / 32 lowercase hex chars. +fn new_trace_id() -> String { + uuid::Uuid::new_v4().simple().to_string() +} + +/// A freshly originated W3C span-id: 8 bytes / 16 lowercase hex chars (the +/// first half of a UUID's hex). +fn new_span_id() -> String { + uuid::Uuid::new_v4().simple().to_string()[..16].to_string() +} + +/// The executor's record of one pipeline invocation: the ordered steps +/// each plugin took, the terminal verdict, and this invocation's span. +/// +/// `verdict` is `None` while the pipeline is still running and is set once +/// at a return point (allow or deny). An audit sink always receives a +/// finalized log. +#[derive(Debug, Clone, Default)] +pub struct DecisionLog { + steps: Vec, + verdict: Option, + span: Option, + input_labels: Vec, + input_hash: Option, + epoch: Option, + stream_id: Option, + stream_seq: Option, + emission_seq: Option, +} + +impl DecisionLog { + /// A fresh log for one pipeline invocation. + pub fn new() -> Self { + Self::default() + } + + /// Append what a plugin did. Called by the executor as each plugin + /// returns; order is execution order. + pub fn record( + &mut self, + plugin_name: impl Into, + phase: PluginMode, + action: PluginAction, + ) { + self.steps.push(DecisionStep { + plugin_name: plugin_name.into(), + phase, + action, + }); + } + + /// Set the terminal verdict. Called once at the pipeline's return + /// point, before the log is handed to audit handlers. + pub fn finalize(&mut self, verdict: Verdict) { + self.verdict = Some(verdict); + } + + /// Attach this invocation's span (trace context). Called by the executor + /// at pipeline entry, derived from the request via [`Span::for_request`]. + pub fn set_span(&mut self, span: Span) { + self.span = Some(span); + } + + /// This invocation's span (trace context) — the node identity and causal + /// parent for the decision graph — if the executor set one. + pub fn span(&self) -> Option<&Span> { + self.span.as_ref() + } + + /// Record the taint labels the request carried at pipeline entry — the + /// input side of this node's provenance. Diffed against the final labels + /// (on `Extensions.security`), it yields the taint the pipeline added. + pub fn set_input_labels(&mut self, labels: Vec) { + self.input_labels = labels; + } + + /// The taint labels present at pipeline entry. + pub fn input_labels(&self) -> &[String] { + &self.input_labels + } + + /// Record the content hash of the payload at pipeline entry — the input + /// side of this node's content provenance. Set by the executor only when + /// content provenance is enabled; otherwise `None`. + pub fn set_input_hash(&mut self, hash: Option) { + self.input_hash = hash; + } + + /// The content hash of the payload at pipeline entry, if captured. + pub fn input_hash(&self) -> Option<&str> { + self.input_hash.as_deref() + } + + /// Stamp the audit-stream identity + sequence numbers, assigned by the + /// executor at emission. The two counters are **distinct claims** — don't + /// use one for the other's job: + /// + /// - `epoch` — the executor's boot time (Unix nanoseconds), captured once + /// at startup. It scopes the counters so a verifier tells a *counter + /// reset* (new, larger epoch) from *records lost* (a gap within an + /// epoch); being ordered, `(epoch, emission_seq)` is a total order across + /// restarts, computable from the record alone. Cross-epoch tail-loss is + /// not provable from the counters alone — that is what a durable sink + /// (the ledger) is for. + /// - `stream_id` — the per-type stream (`"decision"`), the entry-type a + /// merged consumer keys on. + /// - `stream_seq` — a **completeness** claim. Dense (gap-free) within + /// `(epoch, stream_id)`; a gap means a record was dropped. + /// - `emission_seq` — an **ordering** claim *only*. Monotonic across all + /// streams within the epoch (decisions and effects share it) for + /// reconstructing interleaved order. A single-stream consumer sees it + /// *sparse* by design — the gaps are the other stream's records, never a + /// loss signal. + pub fn set_stream( + &mut self, + epoch: u64, + stream_id: String, + stream_seq: u64, + emission_seq: u64, + ) { + self.epoch = Some(epoch); + self.stream_id = Some(stream_id); + self.stream_seq = Some(stream_seq); + self.emission_seq = Some(emission_seq); + } + + /// The executor boot epoch (Unix nanoseconds) this record was emitted in. + /// Orderable, so a new/larger value marks a restart — a reset is + /// distinguishable from a loss, and it extends `emission_seq` to a total + /// order across restarts. + pub fn epoch(&self) -> Option { + self.epoch + } + + /// The per-type stream this record belongs to (scopes `stream_seq`). + pub fn stream_id(&self) -> Option<&str> { + self.stream_id.as_deref() + } + + /// **Completeness** counter — dense within `(epoch, stream_id)`; a gap is a + /// dropped record. + pub fn stream_seq(&self) -> Option { + self.stream_seq + } + + /// **Ordering** counter — monotonic across decisions and effects within the + /// epoch. Sparse for a single-stream consumer by design; not a loss signal. + pub fn emission_seq(&self) -> Option { + self.emission_seq + } + + /// The ordered steps taken this invocation. + pub fn steps(&self) -> &[DecisionStep] { + &self.steps + } + + /// The terminal verdict, or `None` if the pipeline hasn't returned yet. + pub fn verdict(&self) -> Option<&Verdict> { + self.verdict.as_ref() + } + + /// True once finalized with a deny. + pub fn is_denied(&self) -> bool { + self.verdict.as_ref().is_some_and(Verdict::is_deny) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn violation() -> PluginViolation { + PluginViolation::new("missing_permission", "not allowed") + } + + #[test] + fn records_steps_in_order() { + let mut log = DecisionLog::new(); + log.record( + "pii-scanner", + PluginMode::Transform, + PluginAction::ModifiedPayload, + ); + log.record("cedar-pdp", PluginMode::Sequential, PluginAction::Denied); + + let steps = log.steps(); + assert_eq!(steps.len(), 2); + assert_eq!(steps[0].plugin_name, "pii-scanner"); + assert_eq!(steps[0].action, PluginAction::ModifiedPayload); + assert_eq!(steps[1].phase, PluginMode::Sequential); + assert_eq!(steps[1].action, PluginAction::Denied); + } + + #[test] + fn verdict_is_none_until_finalized() { + let mut log = DecisionLog::new(); + assert!(log.verdict().is_none()); + assert!(!log.is_denied()); + + log.finalize(Verdict::Deny(violation())); + assert!(log.is_denied()); + match log.verdict() { + Some(Verdict::Deny(v)) => assert_eq!(v.code, "missing_permission"), + other => panic!("expected deny, got {other:?}"), + } + } + + #[test] + fn allow_verdict_is_not_a_deny() { + let mut log = DecisionLog::new(); + log.finalize(Verdict::Allow); + assert!(!log.is_denied()); + } + + #[test] + fn span_child_model_carries_causal_edge() { + let span = Span::for_request(Some("trace-abc"), Some("upstream-span")); + assert_eq!(span.trace_id, "trace-abc", "trace carried through"); + assert_eq!( + span.parent_span_id.as_deref(), + Some("upstream-span"), + "the request's span becomes the causal parent" + ); + assert_eq!(span.span_id.len(), 16, "own fresh W3C span-id"); + assert_ne!(span.span_id, "upstream-span", "our span, not the parent's"); + } + + #[test] + fn span_originates_trace_root_when_request_has_none() { + let span = Span::for_request(None, None); + assert_eq!(span.trace_id.len(), 32, "originated W3C trace-id"); + assert_eq!(span.span_id.len(), 16, "originated W3C span-id"); + assert!(span.parent_span_id.is_none(), "no parent = trace root"); + } + + #[test] + fn each_invocation_gets_a_distinct_span() { + let a = Span::for_request(Some("t"), Some("p")); + let b = Span::for_request(Some("t"), Some("p")); + assert_ne!(a.span_id, b.span_id, "each interception mints its own span"); + } + + #[test] + fn span_is_none_until_set() { + let mut log = DecisionLog::new(); + assert!(log.span().is_none()); + log.set_span(Span::for_request(Some("t"), None)); + assert_eq!(log.span().unwrap().trace_id, "t"); + } + + #[test] + fn input_labels_default_empty_and_settable() { + let mut log = DecisionLog::new(); + assert!(log.input_labels().is_empty()); + log.set_input_labels(vec!["PII".into(), "secret".into()]); + assert_eq!( + log.input_labels(), + &["PII".to_string(), "secret".to_string()] + ); + } + + #[test] + fn input_hash_default_none_and_settable() { + let mut log = DecisionLog::new(); + assert!(log.input_hash().is_none()); + log.set_input_hash(Some("sha256:abc".into())); + assert_eq!(log.input_hash(), Some("sha256:abc")); + } + + #[test] + fn stream_and_sequences_stamp_and_read_back() { + let mut log = DecisionLog::new(); + assert!(log.epoch().is_none()); + assert!(log.stream_id().is_none()); + assert!(log.stream_seq().is_none()); + assert!(log.emission_seq().is_none()); + log.set_stream(1_700_000_000, "decision".into(), 7, 42); + assert_eq!(log.epoch(), Some(1_700_000_000)); + assert_eq!(log.stream_id(), Some("decision")); + assert_eq!(log.stream_seq(), Some(7)); + assert_eq!(log.emission_seq(), Some(42)); + } +} diff --git a/crates/cpex-core/src/effect.rs b/crates/cpex-core/src/effect.rs new file mode 100644 index 00000000..7338b2ed --- /dev/null +++ b/crates/cpex-core/src/effect.rs @@ -0,0 +1,749 @@ +// Location: ./crates/cpex-core/src/effect.rs +// Copyright 2026 +// SPDX-License-Identifier: Apache-2.0 +// Authors: Teryl Taylor +// +// EffectRecord — an irreversible external effect a plugin *causes* (a token +// mint, an approval grant), audited as its own event, distinct from a +// pipeline decision. +// +// Effects follow a small transaction lifecycle: +// +// prepared -> confirmed | rejected | unknown +// +// `prepared` means CPEX has *durably recorded the intent* to attempt the +// action and nothing external has happened yet; the terminal states record +// the outcome, or `unknown` after a crash (resolved later by reconciling +// against the participant — e.g. the IdP — via the record's `key`). This is +// the write-ahead model in docs/step5-effect-audit-options.md (Option C). +// +// This module holds the effect types and lifecycle (`EffectRecord`, +// `EffectState`), the `EffectEmitter` / `DurableEffectLog` traits, and the +// file-backed write-ahead log (`FileEffectLog`). Framework-mediated effect +// primitives (v2) are a later slice. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use async_trait::async_trait; + +use crate::error::PluginError; +use crate::hooks::payload::Extensions; + +/// Where an effect is in its lifecycle. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub enum EffectState { + /// Intent durably recorded; nothing external has happened yet. + Prepared, + /// The external act completed. + Confirmed, + /// The external act provably did not happen. + Rejected, + /// Crashed after acting, before the outcome was recorded. Resolved by + /// reconciling against the participant via `EffectRecord::key`. + Unknown, +} + +/// A record of an irreversible external effect — emitted to audit sinks as +/// its own event. The causing plugin fills the descriptive fields; the +/// framework stamps `plugin_name`. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct EffectRecord { + /// Machine-readable kind, e.g. `"token_mint"`, `"approval_grant"`. + pub kind: String, + /// Human-readable description. + pub description: String, + /// Idempotency / reconciliation key threaded into the external call, so + /// an `unknown` outcome can be resolved against the participant later. + pub key: String, + /// Where in its lifecycle this record is. + pub state: EffectState, + /// Structured, effect-specific detail (audience, scopes, ttl, …). + pub details: HashMap, + /// Which plugin caused the effect. Set by the framework, not self-reported. + pub plugin_name: Option, + /// The executor's boot time (Unix nanoseconds), scoping the sequences so a + /// verifier tells a counter reset (new, larger epoch) from records lost. + /// Ordered, so `(epoch, emission_seq)` totally orders records across + /// restarts. Stamped at emission. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub epoch: Option, + /// The per-type stream this record belongs to (`"effect"`). Stamped at + /// emission. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stream_id: Option, + /// **Completeness** counter — dense within `(epoch, stream_id)`; a gap means + /// an effect record was dropped. Stamped at emission. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stream_seq: Option, + /// **Ordering** counter — monotonic across decisions and effects within the + /// epoch, for interleaved order. Sparse for an effects-only consumer by + /// design; not a loss signal. Stamped at emission. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub emission_seq: Option, +} + +impl EffectRecord { + /// A fresh `prepared` record — the intent, before the act. `key` is the + /// idempotency/reconciliation key that will ride into the external call. + pub fn prepared( + kind: impl Into, + description: impl Into, + key: impl Into, + ) -> Self { + Self { + kind: kind.into(), + description: description.into(), + key: key.into(), + state: EffectState::Prepared, + details: HashMap::new(), + plugin_name: None, + epoch: None, + stream_id: None, + stream_seq: None, + emission_seq: None, + } + } + + /// Attach a structured detail (builder-style). + pub fn with_detail( + mut self, + key: impl Into, + value: impl Into, + ) -> Self { + self.details.insert(key.into(), value.into()); + self + } + + /// Move to a terminal state (`Confirmed` / `Rejected` / `Unknown`). + pub fn into_state(mut self, state: EffectState) -> Self { + self.state = state; + self + } +} + +/// A per-invocation capability handle for emitting effects, granted to +/// plugins with the `emit_effect` capability. It rides on `Extensions` +/// exactly like the write tokens: `filter_extensions` never sets it; the +/// executor does, for capable plugins, right before `handle`. Calling it +/// fans the record out to the auto-attached audit sinks' `on_effect`. +/// +/// Slice 2 emits (write-ahead *ordering*); it is not yet *durable* — the WAL +/// and fail-closed prepare are a later slice. +#[async_trait] +pub trait EffectEmitter: Send + Sync + std::fmt::Debug { + /// Durably record `effect`, then fan it out to the audit sinks. `ext` + /// supplies ambient context (identity, delegation, correlation). + /// + /// **Fail-closed:** an `Err` means the record could not be durably + /// persisted, so the caller must **not** perform the irreversible act. + /// (With no durable log configured, this is ordering-only and returns + /// `Ok` after emitting — slice 3b installs the real WAL.) + async fn emit(&self, effect: &EffectRecord, ext: &Extensions) -> Result<(), Box>; +} + +/// Resolves an effect left `unknown` after a crash by asking an authoritative +/// issuance ledger whether the act identified by `EffectRecord::key` actually +/// happened. The `EffectRecord` is self-describing (kind, key, details), so a +/// reconciler just reads those fields and performs a keyed lookup — it needs no +/// knowledge of which plugin caused the effect. Returns a terminal state, or +/// `Unknown` when the ledger can't say, so the record is retried on a later +/// sweep. +/// +/// Most effects have no queryable ledger (an OAuth IdP, for instance, exposes +/// no lookup by mint key), so [`LogUnknownsReconciler`] is the default. +#[async_trait] +pub trait EffectReconciler: Send + Sync { + async fn reconcile(&self, effect: &EffectRecord) -> EffectState; +} + +/// The default [`EffectReconciler`]: there is no ledger to query, so it logs +/// each unresolved effect and leaves it `unknown` for an operator to +/// investigate. This is the honest, correct behavior for every effect whose +/// participant exposes no keyed lookup — which today is all of them. A real +/// reconciler (e.g. against a mandate server that owns issuance) queries the +/// ledger by `EffectRecord::key`; nothing about it is plugin-specific. +#[derive(Debug, Default)] +pub struct LogUnknownsReconciler; + +#[async_trait] +impl EffectReconciler for LogUnknownsReconciler { + async fn reconcile(&self, effect: &EffectRecord) -> EffectState { + tracing::warn!( + effect_key = %effect.key, + kind = %effect.kind, + plugin = effect.plugin_name.as_deref().unwrap_or("?"), + "effect left unresolved after a crash; no ledger to reconcile it — \ + leaving `unknown` for investigation" + ); + EffectState::Unknown + } +} + +/// A durable, append-only sink for effect records — the write-ahead log. +/// `append` must not return `Ok` until the record is durably persisted; an +/// `Err` means the caller must **not** perform the irreversible act +/// (fail-closed). `FileEffectLog` is the file-backed implementation. +#[async_trait] +pub trait DurableEffectLog: Send + Sync { + async fn append(&self, effect: &EffectRecord) -> Result<(), Box>; + + /// Recover after a restart: compact completed effects and reconcile the + /// unresolved ones against `reconciler`, recording each confirmed/rejected + /// outcome durably. Returns the effects still `unknown` (the participant + /// couldn't say) for a later sweep. Default: a no-op — for logs with no + /// recoverable on-disk state. + async fn recover_and_reconcile( + &self, + _reconciler: &dyn EffectReconciler, + ) -> Result, Box> { + Ok(Vec::new()) + } +} + +/// A file-backed, append-only write-ahead log for effect records — the +/// durable sink behind `ext.begin_effect`. Each record is appended as one +/// JSON line and `fsync`'d before `append` returns, so a `prepared` intent is +/// on stable storage *before* the irreversible act. `append` returns `Err` +/// (fail-closed) whenever the record cannot be durably persisted, which is +/// what stops the act from proceeding. +/// +/// The append + `fsync` run on a blocking thread (`spawn_blocking`): tokio's +/// `fs` feature is not enabled, and a synchronous `fsync` must never stall an +/// async worker. Concurrent appends are safe — `O_APPEND` makes each write +/// land atomically at the end of the file. v1 opens the file per append; +/// effects are rare (token mints, approval grants), so the open cost is not a +/// hot path, and a pooled handle can be a later optimization. +/// Default number of appends between automatic compactions. Effects are rare, +/// so this bounds the file to roughly this many records between compactions +/// without paying a rewrite on every write. +const DEFAULT_COMPACTION_THRESHOLD: usize = 1024; + +#[derive(Debug, Clone)] +pub struct FileEffectLog { + path: Arc, + /// Serializes appends. `O_APPEND` already makes each write's *positioning* + /// atomic, but `write_all`'s partial-write loop leaves a narrow window + /// where two concurrent writers could interleave a record. One writer at a + /// time closes it and gives a deterministic on-disk order (what the + /// recovery sweep reads back). Effects are rare, so contention is + /// negligible. Cloned handles share the lock, since they share the file. + write_lock: Arc>, + /// Appends since the last compaction. Shared across cloned handles (they + /// share the file). When it crosses `compaction_threshold`, `append` + /// triggers a compact-only `recover()` to bound the file. + appends_since_compaction: Arc, + /// Auto-compaction fires after this many appends. `0` disables it — + /// compaction then happens only on an explicit `recover()`. + compaction_threshold: usize, +} + +impl FileEffectLog { + /// A WAL that appends to `path`, creating the file if it does not exist. + pub fn new(path: impl Into) -> Self { + Self { + path: Arc::new(path.into()), + write_lock: Arc::new(tokio::sync::Mutex::new(())), + appends_since_compaction: Arc::new(AtomicUsize::new(0)), + compaction_threshold: DEFAULT_COMPACTION_THRESHOLD, + } + } + + /// Override the append count that triggers automatic compaction. `0` + /// disables auto-compaction (compaction then happens only on an explicit + /// `recover()`). + pub fn with_compaction_threshold(mut self, threshold: usize) -> Self { + self.compaction_threshold = threshold; + self + } +} + +#[async_trait] +impl DurableEffectLog for FileEffectLog { + async fn append(&self, effect: &EffectRecord) -> Result<(), Box> { + // Serialize on the async thread (cheap, no I/O); do the blocking + // append + fsync off the async worker pool. + let mut line = serde_json::to_vec(effect) + .map_err(|e| wal_error("serialize effect record", Some(Box::new(e))))?; + line.push(b'\n'); + + let path = Arc::clone(&self.path); + // Write under the lock, then release it *before* any compaction: + // `recover()` re-acquires this same lock, so holding it here would + // deadlock. Serialized so records never interleave. + { + let _guard = self.write_lock.lock().await; + tokio::task::spawn_blocking(move || -> Result<(), Box> { + use std::io::Write as _; + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path.as_ref()) + .map_err(|e| wal_error("open WAL", Some(Box::new(e))))?; + file.write_all(&line) + .map_err(|e| wal_error("write WAL record", Some(Box::new(e))))?; + // The durability barrier: the record is on stable storage + // before this returns Ok — and therefore before the caller acts. + file.sync_all() + .map_err(|e| wal_error("fsync WAL", Some(Box::new(e))))?; + Ok(()) + }) + .await + .map_err(|e| wal_error("WAL append task failed", Some(Box::new(e))))??; + } + + // Auto-compaction: once appends since the last compaction cross the + // threshold, run compact-only `recover()` — safe at runtime, since it + // keeps in-flight `prepared` records and only drops matched pairs — to + // bound the file. Threshold 0 disables it. A compaction failure is + // logged, never fatal: the record above is already durable, so the + // append itself succeeded and must not report fail-closed. + if self.compaction_threshold != 0 { + let n = self + .appends_since_compaction + .fetch_add(1, Ordering::Relaxed) + + 1; + if n >= self.compaction_threshold { + self.appends_since_compaction.store(0, Ordering::Relaxed); + if let Err(e) = self.recover().await { + tracing::warn!("effect WAL auto-compaction failed: {e}"); + } + } + } + + Ok(()) + } + + async fn recover_and_reconcile( + &self, + reconciler: &dyn EffectReconciler, + ) -> Result, Box> { + // Compact completed effects, then ask the participant about each + // survivor. A confirmed/rejected answer is recorded so the following + // compaction drops the pair; an `unknown` answer is left for later. + let summary = self.recover().await?; + let mut resolved_any = false; + let mut still_unknown = Vec::new(); + for rec in summary.unresolved { + match reconciler.reconcile(&rec).await { + state @ (EffectState::Confirmed | EffectState::Rejected) => { + self.append(&rec.clone().into_state(state)).await?; + resolved_any = true; + }, + // Still `unknown` (or `prepared`) — keep it for the next sweep. + _ => still_unknown.push(rec), + } + } + if resolved_any { + // A second pass compacts the just-resolved matched pairs out. + self.recover().await?; + } + Ok(still_unknown) + } +} + +/// Build the `PluginError` a durable-write failure surfaces. `begin_effect` +/// treats any `Err` from the WAL as fail-closed, so this is the error that +/// prevents an irreversible act from proceeding. +fn wal_error( + what: &str, + source: Option>, +) -> Box { + PluginError::Execution { + plugin_name: "effect-wal".into(), + message: format!("effect WAL: {what}"), + source, + code: Some("effect_wal_failed".into()), + details: HashMap::new(), + proto_error_code: None, + } + .boxed() +} + +/// The result of a recovery sweep over a [`FileEffectLog`]. +#[derive(Debug, Default)] +pub struct RecoverySummary { + /// Number of effects that completed — a `prepared` matched by a terminal + /// (`confirmed`/`rejected`) record — and were compacted out of the log. + pub compacted: usize, + /// Effects with no terminal record: `prepared`-without-outcome (an act that + /// may or may not have happened before a crash) or an explicit `unknown`. + /// Each needs reconciliation against the participant (the IdP) via its + /// `key`. They are retained in the rewritten log. + pub unresolved: Vec, +} + +impl FileEffectLog { + /// Recover after a restart: read the WAL, drop completed effects (a + /// `prepared` matched by a terminal record), and atomically rewrite the + /// file with only the unresolved records — `prepared`-without-terminal or + /// `unknown`. Returns those unresolved records so the caller can reconcile + /// them against the participant (the IdP) via each record's `key`. This is + /// the compaction that bounds WAL growth (design §6.1) and the entry point + /// for crash recovery. Idempotent; a missing file is a no-op. + pub async fn recover(&self) -> Result> { + let path = Arc::clone(&self.path); + // Serialize against appends while we read + rewrite the log. + let _guard = self.write_lock.lock().await; + tokio::task::spawn_blocking(move || -> Result> { + // A missing log means nothing to recover. + let data = match std::fs::read_to_string(path.as_ref()) { + Ok(d) => d, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return Ok(RecoverySummary::default()) + }, + Err(e) => return Err(wal_error("read WAL for recovery", Some(Box::new(e)))), + }; + + // Parse every record in append order. + let mut records: Vec = Vec::new(); + for (i, line) in data.lines().enumerate() { + if line.trim().is_empty() { + continue; + } + let rec = serde_json::from_str(line).map_err(|e| { + wal_error(&format!("parse WAL line {}", i + 1), Some(Box::new(e))) + })?; + records.push(rec); + } + + // A key is resolved iff some record for it reached a terminal + // state. Everything else (prepared-only, unknown) is unresolved. + let resolved: std::collections::HashSet<&str> = records + .iter() + .filter(|r| matches!(r.state, EffectState::Confirmed | EffectState::Rejected)) + .map(|r| r.key.as_str()) + .collect(); + + // Keep the latest record per unresolved key, in first-seen order. + let mut unresolved: Vec = Vec::new(); + let mut pos: std::collections::HashMap = + std::collections::HashMap::new(); + for r in &records { + if resolved.contains(r.key.as_str()) { + continue; + } + match pos.get(&r.key) { + Some(&i) => unresolved[i] = r.clone(), + None => { + pos.insert(r.key.clone(), unresolved.len()); + unresolved.push(r.clone()); + }, + } + } + let compacted = resolved.len(); + + // Atomic rewrite: write the survivors to a temp file, fsync, then + // rename over the original. rename is atomic on POSIX, so a crash + // mid-compaction leaves either the old log or the new one — never + // a truncated one. + let tmp = path.with_extension("recover.tmp"); + { + use std::io::Write as _; + let mut f = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&tmp) + .map_err(|e| wal_error("open WAL temp", Some(Box::new(e))))?; + for r in &unresolved { + let mut line = serde_json::to_vec(r) + .map_err(|e| wal_error("serialize during compaction", Some(Box::new(e))))?; + line.push(b'\n'); + f.write_all(&line) + .map_err(|e| wal_error("write WAL temp", Some(Box::new(e))))?; + } + f.sync_all() + .map_err(|e| wal_error("fsync WAL temp", Some(Box::new(e))))?; + } + std::fs::rename(&tmp, path.as_ref()) + .map_err(|e| wal_error("rename WAL temp", Some(Box::new(e))))?; + + Ok(RecoverySummary { + compacted, + unresolved, + }) + }) + .await + .map_err(|e| wal_error("WAL recovery task failed", Some(Box::new(e))))? + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn prepared_starts_in_prepared_with_details() { + let e = EffectRecord::prepared("token_mint", "exchange for workday-api", "k-1") + .with_detail("audience", "workday-api") + .with_detail("scopes", json!(["read_compensation"])); + + assert_eq!(e.kind, "token_mint"); + assert_eq!(e.key, "k-1"); + assert_eq!(e.state, EffectState::Prepared); + assert_eq!(e.details["audience"], "workday-api"); + assert_eq!(e.details["scopes"][0], "read_compensation"); + assert!(e.plugin_name.is_none()); + } + + #[test] + fn into_state_transitions_to_terminal() { + let e = EffectRecord::prepared("token_mint", "…", "k-2").into_state(EffectState::Confirmed); + assert_eq!(e.state, EffectState::Confirmed); + } + + /// A temp path unique per (process, call) so parallel tests don't collide, + /// without pulling in a `tempfile` dev-dependency. + fn unique_temp_path(tag: &str) -> std::path::PathBuf { + use std::sync::atomic::{AtomicU64, Ordering}; + static COUNTER: AtomicU64 = AtomicU64::new(0); + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!("cpex_{tag}_{}_{n}.ndjson", std::process::id())) + } + + #[tokio::test] + async fn file_log_appends_one_json_line_per_record() { + let path = unique_temp_path("append"); + let log = FileEffectLog::new(&path); + + let e1 = EffectRecord::prepared("token_mint", "mint A", "k-1") + .with_detail("audience", "workday-api"); + let e2 = EffectRecord::prepared("approval_grant", "grant B", "k-2"); + log.append(&e1).await.expect("append e1"); + log.append(&e2).await.expect("append e2"); + + let contents = std::fs::read_to_string(&path).expect("read WAL"); + let lines: Vec<&str> = contents.lines().collect(); + assert_eq!(lines.len(), 2, "one line per appended record"); + + // Each line round-trips back into an EffectRecord (this is what the + // 3b-iii recovery sweep will rely on). + let r1: EffectRecord = serde_json::from_str(lines[0]).expect("parse line 1"); + assert_eq!(r1.kind, "token_mint"); + assert_eq!(r1.key, "k-1"); + assert_eq!(r1.state, EffectState::Prepared); + assert_eq!(r1.details["audience"], "workday-api"); + + let r2: EffectRecord = serde_json::from_str(lines[1]).expect("parse line 2"); + assert_eq!(r2.kind, "approval_grant"); + + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn file_log_persists_across_reopen() { + // A fresh log over the same path sees prior records — the WAL is real + // on-disk state, not per-instance memory. Recovery depends on this. + let path = unique_temp_path("reopen"); + FileEffectLog::new(&path) + .append(&EffectRecord::prepared("token_mint", "x", "k-a")) + .await + .unwrap(); + FileEffectLog::new(&path) + .append(&EffectRecord::prepared("token_mint", "y", "k-b")) + .await + .unwrap(); + + let contents = std::fs::read_to_string(&path).unwrap(); + assert_eq!( + contents.lines().count(), + 2, + "second instance appends, does not truncate" + ); + + let _ = std::fs::remove_file(&path); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn file_log_concurrent_appends_stay_intact() { + // N tasks append to one shared log across multiple worker threads. The + // serialization lock must yield exactly N whole, parseable records — + // no split or interleaved lines. + let path = unique_temp_path("concurrent"); + let log = FileEffectLog::new(&path); + const N: usize = 64; + + let mut handles = Vec::with_capacity(N); + for i in 0..N { + let log = log.clone(); + handles.push(tokio::spawn(async move { + let e = EffectRecord::prepared("token_mint", format!("mint {i}"), format!("k-{i}")); + log.append(&e).await.unwrap(); + })); + } + for h in handles { + h.await.unwrap(); + } + + let contents = std::fs::read_to_string(&path).expect("read WAL"); + let lines: Vec<&str> = contents.lines().collect(); + assert_eq!(lines.len(), N, "one line per concurrent append"); + + // Every line is a complete record, and all N distinct keys are present + // (nothing was corrupted or lost). + let mut keys = std::collections::HashSet::new(); + for line in lines { + let rec: EffectRecord = + serde_json::from_str(line).expect("each line is one intact record"); + keys.insert(rec.key); + } + assert_eq!(keys.len(), N, "all N records present and distinct"); + + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn recover_compacts_completed_and_keeps_orphans() { + let path = unique_temp_path("recover"); + let log = FileEffectLog::new(&path); + + // Completed: prepared + confirmed (same key). + let done = EffectRecord::prepared("token_mint", "done", "k-done"); + log.append(&done).await.unwrap(); + log.append(&done.clone().into_state(EffectState::Confirmed)) + .await + .unwrap(); + // Orphan: prepared with no terminal (the crash case). + log.append(&EffectRecord::prepared("token_mint", "orphan", "k-orphan")) + .await + .unwrap(); + // Completed: prepared + rejected. + let rej = EffectRecord::prepared("approval_grant", "rej", "k-rej"); + log.append(&rej).await.unwrap(); + log.append(&rej.clone().into_state(EffectState::Rejected)) + .await + .unwrap(); + + let summary = log.recover().await.unwrap(); + assert_eq!( + summary.compacted, 2, + "confirmed + rejected effects compacted out" + ); + assert_eq!(summary.unresolved.len(), 1, "only the orphan is unresolved"); + assert_eq!(summary.unresolved[0].key, "k-orphan"); + + // The rewritten log holds only the orphan. + let contents = std::fs::read_to_string(&path).unwrap(); + assert_eq!(contents.lines().count(), 1); + let rec: EffectRecord = serde_json::from_str(contents.lines().next().unwrap()).unwrap(); + assert_eq!(rec.key, "k-orphan"); + + // Idempotent: a second sweep with no new terminals keeps the orphan. + let again = log.recover().await.unwrap(); + assert_eq!(again.compacted, 0); + assert_eq!(again.unresolved.len(), 1); + + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn recover_on_missing_file_is_noop() { + let path = unique_temp_path("recover_missing"); + let _ = std::fs::remove_file(&path); + let summary = FileEffectLog::new(&path).recover().await.unwrap(); + assert_eq!(summary.compacted, 0); + assert!(summary.unresolved.is_empty()); + } + + #[tokio::test] + async fn recover_and_reconcile_resolves_via_participant() { + // A stand-in IdP: confirms one key, rejects another, can't resolve a third. + struct MockIdp; + #[async_trait] + impl EffectReconciler for MockIdp { + async fn reconcile(&self, effect: &EffectRecord) -> EffectState { + match effect.key.as_str() { + "k-confirm" => EffectState::Confirmed, + "k-reject" => EffectState::Rejected, + _ => EffectState::Unknown, + } + } + } + + let path = unique_temp_path("reconcile"); + let log = FileEffectLog::new(&path); + for key in ["k-confirm", "k-reject", "k-unknown"] { + log.append(&EffectRecord::prepared("token_mint", "orphan", key)) + .await + .unwrap(); + } + + let still = log.recover_and_reconcile(&MockIdp).await.unwrap(); + assert_eq!(still.len(), 1, "only the un-resolvable effect remains"); + assert_eq!(still[0].key, "k-unknown"); + + // The WAL now holds only the still-unknown record; the resolved pair + // for k-confirm and k-reject was compacted out. + let contents = std::fs::read_to_string(&path).unwrap(); + let lines: Vec<&str> = contents.lines().collect(); + assert_eq!(lines.len(), 1); + let rec: EffectRecord = serde_json::from_str(lines[0]).unwrap(); + assert_eq!(rec.key, "k-unknown"); + + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn auto_compaction_bounds_the_wal() { + let path = unique_temp_path("autocompact"); + // Low threshold so completed effects trigger compaction quickly. + let log = FileEffectLog::new(&path).with_compaction_threshold(4); + + // 20 completed effects = 40 appends. Without compaction the file would + // grow to 40 lines; auto-compaction drops each matched pair, so it + // stays bounded near the number of in-flight (here: zero) records. + for i in 0..20 { + let e = EffectRecord::prepared("token_mint", "x", format!("k-{i}")); + log.append(&e).await.unwrap(); + log.append(&e.clone().into_state(EffectState::Confirmed)) + .await + .unwrap(); + } + + let contents = std::fs::read_to_string(&path).unwrap_or_default(); + let lines = contents.lines().count(); + assert!( + lines < 8, + "auto-compaction bounded the WAL (got {lines} lines, not 40)" + ); + + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn compaction_threshold_zero_disables_auto_compaction() { + let path = unique_temp_path("nocompact"); + let log = FileEffectLog::new(&path).with_compaction_threshold(0); + + // Two completed effects (4 appends). With auto-compaction off, every + // line stays — nothing is compacted until an explicit recover(). + for i in 0..2 { + let e = EffectRecord::prepared("token_mint", "x", format!("k-{i}")); + log.append(&e).await.unwrap(); + log.append(&e.clone().into_state(EffectState::Confirmed)) + .await + .unwrap(); + } + + let contents = std::fs::read_to_string(&path).unwrap(); + assert_eq!( + contents.lines().count(), + 4, + "no auto-compaction when threshold is 0" + ); + + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn default_reconciler_leaves_effects_unknown() { + let effect = EffectRecord::prepared("token_mint", "no ledger", "k-x"); + assert_eq!( + LogUnknownsReconciler.reconcile(&effect).await, + EffectState::Unknown + ); + } +} diff --git a/crates/cpex-core/src/executor.rs b/crates/cpex-core/src/executor.rs index ed4382cf..a63e4ff0 100644 --- a/crates/cpex-core/src/executor.rs +++ b/crates/cpex-core/src/executor.rs @@ -26,13 +26,18 @@ use std::any::Any; use std::fmt; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::Duration; +use async_trait::async_trait; use tokio::time::timeout; use tracing::{error, warn}; +use crate::audit::AuditHandler; use crate::context::PluginContextTable; +use crate::decision::{DecisionLog, PluginAction, Verdict}; +use crate::effect::{DurableEffectLog, EffectEmitter, EffectRecord}; use crate::error::PluginError; use crate::extensions::filter_extensions; use crate::hooks::payload::{Extensions, PluginPayload, WriteToken}; @@ -47,6 +52,11 @@ pub struct ExecutorConfig { /// Whether to halt on the first deny in concurrent mode. pub short_circuit_on_deny: bool, + + /// Hash the payload at pipeline entry for audit content provenance + /// (`DecisionLog::input_hash`). Off by default — hashing is on the request + /// path, so it is opt-in. + pub capture_content_provenance: bool, } impl Default for ExecutorConfig { @@ -54,6 +64,7 @@ impl Default for ExecutorConfig { Self { timeout_seconds: 30, short_circuit_on_deny: true, + capture_content_provenance: false, } } } @@ -132,6 +143,11 @@ pub struct PipelineResult { /// Plugin contexts indexed by plugin ID. Thread this into the /// next hook invocation to preserve per-plugin `local_state`. pub context_table: PluginContextTable, + + /// The executor's record of what each plugin did and how the pipeline + /// ruled. Built executor-side and handed to audit sinks; never exposed + /// to plugins through `PluginContext`. + pub decision_log: DecisionLog, } impl PipelineResult { @@ -150,6 +166,7 @@ impl PipelineResult { errors: Vec::new(), metadata: None, context_table, + decision_log: DecisionLog::new(), } } @@ -176,6 +193,7 @@ impl PipelineResult { errors: Vec::new(), metadata: None, context_table, + decision_log: DecisionLog::new(), } } @@ -187,6 +205,12 @@ impl PipelineResult { self } + /// Attach the executor's decision log to a constructed result. + pub fn with_decision_log(mut self, decision_log: DecisionLog) -> Self { + self.decision_log = decision_log; + self + } + /// Whether this result represents a denial. pub fn is_denied(&self) -> bool { !self.continue_processing @@ -269,17 +293,148 @@ impl fmt::Debug for BackgroundTasks { /// SEQUENTIAL → TRANSFORM → AUDIT → CONCURRENT → FIRE_AND_FORGET /// ``` /// -/// The executor is stateless — all state comes from the arguments. -/// One executor instance can serve multiple concurrent hook invocations. +/// The executor's only state is its config and the auto-attached audit +/// sinks; all per-request state comes from the arguments. One executor +/// instance can serve multiple concurrent hook invocations. #[derive(Clone)] pub struct Executor { config: ExecutorConfig, + + /// Observation-only sinks invoked at the verdict of every pipeline run. + /// Set when the manager builds the runtime snapshot; empty otherwise. + /// They receive the decision log but cannot influence the outcome. + audit_handlers: Vec>, + + /// Durable write-ahead log for irreversible effects. `None` (default) = + /// ordering-only: effects still emit to the audit sinks, but + /// `begin_effect` is not crash-safe or fail-closed. Installed from + /// `plugin_settings.effect_log_path` or programmatically. Opt-in. + effect_log: Option>, + + /// Audit stream identity + counters. `epoch` is the executor's boot time + /// (Unix nanos), captured once — it scopes the counters so a restart is + /// distinguishable from a loss and orders records across restarts. Each + /// record carries its per-type counter (`decision_seq` / `effect_seq`, + /// gap-free → completeness) and the shared `emission_seq` (global across + /// both → interleaved order). The counters are `Arc` so copy-on-write + /// snapshot mutations stay on the same stream. + epoch: u64, + decision_seq: Arc, + effect_seq: Arc, + emission_seq: Arc, } impl Executor { /// Create a new executor with the given configuration. pub fn new(config: ExecutorConfig) -> Self { - Self { config } + Self { + config, + audit_handlers: Vec::new(), + effect_log: None, + // Boot time in Unix nanoseconds — an orderable epoch that needs no + // persistence. A new executor (restart or config reload) gets a + // larger value, so a verifier tells a reset from a loss. + epoch: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0), + decision_seq: Arc::new(AtomicU64::new(0)), + effect_seq: Arc::new(AtomicU64::new(0)), + emission_seq: Arc::new(AtomicU64::new(0)), + } + } + + /// Install a durable effect log (WAL). When present, `begin_effect` is + /// crash-safe and fail-closed; when absent, effect auditing is + /// ordering-only. Builder form, used when constructing from config. + pub fn with_effect_log(mut self, effect_log: Arc) -> Self { + self.effect_log = Some(effect_log); + self + } + + /// Install a durable effect log via copy-on-write snapshot mutation — the + /// manager's programmatic path, mirroring [`Self::push_audit_handler`]. + pub fn set_effect_log(&mut self, effect_log: Arc) { + self.effect_log = Some(effect_log); + } + + /// The installed durable effect log, if any — for the host to run + /// crash recovery at startup (see `PluginManager::recover_effects`). + pub fn effect_log(&self) -> Option> { + self.effect_log.clone() + } + + /// Attach observation-only audit sinks, invoked at the verdict of every + /// pipeline run. Used by the manager when it builds the runtime snapshot. + pub fn with_audit_handlers(mut self, audit_handlers: Vec>) -> Self { + self.audit_handlers = audit_handlers; + self + } + + /// Append a single audit sink. Used by the manager's + /// `register_audit_handler` through copy-on-write snapshot mutation. + pub fn push_audit_handler(&mut self, handler: Arc) { + self.audit_handlers.push(handler); + } + + /// Invoke every audit sink with the finalized decision, once per pipeline + /// run. Observation-only — the executor ignores whatever they return. + /// Assign this decision's stream identity + sequence numbers. The executor + /// writes its **own** record here — a step distinct from the read-only + /// handoff in [`Self::emit_audit`] (which takes `&DecisionLog`), so a sink + /// never receives anything mutable. `decision_seq` is gap-free within the + /// decision stream (completeness); `emission_seq` is the shared global + /// counter across decisions and effects (interleaved order). Stamped even + /// with no sinks — it's a property of the stream and rides on + /// `PipelineResult.decision_log`. + fn stamp_decision_stream(&self, decisions: &mut DecisionLog) { + decisions.set_stream( + self.epoch, + "decision".to_string(), + self.decision_seq.fetch_add(1, Ordering::Relaxed), + self.emission_seq.fetch_add(1, Ordering::Relaxed), + ); + } + + async fn emit_audit( + &self, + payload: &dyn PluginPayload, + extensions: &Extensions, + decisions: &DecisionLog, + ) { + use futures::FutureExt; + use std::panic::AssertUnwindSafe; + + if self.audit_handlers.is_empty() { + return; + } + + // Observation-only: an audit sink must never crash or hang the + // request whose verdict is already decided. Contain panics and bound + // each call; log loudly and move on. A lost audit record is itself a + // problem (see the durability plan) but that never justifies letting a + // sink take down the request. + let timeout_dur = Duration::from_secs(self.config.timeout_seconds); + for handler in &self.audit_handlers { + let call = + AssertUnwindSafe(handler.handle(payload, extensions, decisions)).catch_unwind(); + match timeout(timeout_dur, call).await { + Ok(Ok(())) => {}, + Ok(Err(_panic)) => { + error!( + "audit sink '{}' panicked during emit — contained", + handler.name() + ); + }, + Err(_elapsed) => { + error!( + "audit sink '{}' exceeded {}s during emit — skipped", + handler.name(), + timeout_dur.as_secs() + ); + }, + } + } } /// Execute a hook invocation through the 5-phase pipeline. @@ -333,6 +488,36 @@ impl Executor { // read an exact signal instead of comparing payload contents. let mut payload_modified = false; + // The executor's private record of what each plugin did and how the + // pipeline ruled. Threaded through the phases, finalized at each + // return point, and attached to the result for audit sinks. + let mut decisions = DecisionLog::new(); + // This interception's node identity in the decision graph: a fresh + // span whose parent is the request's span (the upstream call that + // triggered us), within the request's trace (child-span model). + let request = current_extensions.request.as_ref(); + decisions.set_span(crate::decision::Span::for_request( + request.and_then(|r| r.trace_id.as_deref()), + request.and_then(|r| r.span_id.as_deref()), + )); + // Capture the taint the request arrived with — the input side of this + // node's provenance. A sink diffs it against the final labels to see + // what the pipeline added. Sorted so the record is deterministic. + if let Some(sec) = current_extensions.security.as_ref() { + let mut labels: Vec = sec.labels.iter().cloned().collect(); + labels.sort_unstable(); + decisions.set_input_labels(labels); + } + // Content-addressed input provenance — the payload's hash at entry, + // before any plugin mutates it. Opt-in (hashing is on the request + // path); only the digest is kept, never the bytes. + if self.config.capture_content_provenance { + let hash = current_payload + .audit_bytes() + .map(|b| crate::hooks::payload::content_hash(&b)); + decisions.set_input_hash(hash); + } + if let Some(v) = self .run_serial_phase( &sequential, @@ -343,12 +528,19 @@ impl Executor { true, // can_modify "SEQUENTIAL", &mut errors, + &mut decisions, &mut payload_modified, ) .await { + decisions.finalize(Verdict::Deny(v.clone())); + self.stamp_decision_stream(&mut decisions); + self.emit_audit(&*current_payload, ¤t_extensions, &decisions) + .await; return ( - PipelineResult::denied(v, current_extensions, ctx_table).with_errors(errors), + PipelineResult::denied(v, current_extensions, ctx_table) + .with_errors(errors) + .with_decision_log(decisions), BackgroundTasks::empty(), ); } @@ -364,6 +556,7 @@ impl Executor { true, // can_modify "TRANSFORM", &mut errors, + &mut decisions, &mut payload_modified, ) .await; @@ -385,12 +578,18 @@ impl Executor { ¤t_extensions, &ctx_table, &mut errors, + &mut decisions, ) .await { + decisions.finalize(Verdict::Deny(violation.clone())); + self.stamp_decision_stream(&mut decisions); + self.emit_audit(&*current_payload, ¤t_extensions, &decisions) + .await; return ( PipelineResult::denied(violation, current_extensions, ctx_table) - .with_errors(errors), + .with_errors(errors) + .with_decision_log(decisions), BackgroundTasks::empty(), ); } @@ -406,9 +605,14 @@ impl Executor { task_tracker, ); + decisions.finalize(Verdict::Allow); + self.stamp_decision_stream(&mut decisions); + self.emit_audit(&*current_payload, ¤t_extensions, &decisions) + .await; ( PipelineResult::allowed_with(current_payload, current_extensions, ctx_table) .with_errors(errors) + .with_decision_log(decisions) .with_payload_modified(payload_modified), BackgroundTasks::from_handles(bg_handles), ) @@ -440,6 +644,7 @@ impl Executor { can_modify: bool, phase_label: &str, errors: &mut Vec, + decisions: &mut DecisionLog, payload_modified: &mut bool, ) -> Option { for entry in entries { @@ -450,6 +655,11 @@ impl Executor { let plugin_name = entry.plugin_ref.name(); let plugin_id = entry.plugin_ref.id(); let on_error = entry.plugin_ref.trusted_config().on_error; + let phase = entry.plugin_ref.trusted_config().mode; + // What this plugin did, recorded after it runs (or inline before a + // halting return). Defaults to Allowed; the modify and error paths + // update it. + let mut action = PluginAction::Allowed; // Take this plugin's context out of the table — pulls its stored // local_state and seeds global_state from the canonical store. @@ -479,6 +689,21 @@ impl Executor { if capabilities.contains("append_delegation") { filtered.delegation_write_token = Some(WriteToken::new()); } + // Grant the effect-emit capability the same way — a per-invoke + // handle on the filtered extensions, only for capable plugins. + if capabilities.contains("emit_effect") { + filtered.effect_emitter = Some(Arc::new(AuditEffectEmitter { + handlers: self.audit_handlers.clone(), + plugin_name: plugin_name.to_string(), + timeout: Duration::from_secs(self.config.timeout_seconds), + // The configured WAL (opt-in). `None` → ordering-only, not + // fail-closed; `Some` → durable-before-fanout, fail-closed. + durable: self.effect_log.clone(), + epoch: self.epoch, + stream_seq: self.effect_seq.clone(), + emission_seq: self.emission_seq.clone(), + })); + } // Execute with timeout — handler borrows payload, gets filtered extensions let timeout_dur = Duration::from_secs(self.config.timeout_seconds); @@ -494,6 +719,7 @@ impl Executor { if !erased.continue_processing && can_block { if let Some(mut v) = erased.violation { v.plugin_name = Some(plugin_name.to_string()); + decisions.record(plugin_name, phase, PluginAction::Denied); return Some(v); } } @@ -502,6 +728,7 @@ impl Executor { if can_modify { if let Some(mp) = erased.modified_payload { *payload = mp; + action = PluginAction::ModifiedPayload; *payload_modified = true; } if let Some(mut owned) = erased.modified_extensions { @@ -592,6 +819,9 @@ impl Executor { ); } else { extensions.merge_owned(owned); + if action == PluginAction::Allowed { + action = PluginAction::ModifiedExtensions; + } } } } @@ -603,6 +833,7 @@ impl Executor { }, Ok(Err(e)) => { error!("{} plugin '{}' failed: {}", phase_label, plugin_name, e); + action = PluginAction::Error(e.to_string()); match on_error { OnError::Fail if can_block => { let mut v = crate::error::PluginViolation::new( @@ -610,6 +841,7 @@ impl Executor { format!("Plugin '{}' failed: {}", plugin_name, e), ); v.plugin_name = Some(plugin_name.to_string()); + decisions.record(plugin_name, phase, action.clone()); return Some(v); }, // Any non-halt outcome (Fail-in-non-blocking-phase, @@ -638,6 +870,7 @@ impl Executor { }, Err(_) => { error!("{} plugin '{}' timed out", phase_label, plugin_name); + action = PluginAction::Error("timed out".to_string()); let timeout_err = crate::error::PluginError::Timeout { plugin_name: plugin_name.to_string(), timeout_ms: timeout_dur.as_millis() as u64, @@ -650,6 +883,7 @@ impl Executor { format!("Plugin '{}' timed out", plugin_name), ); v.plugin_name = Some(plugin_name.to_string()); + decisions.record(plugin_name, phase, action.clone()); return Some(v); }, OnError::Fail => { @@ -674,6 +908,10 @@ impl Executor { }, } + // Record what this plugin did (halting paths recorded inline above + // and returned before reaching here). + decisions.record(plugin_name, phase, action); + // Commit this plugin's context back to the table — replaces the // canonical global_state with its (possibly modified) copy and // stores the local_state for the next hook invocation. The @@ -785,6 +1023,7 @@ impl Executor { extensions: &Extensions, ctx_table: &PluginContextTable, errors: &mut Vec, + decisions: &mut DecisionLog, ) -> Option { use cpex_orchestration::{run_branches, BranchConfig, BranchOutcome, ErasedBranch}; @@ -893,6 +1132,20 @@ impl Executor { let plugin_name = entry.plugin_ref.name(); let on_error = on_error_by_idx[idx]; + // Record what this concurrent plugin did, in input order. + let action = match &outcome { + BranchOutcome::Completed(BranchData::Allow) => PluginAction::Allowed, + BranchOutcome::Completed(BranchData::Deny(_)) => PluginAction::Denied, + BranchOutcome::Completed(BranchData::Error(e)) => { + PluginAction::Error(e.to_string()) + }, + BranchOutcome::TimedOut => PluginAction::Error("timed out".to_string()), + BranchOutcome::Panicked(s) => PluginAction::Error(format!("panicked: {s}")), + // Cancelled because another branch short-circuited the phase. + BranchOutcome::Aborted => PluginAction::Error("aborted".to_string()), + }; + decisions.record(plugin_name, entry.plugin_ref.trusted_config().mode, action); + match outcome { BranchOutcome::Completed(BranchData::Allow) => {}, BranchOutcome::Completed(BranchData::Deny(opt_v)) => { @@ -1086,6 +1339,82 @@ impl Default for Executor { // SerialResult removed — run_serial_phase now returns Option directly. +/// Effect emitter the executor grants to `emit_effect`-capable plugins via +/// `Extensions.effect_emitter`. Fans an effect record out to the audit sinks' +/// `on_effect`, isolated (timeout + catch_unwind) exactly like the verdict +/// emit, and stamps the causing plugin (not self-reported). +struct AuditEffectEmitter { + handlers: Vec>, + plugin_name: String, + timeout: Duration, + /// Write-ahead log. When present, `emit` durably records the effect + /// before fanning out and fails closed if that write fails. `None` until + /// slice 3b wires a real WAL — then emit is ordering-only. + durable: Option>, + /// Boot epoch + counters (shared with the executor). Each emitted record is + /// stamped with `epoch`, `stream_seq` (gap-free within the effect stream), + /// and the global `emission_seq` (interleaved order vs decisions). + epoch: u64, + stream_seq: Arc, + emission_seq: Arc, +} + +impl std::fmt::Debug for AuditEffectEmitter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AuditEffectEmitter") + .field("plugin_name", &self.plugin_name) + .field("sinks", &self.handlers.len()) + .finish() + } +} + +#[async_trait] +impl EffectEmitter for AuditEffectEmitter { + async fn emit(&self, effect: &EffectRecord, ext: &Extensions) -> Result<(), Box> { + use futures::FutureExt; + use std::panic::AssertUnwindSafe; + + // Stamp the causing plugin + stream identity/sequences — all set by the + // framework, not self-reported. `stream_seq` is gap-free within the + // effect stream (completeness); `emission_seq` is the shared global + // counter across decisions and effects (interleaved order). + let mut stamped = effect.clone(); + stamped.plugin_name = Some(self.plugin_name.clone()); + stamped.epoch = Some(self.epoch); + stamped.stream_id = Some("effect".to_string()); + stamped.stream_seq = Some(self.stream_seq.fetch_add(1, Ordering::Relaxed)); + stamped.emission_seq = Some(self.emission_seq.fetch_add(1, Ordering::Relaxed)); + + // Write-ahead: durably record BEFORE any observer sees it. Fail + // closed — if the durable write fails, return Err and do NOT fan out; + // the caller must not perform the act. + if let Some(log) = &self.durable { + log.append(&stamped).await?; + } + + for handler in &self.handlers { + let call = AssertUnwindSafe(handler.on_effect(&stamped, ext)).catch_unwind(); + match timeout(self.timeout, call).await { + Ok(Ok(())) => {}, + Ok(Err(_panic)) => { + error!( + "audit sink '{}' panicked during on_effect — contained", + handler.name() + ); + }, + Err(_elapsed) => { + error!( + "audit sink '{}' exceeded {}s during on_effect — skipped", + handler.name(), + self.timeout.as_secs() + ); + }, + } + } + Ok(()) + } +} + /// Common fields extracted from a type-erased PluginResult. /// /// Handlers return `Box` which wraps this struct. The @@ -1237,4 +1566,70 @@ mod tests { assert!(result.continue_processing); assert!(result.modified_payload.is_some()); } + + #[tokio::test] + async fn effect_emit_fails_closed_when_durable_write_fails() { + use std::sync::Mutex; + + struct FailingLog; + #[async_trait] + impl DurableEffectLog for FailingLog { + async fn append(&self, _e: &EffectRecord) -> Result<(), Box> { + Err(Box::new(PluginError::Config { + message: "wal down".into(), + })) + } + } + struct OkLog; + #[async_trait] + impl DurableEffectLog for OkLog { + async fn append(&self, _e: &EffectRecord) -> Result<(), Box> { + Ok(()) + } + } + struct CountingSink(Arc>); + #[async_trait] + impl AuditHandler for CountingSink { + async fn handle(&self, _p: &dyn PluginPayload, _e: &Extensions, _d: &DecisionLog) {} + async fn on_effect(&self, _e: &EffectRecord, _x: &Extensions) { + *self.0.lock().unwrap() += 1; + } + } + + let effect = EffectRecord::prepared("token_mint", "mint", "k"); + + // Durable write fails → emit fails closed, NO fan-out to sinks. + let calls = Arc::new(Mutex::new(0usize)); + let emitter = AuditEffectEmitter { + handlers: vec![Arc::new(CountingSink(calls.clone()))], + plugin_name: "delegator".into(), + timeout: Duration::from_secs(5), + durable: Some(Arc::new(FailingLog)), + epoch: 0, + stream_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)), + emission_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)), + }; + let res = emitter.emit(&effect, &Extensions::default()).await; + assert!(res.is_err(), "durable write failed → emit fails closed"); + assert_eq!( + *calls.lock().unwrap(), + 0, + "no fan-out when the durable write fails" + ); + + // Durable write succeeds → fan-out proceeds (durable-before-fanout). + let calls2 = Arc::new(Mutex::new(0usize)); + let emitter2 = AuditEffectEmitter { + handlers: vec![Arc::new(CountingSink(calls2.clone()))], + plugin_name: "delegator".into(), + timeout: Duration::from_secs(5), + durable: Some(Arc::new(OkLog)), + epoch: 0, + stream_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)), + emission_seq: Arc::new(std::sync::atomic::AtomicU64::new(0)), + }; + let res2 = emitter2.emit(&effect, &Extensions::default()).await; + assert!(res2.is_ok()); + assert_eq!(*calls2.lock().unwrap(), 1, "durable OK → fan-out proceeds"); + } } diff --git a/crates/cpex-core/src/extensions/container.rs b/crates/cpex-core/src/extensions/container.rs index 899a155e..4d3a0264 100644 --- a/crates/cpex-core/src/extensions/container.rs +++ b/crates/cpex-core/src/extensions/container.rs @@ -122,6 +122,14 @@ pub struct Extensions { pub labels_write_token: Option, #[serde(skip)] pub delegation_write_token: Option, + + /// Effect emitter — a capability handle, set by the executor per plugin + /// (capability `emit_effect`), NOT serialized. Like the write tokens: + /// `filter_extensions` never sets it; the executor does, for capable + /// plugins, right before `handle`. Used by `begin_effect` / + /// `complete_effect` to emit irreversible-effect records to audit sinks. + #[serde(skip)] + pub effect_emitter: Option>, } impl Clone for Extensions { @@ -145,7 +153,99 @@ impl Clone for Extensions { http_write_token: None, labels_write_token: None, delegation_write_token: None, + // Capability handle — set fresh per-invoke by the executor, never + // cloned (same policy as the write tokens above). + effect_emitter: None, + } + } +} + +impl Extensions { + /// Emit an irreversible effect's `prepared` intent to the audit sinks — + /// the write-ahead point (slice 2 emits; durability is a later slice). + /// No-op unless the plugin holds the `emit_effect` capability (the + /// executor sets `effect_emitter` only for those). + pub async fn begin_effect( + &self, + effect: &crate::effect::EffectRecord, + ) -> Result<(), Box> { + match &self.effect_emitter { + Some(emitter) => { + let prepared = effect + .clone() + .into_state(crate::effect::EffectState::Prepared); + emitter.emit(&prepared, self).await + }, + // No capability → nothing to durably record, nothing to emit. + None => Ok(()), + } + } + + /// Emit an effect's terminal outcome (`Confirmed` / `Rejected` / + /// `Unknown`). No-op without the `emit_effect` capability. Unlike + /// `begin_effect`, a durable-write failure here is not fail-closed — the + /// act already happened; the record stays `prepared`/`unknown` for the + /// recovery sweep (slice 3b) to reconcile — so callers may log-and-continue. + pub async fn complete_effect( + &self, + effect: &crate::effect::EffectRecord, + state: crate::effect::EffectState, + ) -> Result<(), Box> { + match &self.effect_emitter { + Some(emitter) => { + let done = effect.clone().into_state(state); + emitter.emit(&done, self).await + }, + None => Ok(()), + } + } + + /// Perform an irreversible external effect under framework-mediated + /// write-ahead. Brackets `act` — the actual I/O, e.g. a token mint at an + /// IdP — between a durable, fail-closed [`Self::begin_effect`] and a + /// best-effort [`Self::complete_effect`], so a caller cannot skip, + /// reorder, or forget the durability protocol. + /// + /// - If the write-ahead fails, `act` never runs and the error is returned + /// (fail-closed: no durable intent → no act). + /// - On `Ok`, the effect is recorded `confirmed`. + /// - On `Err`, it is recorded `unknown` — **not** `rejected`: a failed call + /// may still have taken effect at the participant (e.g. the response was + /// lost after the act landed), so recovery reconciles it via `key` rather + /// than assuming it didn't happen. + /// + /// This is the primitive builtins mint *through*; the structural + /// enforcement is that a builtin never performs the I/O itself, only via + /// primitives built on this bracket. + pub async fn perform_effect( + &self, + effect: &crate::effect::EffectRecord, + act: F, + ) -> Result> + where + F: FnOnce() -> Fut, + Fut: std::future::Future>>, + { + // Write-ahead: durable and fail-closed. If it errors, the act never runs. + self.begin_effect(effect).await?; + + // The irreversible act. + let outcome = act().await; + + // Record the terminal state. `Err` → `Unknown` (conservative): a failed + // call may still have landed at the participant, so reconciliation via + // K decides — never assume rejection. Best-effort: the act has already + // happened, so a completion-write failure is logged, not fatal. + let terminal = if outcome.is_ok() { + crate::effect::EffectState::Confirmed + } else { + crate::effect::EffectState::Unknown + }; + if let Err(e) = self.complete_effect(effect, terminal).await { + tracing::warn!(effect_key = %effect.key, "effect completion not durably recorded: {e}"); } + + outcome } } @@ -503,6 +603,110 @@ mod tests { DelegationExtension, HttpExtension, RequestExtension, SecurityExtension, }; + /// The mediated `perform_effect` bracket: write-ahead before the act, + /// terminal state after — `confirmed` on success, `unknown` on failure + /// (never assume rejection), and fail-closed if the write-ahead fails. + #[tokio::test] + async fn perform_effect_mediates_write_ahead() { + use crate::effect::{EffectEmitter, EffectRecord, EffectState}; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Mutex; + + #[derive(Debug)] + struct RecordingEmitter { + states: Arc>>, + fail_begin: bool, + } + #[async_trait::async_trait] + impl EffectEmitter for RecordingEmitter { + async fn emit( + &self, + effect: &EffectRecord, + _ext: &Extensions, + ) -> Result<(), Box> { + if self.fail_begin && effect.state == EffectState::Prepared { + return Err(crate::error::PluginError::Config { + message: "wal down".into(), + } + .boxed()); + } + self.states.lock().unwrap().push(effect.state.clone()); + Ok(()) + } + } + + fn ext_with(states: &Arc>>, fail_begin: bool) -> Extensions { + let mut ext = Extensions::default(); + ext.effect_emitter = Some(Arc::new(RecordingEmitter { + states: states.clone(), + fail_begin, + })); + ext + } + + // 1. Success → prepared, confirmed; act ran; value returned. + { + let states = Arc::new(Mutex::new(Vec::new())); + let ext = ext_with(&states, false); + let ran = Arc::new(AtomicBool::new(false)); + let ran2 = ran.clone(); + let effect = EffectRecord::prepared("token_mint", "ok", "k-ok"); + let out: Result = ext + .perform_effect(&effect, || async move { + ran2.store(true, Ordering::SeqCst); + Ok(42) + }) + .await; + assert_eq!(out.unwrap(), 42); + assert!(ran.load(Ordering::SeqCst), "act ran"); + assert_eq!( + *states.lock().unwrap(), + vec![EffectState::Prepared, EffectState::Confirmed] + ); + } + + // 2. Act fails → prepared, unknown (conservative, not rejected). + { + let states = Arc::new(Mutex::new(Vec::new())); + let ext = ext_with(&states, false); + let effect = EffectRecord::prepared("token_mint", "boom", "k-boom"); + let out: Result = ext + .perform_effect(&effect, || async move { + Err(crate::error::PluginError::Config { + message: "mint failed".into(), + } + .boxed()) + }) + .await; + assert!(out.is_err()); + assert_eq!( + *states.lock().unwrap(), + vec![EffectState::Prepared, EffectState::Unknown] + ); + } + + // 3. Write-ahead fails → act never runs, error returned, nothing recorded. + { + let states = Arc::new(Mutex::new(Vec::new())); + let ext = ext_with(&states, true); + let ran = Arc::new(AtomicBool::new(false)); + let ran2 = ran.clone(); + let effect = EffectRecord::prepared("token_mint", "nope", "k-nope"); + let out: Result = ext + .perform_effect(&effect, || async move { + ran2.store(true, Ordering::SeqCst); + Ok(7) + }) + .await; + assert!(out.is_err(), "begin failure is fail-closed"); + assert!( + !ran.load(Ordering::SeqCst), + "act must not run without durable intent" + ); + assert!(states.lock().unwrap().is_empty()); + } + } + fn make_extensions() -> Extensions { let mut security = SecurityExtension::default(); security.add_label("PII"); diff --git a/crates/cpex-core/src/hooks/payload.rs b/crates/cpex-core/src/hooks/payload.rs index ef3e6950..352bc70d 100644 --- a/crates/cpex-core/src/hooks/payload.rs +++ b/crates/cpex-core/src/hooks/payload.rs @@ -86,6 +86,44 @@ pub trait PluginPayload: Send + Sync + 'static { /// Downcast to a concrete type via `&mut dyn Any`. fn as_any_mut(&mut self) -> &mut dyn Any; + + /// Canonical bytes of this payload for content-addressed audit provenance, + /// or `None` for payloads that can't or shouldn't be serialized (the + /// default). The bytes feed a content hash — **only the digest is + /// retained, never the bytes** — so a node's provenance is recorded + /// without re-spilling its (possibly sensitive) content. Computed only + /// when content provenance is enabled, so the default keeps the hot path + /// free. + /// + /// **Byte-stability (what a consumer may assume).** + /// `impl_plugin_payload!(_, audit_serialize)` derives this by round-tripping + /// through `serde_json::Value` — whose `Map` is a `BTreeMap`, so object keys + /// are sorted. Identical content therefore serializes to identical bytes + /// across runs and processes, and **two equal digests mean "same content" + /// within a deployment**. It is *sorted-key JSON, not full RFC 8785 (JCS)*: + /// number formatting follows `serde_json` and is stable within a + /// `serde_json` version but is not guaranteed by a canonicalization spec + /// across toolchains. So treat digest equality as same-content within a + /// build; do not assume cross-toolchain canonicalization. A hand-written + /// `audit_bytes` must preserve this property (a canonical, deterministic + /// encoding) or its hashes will not be comparable. + fn audit_bytes(&self) -> Option> { + None + } +} + +/// The content hash of canonical audit bytes — a content-addressed provenance +/// ref (`sha256:`). Only the digest is kept; the bytes are never retained. +pub fn content_hash(bytes: &[u8]) -> String { + use sha2::{Digest, Sha256}; + use std::fmt::Write as _; + let digest = Sha256::digest(bytes); + let mut s = String::with_capacity(7 + 64); + s.push_str("sha256:"); + for b in digest { + let _ = write!(s, "{b:02x}"); + } + s } impl fmt::Debug for dyn PluginPayload { @@ -122,4 +160,73 @@ macro_rules! impl_plugin_payload { } } }; + // `audit_serialize`: opt in to content-provenance hashing for a + // `Serialize` payload. `audit_bytes` round-trips through `Value` so object + // keys are sorted (serde_json's Map is a BTreeMap without `preserve_order`) + // — canonical, cross-process-stable bytes even when the payload holds + // HashMaps. + ($ty:ty, audit_serialize) => { + impl $crate::hooks::payload::PluginPayload for $ty { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + fn audit_bytes(&self) -> Option> { + let value = serde_json::to_value(self).ok()?; + serde_json::to_vec(&value).ok() + } + } + }; +} + +#[cfg(test)] +mod tests { + use super::*; + use serde::Serialize; + + #[derive(Clone, Serialize)] + struct Doc { + a: u32, + note: String, + } + crate::impl_plugin_payload!(Doc, audit_serialize); + + #[derive(Clone)] + struct Opaque; + crate::impl_plugin_payload!(Opaque); + + #[test] + fn audit_serialize_is_some_and_deterministic() { + let d = Doc { + a: 1, + note: "hi".into(), + }; + let b1 = d.audit_bytes().expect("serializable payload → Some"); + let b2 = d.clone().audit_bytes().expect("Some"); + assert_eq!(b1, b2, "canonical bytes are deterministic"); + } + + #[test] + fn default_audit_bytes_is_none() { + // A payload that did not opt into `audit_serialize` yields no bytes. + assert!(Opaque.audit_bytes().is_none()); + } + + #[test] + fn content_hash_is_prefixed_and_stable() { + let h = content_hash(b"hello"); + assert!(h.starts_with("sha256:")); + assert_eq!(h.len(), "sha256:".len() + 64, "sha256 hex is 64 chars"); + assert_eq!(content_hash(b"hello"), h, "deterministic"); + assert_ne!( + content_hash(b"world"), + h, + "different input → different hash" + ); + } } diff --git a/crates/cpex-core/src/lib.rs b/crates/cpex-core/src/lib.rs index fae7bd25..dafac114 100644 --- a/crates/cpex-core/src/lib.rs +++ b/crates/cpex-core/src/lib.rs @@ -28,10 +28,13 @@ // approval, confirmation, step-up, …) // - [`error`] — Error types, violations, and result types +pub mod audit; pub mod cmf; pub mod config; pub mod context; +pub mod decision; pub mod delegation; +pub mod effect; pub mod elicitation; pub mod error; pub mod executor; diff --git a/crates/cpex-core/src/manager.rs b/crates/cpex-core/src/manager.rs index da5bb66c..ce1accf1 100644 --- a/crates/cpex-core/src/manager.rs +++ b/crates/cpex-core/src/manager.rs @@ -32,6 +32,7 @@ use std::sync::{Arc, RwLock}; use hashbrown::HashMap; use tracing::{error, info, warn}; +use crate::audit::AuditHandler; use crate::config::{self, CpexConfig}; use crate::context::PluginContextTable; use crate::error::PluginError; @@ -338,10 +339,21 @@ fn instantiate_plugins_into( /// the route-cache cap from `plugin_settings` so both registration paths /// agree on field-by-field translation. fn snapshot_from_config(registry: PluginRegistry, cpex_config: CpexConfig) -> RuntimeSnapshot { - let executor = Executor::new(ExecutorConfig { + let mut executor = Executor::new(ExecutorConfig { timeout_seconds: cpex_config.plugin_settings.plugin_timeout, short_circuit_on_deny: cpex_config.plugin_settings.short_circuit_on_deny, - }); + capture_content_provenance: cpex_config.plugin_settings.capture_content_provenance, + }) + .with_audit_handlers(registry.audit_handlers()); + // Opt-in durable effect WAL — installed only when a path is configured. + // Absent → effect auditing stays ordering-only (basic logging). + if let Some(path) = &cpex_config.plugin_settings.effect_log_path { + let mut log = crate::effect::FileEffectLog::new(path); + if let Some(threshold) = cpex_config.plugin_settings.effect_log_compaction_threshold { + log = log.with_compaction_threshold(threshold); + } + executor = executor.with_effect_log(Arc::new(log)); + } let route_cache_max_entries = cpex_config.plugin_settings.route_cache_max_entries; RuntimeSnapshot { registry, @@ -376,6 +388,50 @@ impl PluginManager { } } + /// Register an observation-only audit sink. It is invoked at the verdict + /// of every subsequent pipeline run with the decision log — for allowed + /// and denied requests alike. Attached to the current executor via + /// copy-on-write; audit sinks cannot influence pipeline outcomes. + pub fn register_audit_handler(&self, handler: Arc) { + self.mutate_runtime(|snap| snap.executor.push_audit_handler(handler)); + } + + /// Install a durable effect-audit WAL programmatically (copy-on-write). + /// When present, `begin_effect` is crash-safe and fail-closed. Like + /// [`Self::register_audit_handler`], this does not survive `load_config`, + /// which rebuilds the executor from `plugin_settings.effect_log_path` — + /// declare the path in config for the WAL to persist across reloads. + pub fn install_effect_log(&self, effect_log: Arc) { + self.mutate_runtime(|snap| snap.executor.set_effect_log(effect_log)); + } + + /// Run effect-WAL crash recovery with the default reconciler + /// ([`crate::effect::LogUnknownsReconciler`]): compact completed effects + /// and log any unresolved (`prepared`-orphan / `unknown`) ones, returning + /// them. A no-op when no durable effect log is installed. Call once at + /// startup, after config load. Use [`Self::recover_effects_with`] to supply + /// a reconciler that can query an authoritative issuance ledger by key. + pub async fn recover_effects( + &self, + ) -> Result, Box> { + self.recover_effects_with(&crate::effect::LogUnknownsReconciler) + .await + } + + /// Like [`Self::recover_effects`], but with a caller-supplied reconciler + /// that resolves `unknown` effects by looking up `EffectRecord::key` in an + /// authoritative issuance ledger. The reconciler reads the self-describing + /// record; it is not plugin-specific. + pub async fn recover_effects_with( + &self, + reconciler: &dyn crate::effect::EffectReconciler, + ) -> Result, Box> { + match self.load_runtime().executor.effect_log() { + Some(log) => log.recover_and_reconcile(reconciler).await, + None => Ok(Vec::new()), + } + } + /// Load the current runtime snapshot (lock-free, single atomic op). fn load_runtime(&self) -> Arc { self.runtime.load_full() @@ -1852,11 +1908,11 @@ mod tests { // -- Test payload -- - #[derive(Debug, Clone)] + #[derive(Debug, Clone, serde::Serialize)] struct TestPayload { value: String, } - crate::impl_plugin_payload!(TestPayload); + crate::impl_plugin_payload!(TestPayload, audit_serialize); // -- Test hook type -- @@ -2070,6 +2126,795 @@ mod tests { assert_eq!(result.violation.as_ref().unwrap().code, "denied"); } + #[tokio::test] + async fn test_decision_log_records_allow_and_deny() { + use crate::decision::{PluginAction, Verdict}; + + // Allow path: one Allowed step, and a non-deny verdict. + { + let mgr = PluginManager::default(); + let config = make_config("allow-plugin", 10, PluginMode::Sequential); + let plugin = Arc::new(AllowPlugin { + cfg: config.clone(), + }); + mgr.register_handler::(plugin, config).unwrap(); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + assert!(result.continue_processing); + assert!(!result.decision_log.is_denied()); + let steps = result.decision_log.steps(); + assert_eq!(steps.len(), 1); + assert_eq!(steps[0].plugin_name, "allow-plugin"); + assert_eq!(steps[0].action, PluginAction::Allowed); + } + + // Deny path: a Denied step, plus a Deny verdict carrying the violation. + { + let mgr = PluginManager::default(); + let config = make_config("deny-plugin", 10, PluginMode::Sequential); + let plugin = Arc::new(DenyPlugin { + cfg: config.clone(), + }); + mgr.register_handler::(plugin, config).unwrap(); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + assert!(!result.continue_processing); + assert!(result.decision_log.is_denied()); + let steps = result.decision_log.steps(); + assert_eq!(steps.len(), 1); + assert_eq!(steps[0].action, PluginAction::Denied); + match result.decision_log.verdict() { + Some(Verdict::Deny(v)) => assert_eq!(v.code, "denied"), + other => panic!("expected a deny verdict, got {other:?}"), + } + } + } + + #[tokio::test] + async fn test_audit_handler_observes_verdict() { + use crate::audit::AuditHandler; + use crate::decision::DecisionLog; + use std::sync::Mutex; + + // An audit sink that records whether each observed verdict was a deny. + struct CapturingAudit { + denied: Arc>>, + } + #[async_trait] + impl AuditHandler for CapturingAudit { + async fn handle( + &self, + _payload: &dyn PluginPayload, + _extensions: &Extensions, + decisions: &DecisionLog, + ) { + self.denied.lock().unwrap().push(decisions.is_denied()); + } + } + + let denied = Arc::new(Mutex::new(Vec::new())); + + // Allow path fires the sink once, observed as not-denied. + { + let mgr = PluginManager::default(); + let config = make_config("allow-plugin", 10, PluginMode::Sequential); + mgr.register_handler::( + Arc::new(AllowPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.register_audit_handler(Arc::new(CapturingAudit { + denied: denied.clone(), + })); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + assert!(result.continue_processing); + } + + // Deny path fires the sink once too — the whole point of the seam: + // a blocked request is still audited. + { + let mgr = PluginManager::default(); + let config = make_config("deny-plugin", 10, PluginMode::Sequential); + mgr.register_handler::( + Arc::new(DenyPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.register_audit_handler(Arc::new(CapturingAudit { + denied: denied.clone(), + })); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + assert!(!result.continue_processing); + } + + let seen = denied.lock().unwrap(); + assert_eq!(seen.len(), 2, "audit sink fires once per invocation"); + assert!(!seen[0], "allow verdict observed as not-denied"); + assert!(seen[1], "deny verdict observed as denied"); + } + + #[tokio::test] + async fn test_panicking_audit_handler_is_contained() { + use crate::audit::AuditHandler; + use crate::decision::DecisionLog; + + // An audit sink that panics. It must not take down the request — the + // verdict is already decided by the time the sink runs. (The panic + // prints to stderr via the default hook before being caught; that's + // expected noise, not a failure.) + struct PanicAudit; + #[async_trait] + impl AuditHandler for PanicAudit { + async fn handle( + &self, + _payload: &dyn PluginPayload, + _extensions: &Extensions, + _decisions: &DecisionLog, + ) { + panic!("audit sink boom"); + } + } + + let mgr = PluginManager::default(); + let config = make_config("allow-plugin", 10, PluginMode::Sequential); + mgr.register_handler::( + Arc::new(AllowPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.register_audit_handler(Arc::new(PanicAudit)); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + // The sink panicked, but the request still completed with its verdict. + assert!(result.continue_processing); + } + + #[tokio::test] + async fn test_audit_plugin_auto_attaches_from_config() { + use crate::audit::AuditHandler; + use crate::decision::DecisionLog; + use std::sync::Mutex; + + // A capturing audit sink that auto-attaches in audit-only mode + // (no hooks), mirroring how `audit-logger` opts in. + struct CapturingAudit { + cfg: PluginConfig, + fired: Arc>, + } + #[async_trait] + impl Plugin for CapturingAudit { + fn config(&self) -> &PluginConfig { + &self.cfg + } + fn as_audit_handler(self: Arc) -> Option> { + if self.cfg.hooks.is_empty() { + Some(self) + } else { + None + } + } + } + #[async_trait] + impl AuditHandler for CapturingAudit { + async fn handle( + &self, + _payload: &dyn PluginPayload, + _extensions: &Extensions, + _decisions: &DecisionLog, + ) { + *self.fired.lock().unwrap() += 1; + } + } + struct CapturingAuditFactory(Arc>); + impl crate::factory::PluginFactory for CapturingAuditFactory { + fn create( + &self, + config: &PluginConfig, + ) -> Result> { + Ok(crate::factory::PluginInstance { + plugin: Arc::new(CapturingAudit { + cfg: config.clone(), + fired: Arc::clone(&self.0), + }), + handlers: vec![], + }) + } + } + + // Config declares the audit sink with NO hooks — it must auto-attach. + let yaml = r#" +plugins: + - name: audit + kind: test/audit + mode: audit + - name: gate + kind: test/allow + hooks: [test_hook] + mode: sequential +"#; + let cpex_config = crate::config::parse_config(yaml).unwrap(); + + let fired = Arc::new(Mutex::new(0usize)); + let mgr = PluginManager::default(); + mgr.register_factory( + "test/audit", + Box::new(CapturingAuditFactory(Arc::clone(&fired))), + ); + mgr.register_factory("test/allow", Box::new(AllowPluginFactory)); + mgr.load_config(cpex_config).unwrap(); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + assert!(result.continue_processing); + + // The hookless audit plugin auto-attached from config and fired at the + // verdict — no programmatic register_audit_handler, no `hooks:`. + assert_eq!(*fired.lock().unwrap(), 1); + } + + #[tokio::test] + async fn test_effect_emit_is_capability_gated() { + use crate::audit::AuditHandler; + use crate::decision::DecisionLog; + use crate::effect::EffectRecord; + use std::sync::Mutex; + + // A plugin that emits a prepared token-mint effect during handle(). + struct EffectPlugin { + cfg: PluginConfig, + } + #[async_trait] + impl Plugin for EffectPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } + } + impl HookHandler for EffectPlugin { + async fn handle( + &self, + _payload: &TestPayload, + ext: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let effect = + EffectRecord::prepared("token_mint", "exchange for workday-api", "k-1") + .with_detail("audience", "workday-api"); + // Fail-closed: if the durable prepare fails, don't act. + if ext.begin_effect(&effect).await.is_err() { + return PluginResult::deny(PluginViolation::new( + "effect_prepare_failed", + "could not durably record effect intent", + )); + } + PluginResult::allow() + } + } + + // A sink that records the effects it observes. + struct CapturingEffectAudit { + seen: Arc>>, // (kind, state) + } + #[async_trait] + impl AuditHandler for CapturingEffectAudit { + async fn handle(&self, _p: &dyn PluginPayload, _e: &Extensions, _d: &DecisionLog) {} + async fn on_effect(&self, effect: &EffectRecord, _ext: &Extensions) { + self.seen + .lock() + .unwrap() + .push((effect.kind.clone(), format!("{:?}", effect.state))); + } + } + + // With the `emit_effect` capability → the sink observes the effect. + { + let seen = Arc::new(Mutex::new(Vec::new())); + let mgr = PluginManager::default(); + let mut config = make_config("effect-plugin", 10, PluginMode::Sequential); + config.capabilities.insert("emit_effect".to_string()); + mgr.register_handler::( + Arc::new(EffectPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.register_audit_handler(Arc::new(CapturingEffectAudit { seen: seen.clone() })); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let _ = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + let s = seen.lock().unwrap(); + assert_eq!(s.len(), 1, "capable plugin's effect reaches the sink"); + assert_eq!(s[0].0, "token_mint"); + assert_eq!(s[0].1, "Prepared"); + } + + // WITHOUT the capability → begin_effect is a no-op; the sink sees nothing. + { + let seen = Arc::new(Mutex::new(Vec::new())); + let mgr = PluginManager::default(); + let config = make_config("effect-plugin", 10, PluginMode::Sequential); // no emit_effect cap + mgr.register_handler::( + Arc::new(EffectPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.register_audit_handler(Arc::new(CapturingEffectAudit { seen: seen.clone() })); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let _ = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + assert!( + seen.lock().unwrap().is_empty(), + "no capability → no effect emitted" + ); + } + } + + /// The global `emission_seq` orders effect records against the decision + /// record within one invocation — effects emit during `handle`, the + /// decision at the verdict — so a consumer that merges the two streams can + /// reconstruct their interleave. Per-stream ids keep the two streams + /// distinguishable. + #[tokio::test] + async fn emission_seq_interleaves_effects_and_decision() { + use crate::audit::AuditHandler; + use crate::decision::DecisionLog; + use crate::effect::{EffectRecord, EffectState}; + use std::sync::Mutex; + + // (label, stream_id, emission_seq) for every record the sink observes. + let log: Arc>> = Arc::new(Mutex::new(Vec::new())); + + struct EffectPlugin { + cfg: PluginConfig, + } + #[async_trait] + impl Plugin for EffectPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } + } + impl HookHandler for EffectPlugin { + async fn handle( + &self, + _payload: &TestPayload, + ext: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let e = EffectRecord::prepared("token_mint", "exchange", "k-1"); + ext.begin_effect(&e).await.unwrap(); + let _ = ext.complete_effect(&e, EffectState::Confirmed).await; + PluginResult::allow() + } + } + + struct RecordingSink { + log: Arc>>, + } + #[async_trait] + impl AuditHandler for RecordingSink { + async fn handle(&self, _p: &dyn PluginPayload, _e: &Extensions, d: &DecisionLog) { + self.log.lock().unwrap().push(( + "decision".into(), + d.stream_id().unwrap_or_default().to_string(), + d.emission_seq().unwrap(), + )); + } + async fn on_effect(&self, effect: &EffectRecord, _e: &Extensions) { + self.log.lock().unwrap().push(( + format!("effect:{:?}", effect.state), + effect.stream_id.clone().unwrap_or_default(), + effect.emission_seq.unwrap(), + )); + } + } + + let mgr = PluginManager::default(); + let mut config = make_config("effect-plugin", 10, PluginMode::Sequential); + config.capabilities.insert("emit_effect".to_string()); + mgr.register_handler::( + Arc::new(EffectPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.register_audit_handler(Arc::new(RecordingSink { log: log.clone() })); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let _ = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + let l = log.lock().unwrap(); + assert_eq!( + l.len(), + 3, + "prepared + confirmed (during handle) + decision; got {l:?}" + ); + // Global emission_seq strictly increasing in emission order — the effects + // (emitted during handle) precede the decision (emitted at the verdict). + assert!( + l[0].2 < l[1].2 && l[1].2 < l[2].2, + "emission_seq orders the interleave; got {l:?}" + ); + assert_eq!(l[0].0, "effect:Prepared"); + assert_eq!(l[2].0, "decision"); + // Distinct per-type streams. + assert_eq!(l[0].1, "effect", "effect stream id"); + assert_eq!(l[2].1, "decision", "decision stream id"); + } + + /// `on_effect` fires on the `begin_effect` (prepared) leg — a sink observes + /// the intent record *before* the effect body runs, not only at completion. + /// Regression guard: evidence-of-intent is the write-ahead's whole value, so + /// a change that silently reduced sinks to completions-only must fail here. + #[tokio::test] + async fn on_effect_fires_on_the_prepared_leg_before_the_act() { + use crate::audit::AuditHandler; + use crate::decision::DecisionLog; + use crate::effect::{EffectRecord, EffectState}; + use std::sync::Mutex; + + // One ordered log that both the plugin and the sink append to. + let order: Arc>> = Arc::new(Mutex::new(Vec::new())); + + struct OrderingPlugin { + cfg: PluginConfig, + order: Arc>>, + } + #[async_trait] + impl Plugin for OrderingPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } + } + impl HookHandler for OrderingPlugin { + async fn handle( + &self, + _payload: &TestPayload, + ext: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let effect = EffectRecord::prepared("token_mint", "exchange", "k-1"); + // `begin_effect` is awaited, so the sink has observed `prepared` + // by the time it returns — before we record the act below. + ext.begin_effect(&effect).await.unwrap(); + self.order.lock().unwrap().push("act"); + let _ = ext.complete_effect(&effect, EffectState::Confirmed).await; + PluginResult::allow() + } + } + + struct OrderingSink { + order: Arc>>, + } + #[async_trait] + impl AuditHandler for OrderingSink { + async fn handle(&self, _p: &dyn PluginPayload, _e: &Extensions, _d: &DecisionLog) {} + async fn on_effect(&self, effect: &EffectRecord, _e: &Extensions) { + if effect.state == EffectState::Prepared { + self.order.lock().unwrap().push("on_effect:prepared"); + } + } + } + + let mgr = PluginManager::default(); + let mut config = make_config("ordering-plugin", 10, PluginMode::Sequential); + config.capabilities.insert("emit_effect".to_string()); + mgr.register_handler::( + Arc::new(OrderingPlugin { + cfg: config.clone(), + order: order.clone(), + }), + config, + ) + .unwrap(); + mgr.register_audit_handler(Arc::new(OrderingSink { + order: order.clone(), + })); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let _ = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + let log = order.lock().unwrap(); + assert_eq!( + *log, + ["on_effect:prepared", "act"], + "sink must observe the prepared record before the effect body runs; got {:?}", + *log + ); + } + + /// End-to-end (slice 3b wiring): an `emit_effect` plugin's `begin_effect` + /// intent is durably written to the configured WAL file and round-trips + /// as a `prepared` token_mint. + #[tokio::test] + async fn effect_wal_persists_prepared_intent_end_to_end() { + use crate::effect::{EffectRecord, EffectState, FileEffectLog}; + + struct WalEffectPlugin { + cfg: PluginConfig, + } + #[async_trait] + impl Plugin for WalEffectPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } + } + impl HookHandler for WalEffectPlugin { + async fn handle( + &self, + _payload: &TestPayload, + ext: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + let effect = + EffectRecord::prepared("token_mint", "exchange for workday-api", "k-1") + .with_detail("audience", "workday-api"); + // Fail-closed: don't act unless the intent is durable. + if ext.begin_effect(&effect).await.is_err() { + return PluginResult::deny(PluginViolation::new( + "effect_prepare_failed", + "could not durably record effect intent", + )); + } + PluginResult::allow() + } + } + + let path = std::env::temp_dir().join(format!("cpex_wal_e2e_{}.ndjson", std::process::id())); + let _ = std::fs::remove_file(&path); + + let mgr = PluginManager::default(); + let mut config = make_config("wal-effect-plugin", 10, PluginMode::Sequential); + config.capabilities.insert("emit_effect".to_string()); + mgr.register_handler::( + Arc::new(WalEffectPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.install_effect_log(Arc::new(FileEffectLog::new(&path))); + mgr.initialize().await.unwrap(); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let _ = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + + // The WAL file holds one durable `prepared` record, written before the + // (would-be) act by `begin_effect`. + let contents = std::fs::read_to_string(&path).expect("WAL written by begin_effect"); + let line = contents.lines().next().expect("one prepared record"); + let rec: EffectRecord = serde_json::from_str(line).expect("record round-trips"); + assert_eq!(rec.kind, "token_mint"); + assert_eq!(rec.state, EffectState::Prepared); + assert_eq!(rec.key, "k-1"); + + let _ = std::fs::remove_file(&path); + } + + /// Startup recovery: the manager reconciles an orphaned intent in the + /// installed WAL and compacts it out. + #[tokio::test] + async fn manager_recover_effects_reconciles_installed_wal() { + use crate::effect::{ + DurableEffectLog, EffectReconciler, EffectRecord, EffectState, FileEffectLog, + }; + + struct AlwaysConfirm; + #[async_trait] + impl EffectReconciler for AlwaysConfirm { + async fn reconcile(&self, _e: &EffectRecord) -> EffectState { + EffectState::Confirmed + } + } + + let path = + std::env::temp_dir().join(format!("cpex_recover_mgr_{}.ndjson", std::process::id())); + let _ = std::fs::remove_file(&path); + + // Seed an orphaned `prepared` intent (a crash before the outcome). + let log = Arc::new(FileEffectLog::new(&path)); + log.append(&EffectRecord::prepared("token_mint", "orphan", "k-1")) + .await + .unwrap(); + + let mgr = PluginManager::default(); + mgr.install_effect_log(log.clone()); + + let still = mgr.recover_effects_with(&AlwaysConfirm).await.unwrap(); + assert!(still.is_empty(), "the orphan was confirmed and compacted"); + + let contents = std::fs::read_to_string(&path).unwrap(); + assert!( + contents.trim().is_empty(), + "WAL compacted to empty after recovery" + ); + + let _ = std::fs::remove_file(&path); + } + + /// The no-arg default reconciler leaves an unresolvable orphan `unknown` + /// (it has no ledger to query), returning it for the operator. + #[tokio::test] + async fn manager_recover_effects_default_leaves_unknowns() { + use crate::effect::{DurableEffectLog, EffectRecord, FileEffectLog}; + + let path = std::env::temp_dir().join(format!( + "cpex_recover_default_{}.ndjson", + std::process::id() + )); + let _ = std::fs::remove_file(&path); + + let log = Arc::new(FileEffectLog::new(&path)); + log.append(&EffectRecord::prepared("token_mint", "orphan", "k-1")) + .await + .unwrap(); + + let mgr = PluginManager::default(); + mgr.install_effect_log(log.clone()); + + let still = mgr.recover_effects().await.unwrap(); + assert_eq!( + still.len(), + 1, + "default reconciler leaves the orphan unknown" + ); + assert_eq!(still[0].key, "k-1"); + + let _ = std::fs::remove_file(&path); + } + + /// The decision log attached to a pipeline result carries a child span: + /// same trace, the request's span as the causal parent, a fresh own span. + #[tokio::test] + async fn decision_log_carries_child_span_from_request() { + use crate::extensions::RequestExtension; + + let mgr = PluginManager::default(); + let config = make_config("allow-plugin", 10, PluginMode::Sequential); + mgr.register_handler::( + Arc::new(AllowPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.initialize().await.unwrap(); + + let mut ext = Extensions::default(); + ext.request = Some(Arc::new(RequestExtension { + trace_id: Some("trace-xyz".into()), + span_id: Some("upstream-span".into()), + ..Default::default() + })); + + let payload: Box = Box::new(TestPayload { value: "x".into() }); + let (result, _) = mgr.invoke_by_name("test_hook", payload, ext, None).await; + + let span = result + .decision_log + .span() + .expect("span set at pipeline entry"); + assert_eq!(span.trace_id, "trace-xyz", "same trace as the request"); + assert_eq!( + span.parent_span_id.as_deref(), + Some("upstream-span"), + "request span becomes the causal parent" + ); + assert!(!span.span_id.is_empty()); + assert_ne!( + span.span_id, "upstream-span", + "own fresh span, not the parent's" + ); + } + + /// Content provenance is opt-in: the executor captures the input hash only + /// when `capture_content_provenance` is set. + #[tokio::test] + async fn input_hash_captured_only_when_provenance_enabled() { + // Flag ON → the decision log carries the input content hash. + let mgr = PluginManager::new(ManagerConfig { + executor: ExecutorConfig { + capture_content_provenance: true, + ..Default::default() + }, + ..Default::default() + }); + let config = make_config("allow-plugin", 10, PluginMode::Sequential); + mgr.register_handler::( + Arc::new(AllowPlugin { + cfg: config.clone(), + }), + config, + ) + .unwrap(); + mgr.initialize().await.unwrap(); + let payload: Box = Box::new(TestPayload { + value: "hello".into(), + }); + let (result, _) = mgr + .invoke_by_name("test_hook", payload, Extensions::default(), None) + .await; + let hash = result + .decision_log + .input_hash() + .expect("provenance on → input hash captured"); + assert!(hash.starts_with("sha256:"), "content-addressed ref: {hash}"); + + // Flag OFF (default) → no hash on the hot path. + let mgr2 = PluginManager::default(); + let config2 = make_config("allow-plugin", 10, PluginMode::Sequential); + mgr2.register_handler::( + Arc::new(AllowPlugin { + cfg: config2.clone(), + }), + config2, + ) + .unwrap(); + mgr2.initialize().await.unwrap(); + let payload2: Box = Box::new(TestPayload { + value: "hello".into(), + }); + let (result2, _) = mgr2 + .invoke_by_name("test_hook", payload2, Extensions::default(), None) + .await; + assert!( + result2.decision_log.input_hash().is_none(), + "provenance off → no hash" + ); + } + #[tokio::test] async fn test_invoke_typed() { let mgr = PluginManager::default(); @@ -3052,6 +3897,7 @@ mod tests { executor: crate::executor::ExecutorConfig { timeout_seconds: 30, short_circuit_on_deny: false, + capture_content_provenance: false, }, route_cache_max_entries: DEFAULT_ROUTE_CACHE_MAX_ENTRIES, }; @@ -3223,6 +4069,7 @@ mod tests { executor: crate::executor::ExecutorConfig { timeout_seconds: 1, short_circuit_on_deny: true, + capture_content_provenance: false, }, route_cache_max_entries: DEFAULT_ROUTE_CACHE_MAX_ENTRIES, }; diff --git a/crates/cpex-core/src/plugin.rs b/crates/cpex-core/src/plugin.rs index 21282dc2..c1fe2adb 100644 --- a/crates/cpex-core/src/plugin.rs +++ b/crates/cpex-core/src/plugin.rs @@ -98,6 +98,15 @@ pub trait Plugin: Send + Sync { async fn shutdown(&self) -> Result<(), Box> { Ok(()) } + + /// If this plugin is also an audit sink, return it as one so the manager + /// auto-attaches it to the executor's verdict emit when the plugin is + /// registered (including from YAML config). Default: not an audit sink. + fn as_audit_handler( + self: std::sync::Arc, + ) -> Option> { + None + } } /// Declared plugin configuration from the unified YAML config. diff --git a/crates/cpex-core/src/registry.rs b/crates/cpex-core/src/registry.rs index 63eb2cf4..d9497958 100644 --- a/crates/cpex-core/src/registry.rs +++ b/crates/cpex-core/src/registry.rs @@ -351,6 +351,16 @@ impl PluginRegistry { Ok(()) } + /// Collect the audit sinks among the registered plugins. A plugin opts in + /// by overriding `Plugin::as_audit_handler`; most return `None`. The + /// manager attaches these to the executor's verdict emit. + pub fn audit_handlers(&self) -> Vec> { + self.plugins + .values() + .filter_map(|r| r.plugin().clone().as_audit_handler()) + .collect() + } + /// Internal: register handler under one or more hook names. fn register_for_names_inner( &mut self, diff --git a/docs/content/docs/auditing.md b/docs/content/docs/auditing.md new file mode 100644 index 00000000..1acecf31 --- /dev/null +++ b/docs/content/docs/auditing.md @@ -0,0 +1,301 @@ +--- +title: "Auditing" +weight: 65 +--- + +# Auditing + +CPEX can audit its own enforcement — every allow, deny, and modify — plus the +irreversible effects (token mints, approval grants) that plugins cause. This is +the operator and developer guide: how to turn auditing on, what the records look +like, and how to write your own audit sink. + +## What you get + +- **Decision auditing** — one record per invocation with the verdict + (allow / deny / modify), the ordered plugin steps, the request's trace span, + and the taint labels. Unlike a plain observer plugin, this sees **denials**. +- **Effect auditing** — a separate, crash-safe record per irreversible external + action (a token mint, an approval grant), written *ahead* of the act. +- **Provenance** (opt-in) — a content hash of the payload for tamper-evident, + content-addressed lineage without storing the content itself. + +Everything is opt-in: nothing changes until you declare a sink and (optionally) +turn on the effect log and content provenance. + +## Quick start — a decision-audit sink + +Declare the `audit-logger` builtin **with no `hooks:`**. That is what makes it a +decision-audit sink rather than a legacy post-hook observer: + +```yaml +plugins: + - name: audit-logger + kind: audit/logger # the builtin's registered kind + # no hooks: → auto-attaches to the executor's verdict path + config: + destination: stderr # stderr (default) | tracing (target "apl.audit") + source: "prod-gateway-1" # optional label stamped on every record +``` + +That is the whole setup for decision auditing. The sink receives the finalized +decision record and the final extensions directly, so it needs no `read_*` +capabilities. + +> **The one rule that trips people up.** The plugin auto-attaches as a sink +> *only when its `hooks:` list is empty*. If you list hooks on an audit plugin, +> it silently reverts to the **legacy** post-hook mode, which sees only +> *allowed* traffic and **misses every denial** — the exact gap this feature +> closes. **No hooks = sink.** + +## Effect auditing + +To record irreversible effects crash-safely, do two things: turn on the durable +write-ahead log, and grant the *causing* plugin the `emit_effect` capability. + +```yaml +plugin_settings: + effect_log_path: /var/lib/cpex/effects.wal # turns on the write-ahead log (fail-closed) + effect_log_compaction_threshold: 1024 # optional; default 1024, 0 disables + +plugins: + - name: audit-logger + kind: audit/logger + config: { destination: stderr } + + - name: oauth-delegator + kind: delegator/oauth + hooks: [token_delegate] + capabilities: [emit_effect] # ← lets its mints be audited write-ahead + config: { token_endpoint: "https://idp…/token", client_id: "…" } +``` + +- **Without `effect_log_path`**, effect auditing is *ordering-only*: records + still reach the sink, but are not crash-safe and the write-ahead is not + fail-closed. This is the "basic logging" mode. +- **Without `emit_effect`** on the causing plugin, its effect calls are no-ops — + its mints are not audited. + +### Effect lifecycle + +An effect moves through `prepared → confirmed | rejected | unknown`: + +- `prepared` — intent durably recorded *before* the act (fail-closed: no durable + record, no act). +- `confirmed` / `rejected` — the act completed / provably did not. +- `unknown` — the process crashed after the act, before recording the outcome. + A failed call is recorded `unknown`, not `rejected` — it may still have landed + at the participant. + +### Recovery at startup + +When an effect log is configured, the host runs recovery once at boot. It +compacts completed effects and surfaces any that crashed mid-flight, returning +the ones it could not resolve: + +```rust +// Default: logs unresolved effects and leaves them `unknown`. +let unresolved = manager.recover_effects().await?; + +// Or supply a reconciler that can confirm a mint against an issuance ledger +// by its key: +let unresolved = manager.recover_effects_with(&my_reconciler).await?; +``` + +Today an OAuth IdP offers no lookup by mint key, so the default reconciler logs +unresolved mints and leaves them `unknown` for an operator to investigate. + +## Content provenance + +```yaml +plugin_settings: + capture_content_provenance: true # executor hashes the payload at entry (sha256:…) +``` + +Opt-in, because hashing is on the request path. The executor hashes the payload +at pipeline entry; the sink hashes the output. **Only digests are kept, never +the content** — you get lineage and tamper-evidence without re-spilling the data +a PII scanner exists to redact. A payload type opts in to being hashable (the +CMF message payload does); others emit no hash. + +## What a record looks like + +With `destination: stderr` you get one JSON line per decision, and a separate +line per effect: + +```jsonc +// decision +{ + "ts": "2026-08-14T…", "plugin": "audit-logger", "source": "prod-gateway-1", + "subject": { "id": "alice@corp.com", "roles": ["hr"] }, + "verdict": { "deny": { "code": "missing_permission", "reason": "…" } }, + "decision_steps": [ + { "plugin": "pii-scanner", "phase": "Transform", "action": "ModifiedPayload" }, + { "plugin": "cedar-pdp", "phase": "Sequential", "action": "Denied" } + ], + "span": { "trace_id": "…", "span_id": "…", "parent_span_id": "…" }, + "taint": { "input": ["PII"], "final": ["PII", "secret"] }, + "content": { "input_hash": "sha256:…", "output_hash": "sha256:…" }, + "epoch": 1723680000000000000, "stream_id": "decision", "stream_seq": 413, "emission_seq": 913 +} + +// effect +{ + "ts": "2026-08-14T…", + "effect": { + "kind": "token_mint", "state": "confirmed", "key": "…", + "caused_by": "oauth-delegator", + "details": { "audience": "workday-api", "scope": "read_compensation" }, + "epoch": 1723680000000000000, "stream_id": "effect", "stream_seq": 7, "emission_seq": 912 + } +} +``` + +Fields appear only when present: `span` always; `taint` when labels exist; +`content` only when content provenance is enabled; `subject` when a subject is +resolved. + +### Sequence numbers — completeness vs. order + +Four fields — `epoch`, `stream_id`, `stream_seq`, `emission_seq` — let a +downstream store prove properties about the stream it received. Two are +**claims** a verifier checks; two **scope** those claims. Don't use one claim +for the other's job: + +- `stream_seq` is a **completeness** claim. It is dense (gap-free) within its + `(epoch, stream_id)`. **A gap means a record was dropped** — a consumer of one + stream can prove nothing was silently lost. +- `emission_seq` is an **ordering** claim only. It is monotonic across *both* + streams within an epoch, so a consumer that merges decisions and effects can + reconstruct their interleave (an effect emits during a request, so it carries + a lower `emission_seq` than the decision that closed the request). **A + single-stream consumer sees it sparse by design — the gaps are the other + stream's records, not a loss.** Do not detect loss from `emission_seq`. +- `stream_id` scopes `stream_seq` — it names the per-type stream, `"decision"` + or `"effect"` (the entry-type a merged consumer keys on). Decisions and + effects each get their own dense counter, so a consumer of just one still has + gap-free completeness. +- `epoch` scopes both counters. It is the executor's boot time (Unix + nanoseconds), so a *new, larger* value marks a restart: `stream_seq` proves + completeness within an epoch, and across a restart the epoch changes, so a + verifier tells a **counter reset from records lost** — and `(epoch, + emission_seq)` is a total order across restarts. Detecting loss of the *tail* + of a previous epoch (a crash between emit and persist) is not possible from + the counters alone — that is what a durable sink (an append-only ledger) is + for. + +### Destinations + +| `destination` | Behaviour | +|---|---| +| `stderr` (default) | One JSON line per record to stderr — grep / `jq` / forward. | +| `tracing` | Emitted via `tracing::info!` at target `apl.audit`; the host's subscriber routes it wherever traces go. | + +## Writing a custom audit sink + +The `audit-logger` is one implementation of the `AuditHandler` trait. Write your +own to forward audit to a SIEM, a schema like OCSF, or an internal event bus. +Depend on `cpex-core` for the traits. + +An audit handler is **observation-only** — it is handed the finalized decision +and cannot influence it (its methods return nothing): + +```rust +use std::sync::Arc; +use cpex_core::audit::AuditHandler; +use cpex_core::decision::DecisionLog; +use cpex_core::effect::EffectRecord; +use cpex_core::hooks::payload::{Extensions, PluginPayload}; + +struct MyAuditSink { /* destination handle, config, … */ } + +#[async_trait::async_trait] +impl AuditHandler for MyAuditSink { + // Called once per invocation, at the verdict — for allows, denies, and + // modifies alike. + async fn handle(&self, _payload: &dyn PluginPayload, ext: &Extensions, decisions: &DecisionLog) { + let verdict = decisions.verdict(); // Allow | Deny(violation) + let steps = decisions.steps(); // what each plugin did + let span = decisions.span(); // trace_id / span_id / parent_span_id + let labels = decisions.input_labels(); // taint the request arrived with + let hash = decisions.input_hash(); // Some(..) when provenance is on + let _ = (verdict, steps, span, labels, hash, ext); + // build your event and ship it — do not block or mutate + } + + // Optional: called per irreversible effect (token mint, approval grant). + // Omit it if your sink only cares about decisions. + async fn on_effect(&self, effect: &EffectRecord, _ext: &Extensions) { + let _ = (&effect.kind, &effect.state, &effect.key, &effect.details); + // e.g. emit a token mint as its own event + } + + // Names the sink in error logs if it panics or times out. + fn name(&self) -> &str { "my-audit-sink" } +} +``` + +Attach it in one of two ways: + +**As a builtin plugin (config-driven).** Implement `Plugin` and return the sink +from `as_audit_handler` so the manager auto-attaches it when it is declared with +no hooks (exactly like `audit-logger`): + +```rust +impl cpex_core::plugin::Plugin for MyAuditSink { + fn config(&self) -> &cpex_core::plugin::PluginConfig { &self.cfg } + + fn as_audit_handler(self: Arc) -> Option> { + Some(self) // declare with no hooks → auto-attaches as a decision-audit sink + } +} +``` + +Register a `PluginFactory` for it under a `kind`, and it configures like any +other builtin (see [Builtins]({{< relref "/docs/builtins" >}}) and +[Configuration]({{< relref "/docs/configuration" >}})). + +**Programmatically (embedding).** Attach it directly to a running manager: + +```rust +manager.register_audit_handler(Arc::new(MyAuditSink { /* … */ })); +``` + +Two rules the framework enforces so a sink can never harm a request: it receives +the decision record but it is **never on the plugin context** (a sink cannot see +or change another plugin's state through it), and a sink that panics or exceeds +its timeout is contained and logged — the request, whose verdict is already +decided, proceeds regardless. + +### Sinks run on the request path — keep them cheap + +`handle` and `on_effect` are **awaited** where the verdict (or effect) is +emitted, *before* the request returns — they are **not** fire-and-forget. That +is deliberate, and it is the property that makes the record trustworthy: a crash +cannot lose a verdict that was emitted, so a downstream evidence chain needs no +drop-detection for the steady state and can rely on this ordering. It is a +stable contract — changing it to fire-and-forget would silently break consumers +built on it. + +The cost of that guarantee is that **sink latency is on the request path** +(bounded per sink by the plugin timeout, and sinks run one at a time). So keep +the work in a sink cheap — serialize, hash, append. A sink that does something +slow — a network call to a SIEM, a write to a remote ledger — should hand the +record to an **internal queue and return immediately**, doing the slow work on +its own side of the boundary rather than blocking the request. + +## Troubleshooting + +| Symptom | Cause | +|---|---| +| Denials never appear in the log | The audit plugin has `hooks:` listed → legacy post-hook mode. Remove the hooks. | +| Effect records appear but aren't crash-safe | No `effect_log_path` set → ordering-only mode. | +| A plugin's mints aren't audited at all | The plugin is missing the `emit_effect` capability. | +| No `content` field | `capture_content_provenance` is off, or the payload type isn't hashable. | +| Mints stuck `unknown` after a restart | Expected with the default reconciler — no ledger to confirm them. Investigate, or wire a reconciler. | + +## See also + +- [Builtins]({{< relref "/docs/builtins" >}}) — the `audit/logger` builtin and the others. +- [Configuration]({{< relref "/docs/configuration" >}}) — the full config structure. +- [Extensions & Capability-Gating]({{< relref "/docs/extensions" >}}) — capabilities like `emit_effect`.