diff --git a/Cargo.lock b/Cargo.lock index 3b35e45ea8..30528b4513 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5278,6 +5278,7 @@ dependencies = [ "miette", "openshell-core", "openshell-extension-core", + "openshell-otel", "openshell-supervisor-middleware-builtins", "prost", "prost-types", @@ -5285,6 +5286,7 @@ dependencies = [ "tokio-stream", "tonic", "tracing", + "tracing-subscriber", ] [[package]] diff --git a/crates/openshell-supervisor-middleware/Cargo.toml b/crates/openshell-supervisor-middleware/Cargo.toml index 104f1f41e6..5910bd1a3f 100644 --- a/crates/openshell-supervisor-middleware/Cargo.toml +++ b/crates/openshell-supervisor-middleware/Cargo.toml @@ -13,6 +13,7 @@ rust-version.workspace = true [dependencies] openshell-core = { path = "../openshell-core", default-features = false } openshell-extension-core = { path = "../openshell-extension-core" } +openshell-otel = { path = "../openshell-otel" } miette = { workspace = true } futures = { workspace = true } @@ -26,6 +27,7 @@ tracing = { workspace = true } [dev-dependencies] openshell-supervisor-middleware-builtins = { path = "../openshell-supervisor-middleware-builtins" } tokio-stream = { workspace = true, features = ["net"] } +tracing-subscriber = { workspace = true } [lints] workspace = true diff --git a/crates/openshell-supervisor-middleware/src/lib.rs b/crates/openshell-supervisor-middleware/src/lib.rs index 63e71c869b..361fbaa73f 100644 --- a/crates/openshell-supervisor-middleware/src/lib.rs +++ b/crates/openshell-supervisor-middleware/src/lib.rs @@ -255,6 +255,20 @@ pub enum MiddlewareWorkAdmissionOutcome { QueueExhausted, } +/// Typed load-shedding signal emitted when the bounded middleware queue is full. +#[derive(Debug)] +pub struct MiddlewareAdmissionExhausted; + +impl std::fmt::Display for MiddlewareAdmissionExhausted { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("middleware admission queue is full; refusing additional buffered work") + } +} + +impl std::error::Error for MiddlewareAdmissionExhausted {} + +impl miette::Diagnostic for MiddlewareAdmissionExhausted {} + impl MiddlewareWorkAdmissionOutcome { /// Preserve the existing failure behavior for protocols whose outer layer /// already translates middleware admission errors into a stable response @@ -262,9 +276,7 @@ impl MiddlewareWorkAdmissionOutcome { pub fn into_admission(self) -> Result { match self { Self::Admitted(admission) => Ok(admission), - Self::QueueExhausted => Err(miette!( - "middleware admission queue is full; refusing additional buffered work" - )), + Self::QueueExhausted => Err(MiddlewareAdmissionExhausted.into()), } } } @@ -497,6 +509,18 @@ pub struct ChainOutcome { /// denied the request. Fail-closed service errors and transformed-body /// policy denials are not represented as middleware decisions. pub denial: Option, + /// Typed reason the chain stopped. Consumers must use this instead of + /// inferring security semantics from free-form diagnostic text. + pub termination: ChainTermination, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ChainTermination { + Completed, + MiddlewareDenied, + MiddlewareFailure, + TransformedBodyPolicyDenied, + TransformedBodyPolicyEvaluationFailed, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -1748,6 +1772,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1765,6 +1790,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1791,6 +1817,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1812,6 +1839,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1842,6 +1870,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1861,6 +1890,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1885,6 +1915,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1929,6 +1960,7 @@ impl ChainRunner { metadata, applied, denial: Some(denial), + termination: ChainTermination::MiddlewareDenied, }); } @@ -1945,6 +1977,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -1979,6 +2012,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::MiddlewareFailure, }); } } @@ -2024,12 +2058,15 @@ impl ChainRunner { if body_transformed && let TransformedBodyPolicy::Reevaluate(validate) = transformed_body_policy { - let denied = match validate(&body) { - Ok(reason) => reason, - Err(error) => Some(format!( - "transformed_body_policy_evaluation_failed: {}", - safe_reason(&error.to_string()) - )), + let (denied, termination) = match validate(&body) { + Ok(reason) => (reason, ChainTermination::TransformedBodyPolicyDenied), + Err(error) => ( + Some(format!( + "transformed_body_policy_evaluation_failed: {}", + safe_reason(&error.to_string()) + )), + ChainTermination::TransformedBodyPolicyEvaluationFailed, + ), }; if let Some(reason) = denied { return Ok(ChainOutcome { @@ -2041,6 +2078,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination, }); } } @@ -2055,6 +2093,7 @@ impl ChainRunner { metadata, applied, denial: None, + termination: ChainTermination::Completed, }) } } @@ -6026,10 +6065,15 @@ mod tests { overflow, MiddlewareWorkAdmissionOutcome::QueueExhausted )); - runner + let error = runner .reserve_middleware_work_admission() .await .expect_err("WebSocket callers retain their existing failure path"); + assert!( + error + .downcast_ref::() + .is_some() + ); drop(active); for waiter in waiters { diff --git a/crates/openshell-supervisor-middleware/src/remote.rs b/crates/openshell-supervisor-middleware/src/remote.rs index 78f27b8cf7..727fd0f304 100644 --- a/crates/openshell-supervisor-middleware/src/remote.rs +++ b/crates/openshell-supervisor-middleware/src/remote.rs @@ -23,7 +23,31 @@ use tonic::{Request, Response, Status}; use crate::MIDDLEWARE_GRPC_MESSAGE_BYTES; -type ExtensionChannel = InterceptedService; +type ExtensionChannel = InterceptedService; + +#[derive(Clone, Debug)] +struct MiddlewareClientInterceptor { + bearer: BearerTokenInterceptor, + trace_context: openshell_otel::TraceContextInterceptor, +} + +impl MiddlewareClientInterceptor { + fn new(bearer: BearerTokenInterceptor) -> Self { + Self { + bearer, + trace_context: openshell_otel::TraceContextInterceptor, + } + } +} + +impl tonic::service::Interceptor for MiddlewareClientInterceptor { + fn call(&mut self, mut request: Request<()>) -> std::result::Result, Status> { + request.metadata_mut().remove("traceparent"); + request.metadata_mut().remove("tracestate"); + let request = tonic::service::Interceptor::call(&mut self.bearer, request)?; + tonic::service::Interceptor::call(&mut self.trace_context, request) + } +} /// Adapts the borrowed runtime request contract to the owned protobuf service /// contract only when dispatch crosses a gRPC-shaped boundary. @@ -147,7 +171,8 @@ impl RemoteMiddlewareService { })?; let interceptor = bearer.map_or_else(BearerTokenInterceptor::disabled, |slot| slot.interceptor()); - let channel = InterceptedService::new(channel, interceptor); + let channel = + InterceptedService::new(channel, MiddlewareClientInterceptor::new(interceptor)); Ok(Self { client: SupervisorMiddlewareClient::new(channel.clone()) @@ -178,6 +203,7 @@ impl SupervisorMiddlewareEndpoint for RemoteMiddlewareService { client.validate_config(request).await } + #[tracing::instrument(name = "openshell.middleware.rpc", skip_all, fields(otel.kind = "client", rpc.method = "EvaluateHttpRequest"))] async fn evaluate_http_request( &self, request: Request, @@ -186,6 +212,7 @@ impl SupervisorMiddlewareEndpoint for RemoteMiddlewareService { client.evaluate_http_request(request).await } + #[tracing::instrument(name = "openshell.middleware.rpc", skip_all, fields(otel.kind = "client", rpc.method = "EvaluateWebSocketSession"))] async fn open_websocket_session( &self, receiver: tokio::sync::mpsc::Receiver, @@ -200,6 +227,7 @@ impl SupervisorMiddlewareEndpoint for RemoteMiddlewareService { Ok(Box::pin(responses)) } + #[tracing::instrument(name = "openshell.middleware.rpc", skip_all, fields(otel.kind = "client", rpc.method = "HttpResponsePreReturn"))] async fn open_http_response_pre_return( &self, receiver: tokio::sync::mpsc::Receiver, @@ -214,3 +242,56 @@ impl SupervisorMiddlewareEndpoint for RemoteMiddlewareService { Ok(Box::pin(responses)) } } + +#[cfg(test)] +mod tests { + use tonic::service::Interceptor as _; + use tracing_subscriber::layer::SubscriberExt as _; + + use super::*; + + #[test] + fn client_interceptor_replaces_untrusted_trace_context_and_keeps_bearer() { + let slot = BearerTokenSlot::new("middleware-secret", i64::MAX).unwrap(); + let mut interceptor = MiddlewareClientInterceptor::new(slot.interceptor()); + let forged = "00-11111111111111111111111111111111-2222222222222222-01"; + let provider = openshell_otel::SdkTracerProvider::builder().build(); + let subscriber = tracing_subscriber::registry() + .with(openshell_otel::layer(&provider, "middleware-client-test")); + let (request, expected_traceparent) = tracing::subscriber::with_default(subscriber, || { + let span = tracing::info_span!("middleware.client"); + let _entered = span.enter(); + let expected_traceparent = openshell_otel::current_trace_context_carrier() + .and_then(|carrier| carrier.get("traceparent").cloned()) + .expect("active host span has trace context"); + let mut request = Request::new(()); + request + .metadata_mut() + .insert("traceparent", forged.parse().unwrap()); + request + .metadata_mut() + .insert("tracestate", "vendor=untrusted".parse().unwrap()); + (interceptor.call(request).unwrap(), expected_traceparent) + }); + + assert_eq!( + request.metadata().get("authorization").unwrap(), + "Bearer middleware-secret" + ); + let traceparent = request + .metadata() + .get("traceparent") + .expect("active host span is propagated") + .to_str() + .unwrap(); + assert_ne!(traceparent, forged); + assert_eq!(traceparent, expected_traceparent); + assert_ne!( + request + .metadata() + .get("tracestate") + .and_then(|value| value.to_str().ok()), + Some("vendor=untrusted") + ); + } +} diff --git a/crates/openshell-supervisor-middleware/src/response.rs b/crates/openshell-supervisor-middleware/src/response.rs index 57e1217fcf..86ce967068 100644 --- a/crates/openshell-supervisor-middleware/src/response.rs +++ b/crates/openshell-supervisor-middleware/src/response.rs @@ -13,7 +13,7 @@ use validation::{ validate_body_result, validate_trailers_result, }; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::time::Duration; use futures::StreamExt as _; @@ -31,9 +31,10 @@ use openshell_core::proto::{ }; use super::{ - ChainEntry, ChainRunner, DescribedChainEntry, MAX_MIDDLEWARE_CHAIN_TIMEOUT, - MAX_MIDDLEWARE_CONTEXT_BYTES, MAX_MIDDLEWARE_FINDING_BYTES, MAX_MIDDLEWARE_FINDINGS_PER_STAGE, - MAX_MIDDLEWARE_HEADER_BYTES, MAX_MIDDLEWARE_HEADER_MUTATION_WIRE_BYTES, MAX_MIDDLEWARE_HEADERS, + ChainEntry, ChainRunner, DescribedChainEntry, MAX_MIDDLEWARE_CHAIN_FINDINGS, + MAX_MIDDLEWARE_CHAIN_STAGES, MAX_MIDDLEWARE_CHAIN_TIMEOUT, MAX_MIDDLEWARE_CONTEXT_BYTES, + MAX_MIDDLEWARE_FINDING_BYTES, MAX_MIDDLEWARE_FINDINGS_PER_STAGE, MAX_MIDDLEWARE_HEADER_BYTES, + MAX_MIDDLEWARE_HEADER_MUTATION_WIRE_BYTES, MAX_MIDDLEWARE_HEADERS, MAX_MIDDLEWARE_METADATA_BYTES, MAX_MIDDLEWARE_METADATA_ENTRIES, MAX_MIDDLEWARE_REASON_BYTES, MAX_MIDDLEWARE_REASON_CODE_BYTES, MAX_MIDDLEWARE_TARGET_BYTES, MiddlewareDiagnosticPolicy, MiddlewareSessionAdmission, MiddlewareSessionPermit, NamespacedFinding, OnError, headers, @@ -100,12 +101,20 @@ pub struct HttpResponseInvocation { pub sequence: Option, pub input_size: usize, pub output_size: Option, + pub transformed: bool, pub failed: bool, pub stage_disabled: bool, pub reason_code: Option, pub failure_category: Option, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +enum HttpResponseMiddlewareFailureKind { + #[default] + Processing, + AdmissionExhausted, +} + pub struct HttpResponsePreflightOutcome { pub allowed: bool, pub reason: String, @@ -122,6 +131,7 @@ pub struct HttpResponsePreflightOutcome { pub struct HttpResponseMiddlewareFailure { pub reason: String, pub denial: Option, + kind: HttpResponseMiddlewareFailureKind, /// Exchange diagnostics collected before a consuming operation failed. pub diagnostics: HttpResponseDiagnostics, } @@ -139,6 +149,11 @@ impl HttpResponseMiddlewareFailure { self.diagnostics = diagnostics; self } + + #[must_use] + pub fn admission_exhausted(&self) -> bool { + self.kind == HttpResponseMiddlewareFailureKind::AdmissionExhausted + } } #[derive(Debug)] @@ -152,6 +167,7 @@ pub struct HttpResponseFinish { pub findings: Vec, pub metadata: BTreeMap>, pub invocations: Vec, + pub diagnostics_truncated: bool, } #[derive(Debug, Default)] @@ -159,6 +175,104 @@ pub struct HttpResponseDiagnostics { pub findings: Vec, pub metadata: BTreeMap>, pub invocations: Vec, + pub diagnostics_truncated: bool, +} + +#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)] +struct FindingKey { + middleware: String, + r#type: String, + label: String, + count: u32, + confidence: String, + severity: String, +} + +#[derive(Debug, Default)] +struct HttpResponseDiagnosticBudget { + findings: BTreeSet, + finding_count: usize, + routine_invocations: usize, + terminal_invocations: usize, + truncation_pending: bool, + truncation_reported: bool, +} + +impl HttpResponseDiagnosticBudget { + fn from_preflight(findings: &[NamespacedFinding]) -> Self { + let mut budget = Self { + finding_count: findings.len(), + ..Self::default() + }; + budget + .findings + .extend(findings.iter().map(FindingKey::from)); + budget + } + + fn retain_finding(&mut self, finding: &NamespacedFinding) -> bool { + let key = FindingKey::from(finding); + if self.findings.contains(&key) { + return false; + } + if self.finding_count >= MAX_MIDDLEWARE_CHAIN_FINDINGS { + self.mark_truncated(); + return false; + } + self.findings.insert(key); + self.finding_count += 1; + true + } + + fn retain_invocation(&mut self, invocation: &HttpResponseInvocation) -> bool { + let routine = matches!( + invocation.outcome, + HttpResponseInvocationOutcome::PassThrough | HttpResponseInvocationOutcome::Transform + ); + let (count, limit) = if routine { + (self.routine_invocations, MAX_MIDDLEWARE_CHAIN_FINDINGS) + } else { + (self.terminal_invocations, MAX_MIDDLEWARE_CHAIN_STAGES) + }; + if count >= limit { + self.mark_truncated(); + return false; + } + if routine { + self.routine_invocations += 1; + } else { + self.terminal_invocations += 1; + } + true + } + + fn mark_truncated(&mut self) { + if !self.truncation_reported { + self.truncation_pending = true; + } + } + + fn take_truncation_signal(&mut self) -> bool { + if !self.truncation_pending || self.truncation_reported { + return false; + } + self.truncation_pending = false; + self.truncation_reported = true; + true + } +} + +impl From<&NamespacedFinding> for FindingKey { + fn from(finding: &NamespacedFinding) -> Self { + Self { + middleware: finding.middleware.clone(), + r#type: finding.finding.r#type.clone(), + label: finding.finding.label.clone(), + count: finding.finding.count, + confidence: finding.finding.confidence.clone(), + severity: finding.finding.severity.clone(), + } + } } struct HttpResponseStageTransport { @@ -228,6 +342,7 @@ pub struct HttpResponseSession { findings: Vec, metadata: BTreeMap>, invocations: Vec, + diagnostic_budget: HttpResponseDiagnosticBudget, session_admission: Option, body_transformed: bool, retained_body_bytes: usize, @@ -243,6 +358,13 @@ impl HttpResponseSession { findings: std::mem::take(&mut self.findings), metadata: std::mem::take(&mut self.metadata), invocations: std::mem::take(&mut self.invocations), + diagnostics_truncated: self.diagnostic_budget.take_truncation_signal(), + } + } + + fn record_invocation(&mut self, invocation: HttpResponseInvocation) { + if self.diagnostic_budget.retain_invocation(&invocation) { + self.invocations.push(invocation); } } @@ -325,6 +447,7 @@ impl HttpResponseSession { return Err(HttpResponseMiddlewareFailure { reason: "response_stream_unit_over_capacity".into(), denial: None, + kind: HttpResponseMiddlewareFailureKind::Processing, diagnostics: HttpResponseDiagnostics::default(), }); } @@ -335,6 +458,14 @@ impl HttpResponseSession { .map_err(|error| HttpResponseMiddlewareFailure { reason: format!("middleware_failed: {error}"), denial: None, + kind: if error + .downcast_ref::() + .is_some() + { + HttpResponseMiddlewareFailureKind::AdmissionExhausted + } else { + HttpResponseMiddlewareFailureKind::Processing + }, diagnostics: HttpResponseDiagnostics::default(), })?; let deadline = Instant::now() + MAX_MIDDLEWARE_CHAIN_TIMEOUT; @@ -373,6 +504,14 @@ impl HttpResponseSession { return Err(HttpResponseMiddlewareFailure { reason: format!("middleware_failed: {error}"), denial: None, + kind: if error + .downcast_ref::() + .is_some() + { + HttpResponseMiddlewareFailureKind::AdmissionExhausted + } else { + HttpResponseMiddlewareFailureKind::Processing + }, diagnostics: self.take_diagnostics(), }); } @@ -417,6 +556,7 @@ impl HttpResponseSession { }; self.end_all(MiddlewareSessionEndReason::Normal).await; self.session_admission.take(); + let diagnostics_truncated = self.diagnostic_budget.take_truncation_signal(); Ok(HttpResponseFinish { body_units: released, trailers, @@ -424,6 +564,7 @@ impl HttpResponseSession { findings: self.findings, metadata: self.metadata, invocations: self.invocations, + diagnostics_truncated, }) } @@ -617,26 +758,28 @@ impl HttpResponseSession { } self.retained_body_bytes = retained; } - let stage = &mut self.stages[index]; - collect_diagnostics( - stage, + collect_session_diagnostics( + &self.stages[index], decision.findings, decision.metadata, &mut self.findings, &mut self.metadata, + &mut self.diagnostic_budget, ); let reason_code = (!decision.reason_code.is_empty()).then_some(decision.reason_code); match decision.action { BodyAction::PassThrough => { let output_size = original.len(); - self.invocations.push(body_invocation_with_reason( - stage, + let invocation = body_invocation_with_reason( + &self.stages[index], HttpResponseInvocationOutcome::PassThrough, sequence, input_size, output_size, + false, reason_code, - )); + ); + self.record_invocation(invocation); Ok((!original.is_empty()) .then_some(original) .into_iter() @@ -644,20 +787,23 @@ impl HttpResponseSession { } BodyAction::Transform(replacement) => { self.body_transformed = true; - self.invocations.push(body_invocation_with_reason( - stage, + let invocation = body_invocation_with_reason( + &self.stages[index], HttpResponseInvocationOutcome::Transform, sequence, input_size, replacement.len(), + true, reason_code, - )); + ); + self.record_invocation(invocation); Ok((!replacement.is_empty()) .then_some(replacement) .into_iter() .collect()) } BodyAction::SkipRemaining(action) => { + let transformed = matches!(&action, CurrentBodyAction::Transform(_)); let output = match action { CurrentBodyAction::PassThrough => original, CurrentBodyAction::Transform(replacement) => { @@ -665,30 +811,39 @@ impl HttpResponseSession { replacement } }; - stage.mode = StageMode::HeadersOnly; - self.invocations.push(body_invocation_with_reason( - stage, - HttpResponseInvocationOutcome::SkipRemaining, - sequence, - input_size, - output.len(), - reason_code, - )); - stage.end(MiddlewareSessionEndReason::Normal).await; + let invocation = { + let stage = &mut self.stages[index]; + stage.mode = StageMode::HeadersOnly; + let invocation = body_invocation_with_reason( + stage, + HttpResponseInvocationOutcome::SkipRemaining, + sequence, + input_size, + output.len(), + transformed, + reason_code, + ); + stage.end(MiddlewareSessionEndReason::Normal).await; + invocation + }; + self.record_invocation(invocation); self.release_admission_if_idle(); Ok((!output.is_empty()).then_some(output).into_iter().collect()) } BodyAction::BlockDelivery => { + let stage = &self.stages[index]; let config_name = stage.entry.entry.name.clone(); let denial_reason = middleware_denial_reason(&config_name, reason_code.as_deref()); - self.invocations.push(body_invocation_with_reason( + let invocation = body_invocation_with_reason( stage, HttpResponseInvocationOutcome::BlockDelivery, sequence, input_size, 0, + false, reason_code.clone(), - )); + ); + self.record_invocation(invocation); self.end_all(MiddlewareSessionEndReason::MiddlewareDenial) .await; Err(HttpResponseMiddlewareFailure { @@ -697,6 +852,7 @@ impl HttpResponseSession { config_name, reason_code, }), + kind: HttpResponseMiddlewareFailureKind::Processing, diagnostics: HttpResponseDiagnostics::default(), }) } @@ -710,28 +866,29 @@ impl HttpResponseSession { sequence: Option, original: Vec, ) -> Result>, HttpResponseMiddlewareFailure> { - let stage = &mut self.stages[index]; - let fail_open = stage.entry.on_error() == OnError::FailOpen; + let fail_open = self.stages[index].entry.on_error() == OnError::FailOpen; let outcome = if fail_open { HttpResponseInvocationOutcome::FailOpen } else { HttpResponseInvocationOutcome::FailClosed }; - self.invocations.push(HttpResponseInvocation { - config_name: stage.entry.entry.name.clone(), - implementation: stage.entry.entry.implementation.clone(), + let invocation = HttpResponseInvocation { + config_name: self.stages[index].entry.entry.name.clone(), + implementation: self.stages[index].entry.entry.implementation.clone(), outcome, sequence, input_size: original.len(), output_size: None, + transformed: false, failed: true, stage_disabled: true, reason_code: None, failure_category: Some(response_failure_category(reason).into()), - }); - stage + }; + self.stages[index] .end(MiddlewareSessionEndReason::MiddlewareFailure) .await; + self.record_invocation(invocation); self.release_admission_if_idle(); if fail_open { if original.is_empty() { @@ -743,6 +900,7 @@ impl HttpResponseSession { Err(HttpResponseMiddlewareFailure { reason: format!("middleware_failed: {reason}"), denial: None, + kind: HttpResponseMiddlewareFailureKind::Processing, diagnostics: HttpResponseDiagnostics::default(), }) } @@ -818,26 +976,28 @@ impl HttpResponseSession { trailers = decision.headers; let output_size = encoded_header_bytes(&trailers); let reason_code = (!decision.reason_code.is_empty()).then_some(decision.reason_code); - let stage = &mut self.stages[index]; - collect_diagnostics( - stage, + collect_session_diagnostics( + &self.stages[index], decision.findings, decision.metadata, &mut self.findings, &mut self.metadata, + &mut self.diagnostic_budget, ); - self.invocations.push(HttpResponseInvocation { - config_name: stage.entry.entry.name.clone(), - implementation: stage.entry.entry.implementation.clone(), + let invocation = HttpResponseInvocation { + config_name: self.stages[index].entry.entry.name.clone(), + implementation: self.stages[index].entry.entry.implementation.clone(), outcome: HttpResponseInvocationOutcome::Trailers, sequence: None, input_size, output_size: Some(output_size), + transformed: false, failed: false, stage_disabled: false, reason_code, failure_category: None, - }); + }; + self.record_invocation(invocation); } Ok(trailers) } @@ -848,11 +1008,10 @@ impl HttpResponseSession { reason: &str, original: Vec, ) -> Result, HttpResponseMiddlewareFailure> { - let stage = &mut self.stages[index]; - let fail_open = stage.entry.on_error() == OnError::FailOpen; - self.invocations.push(HttpResponseInvocation { - config_name: stage.entry.entry.name.clone(), - implementation: stage.entry.entry.implementation.clone(), + let fail_open = self.stages[index].entry.on_error() == OnError::FailOpen; + let invocation = HttpResponseInvocation { + config_name: self.stages[index].entry.entry.name.clone(), + implementation: self.stages[index].entry.entry.implementation.clone(), outcome: if fail_open { HttpResponseInvocationOutcome::FailOpen } else { @@ -861,14 +1020,16 @@ impl HttpResponseSession { sequence: None, input_size: encoded_header_bytes(&original), output_size: None, + transformed: false, failed: true, stage_disabled: true, reason_code: None, failure_category: Some(response_failure_category(reason).into()), - }); - stage + }; + self.stages[index] .end(MiddlewareSessionEndReason::MiddlewareFailure) .await; + self.record_invocation(invocation); self.release_admission_if_idle(); if fail_open { Ok(original) @@ -876,6 +1037,7 @@ impl HttpResponseSession { Err(HttpResponseMiddlewareFailure { reason: format!("middleware_failed: {reason}"), denial: None, + kind: HttpResponseMiddlewareFailureKind::Processing, diagnostics: HttpResponseDiagnostics::default(), }) } @@ -951,6 +1113,7 @@ fn body_invocation_with_reason( sequence: u64, input_size: usize, output_size: usize, + transformed: bool, reason_code: Option, ) -> HttpResponseInvocation { HttpResponseInvocation { @@ -960,6 +1123,7 @@ fn body_invocation_with_reason( sequence: Some(sequence), input_size, output_size: Some(output_size), + transformed, failed: false, stage_disabled: false, reason_code, @@ -1000,6 +1164,29 @@ fn collect_diagnostics( } } +fn collect_session_diagnostics( + stage: &HttpResponseStage, + findings: Vec, + metadata: std::collections::HashMap, + all_findings: &mut Vec, + all_metadata: &mut BTreeMap>, + budget: &mut HttpResponseDiagnosticBudget, +) { + let mut current_findings = Vec::new(); + collect_diagnostics( + stage, + findings, + metadata, + &mut current_findings, + all_metadata, + ); + all_findings.extend( + current_findings + .into_iter() + .filter(|finding| budget.retain_finding(finding)), + ); +} + fn collect_preflight_diagnostics( entry: &DescribedChainEntry, findings: Vec, @@ -1034,6 +1221,7 @@ fn collect_preflight_failure( sequence: None, input_size: 0, output_size: None, + transformed: false, failed: true, stage_disabled: true, reason_code: None, @@ -1224,12 +1412,23 @@ mod tests { InvalidSkipReason, TrailerMutation, InvalidTrailerMutation, + DiagnosticBudget, } struct ResponseService { script: Script, } + fn diagnostic_finding(index: usize) -> Finding { + Finding { + r#type: format!("diagnostic-{index}"), + label: "bounded diagnostic".into(), + count: 1, + confidence: "high".into(), + severity: "medium".into(), + } + } + struct PreflightLifecycleService { completion_tx: mpsc::UnboundedSender<(String, Vec)>, } @@ -1485,7 +1684,8 @@ mod tests { | Script::DeleteBody | Script::SkipBody | Script::TrailerMutation - | Script::InvalidTrailerMutation => { + | Script::InvalidTrailerMutation + | Script::DiagnosticBudget => { (HttpResponseBodyMode::StreamBytes, Vec::new()) } Script::WholeBody => { @@ -1507,6 +1707,13 @@ mod tests { }, ), ), + findings: matches!( + selected_script, + Script::DiagnosticBudget + ) + .then(|| diagnostic_finding(0)) + .into_iter() + .collect(), ..Default::default() }, ), @@ -1522,6 +1729,11 @@ mod tests { else { break; }; + let diagnostic_sequence = usize::try_from(body.sequence) + .expect("test sequence fits in usize"); + let diagnostic_budget_exhausted = + matches!(selected_script, Script::DiagnosticBudget) + && diagnostic_sequence > MAX_MIDDLEWARE_CHAIN_FINDINGS; let replacement = match selected_script { Script::Expansion => vec![b'x'; 128 * 1024], Script::DeleteBody => Vec::new(), @@ -1530,7 +1742,8 @@ mod tests { | Script::InvalidSequence | Script::LargeStream | Script::TrailerMutation - | Script::InvalidTrailerMutation => data.to_ascii_uppercase(), + | Script::InvalidTrailerMutation + | Script::DiagnosticBudget => data.to_ascii_uppercase(), Script::WholeBody => [b"whole:".as_slice(), &data].concat(), Script::HeadersOnly | Script::Configured @@ -1553,6 +1766,14 @@ mod tests { ), }, ) + } else if diagnostic_budget_exhausted { + http_response_body_result::Action::BlockDelivery( + openshell_core::proto::HttpResponseBlockDelivery {}, + ) + } else if matches!(selected_script, Script::DiagnosticBudget) { + http_response_body_result::Action::PassThrough( + HttpResponseBodyPassThrough {}, + ) } else { http_response_body_result::Action::Transform(transform) }; @@ -1568,6 +1789,18 @@ mod tests { body.sequence }, action: Some(action), + findings: matches!( + selected_script, + Script::DiagnosticBudget + ) + .then(|| diagnostic_finding(diagnostic_sequence - 1)) + .into_iter() + .collect(), + reason_code: if diagnostic_budget_exhausted { + "diagnostic_budget_block".into() + } else { + String::new() + }, ..Default::default() }, )), @@ -2726,4 +2959,136 @@ mod tests { .expect("join response middleware server") .expect("serve response middleware"); } + + fn budget_finding(index: usize) -> NamespacedFinding { + NamespacedFinding { + middleware: "response-filter".into(), + finding: Finding { + r#type: format!("finding-{index}"), + label: "bounded finding".into(), + count: 1, + confidence: "high".into(), + severity: "medium".into(), + }, + } + } + + fn budget_invocation(outcome: HttpResponseInvocationOutcome) -> HttpResponseInvocation { + HttpResponseInvocation { + config_name: "response-filter".into(), + implementation: "test/response-filter".into(), + outcome, + sequence: Some(1), + input_size: 1, + output_size: Some(1), + transformed: false, + failed: false, + stage_disabled: false, + reason_code: None, + failure_category: None, + } + } + + #[test] + fn response_diagnostic_budget_deduplicates_findings_and_signals_once() { + let mut budget = HttpResponseDiagnosticBudget::default(); + let finding = budget_finding(0); + + assert!(budget.retain_finding(&finding)); + assert!(!budget.retain_finding(&finding)); + assert!(!budget.take_truncation_signal()); + assert!(budget.retain_finding(&budget_finding(1))); + assert_eq!(budget.finding_count, 2); + } + + #[test] + fn response_diagnostic_budget_includes_preflight_findings() { + let preflight = (0..MAX_MIDDLEWARE_CHAIN_FINDINGS) + .map(budget_finding) + .collect::>(); + let mut budget = HttpResponseDiagnosticBudget::from_preflight(&preflight); + + assert!(!budget.retain_finding(&preflight[0])); + assert!(!budget.take_truncation_signal()); + assert!(!budget.retain_finding(&budget_finding(MAX_MIDDLEWARE_CHAIN_FINDINGS))); + assert!(budget.take_truncation_signal()); + assert!(!budget.take_truncation_signal()); + } + + #[test] + fn response_diagnostic_budget_caps_unique_findings() { + let mut budget = HttpResponseDiagnosticBudget::default(); + for index in 0..MAX_MIDDLEWARE_CHAIN_FINDINGS { + assert!(budget.retain_finding(&budget_finding(index))); + } + + assert!(!budget.retain_finding(&budget_finding(MAX_MIDDLEWARE_CHAIN_FINDINGS))); + assert_eq!(budget.finding_count, MAX_MIDDLEWARE_CHAIN_FINDINGS); + assert!(budget.take_truncation_signal()); + } + + #[test] + fn response_diagnostic_budget_preserves_terminal_invocations() { + let mut budget = HttpResponseDiagnosticBudget::default(); + let routine = budget_invocation(HttpResponseInvocationOutcome::PassThrough); + for _ in 0..MAX_MIDDLEWARE_CHAIN_FINDINGS { + assert!(budget.retain_invocation(&routine)); + } + assert!(!budget.retain_invocation(&routine)); + + let terminal = budget_invocation(HttpResponseInvocationOutcome::BlockDelivery); + for _ in 0..MAX_MIDDLEWARE_CHAIN_STAGES { + assert!(budget.retain_invocation(&terminal)); + } + assert!(!budget.retain_invocation(&terminal)); + assert!(budget.take_truncation_signal()); + } + + #[tokio::test] + async fn response_session_bounds_diagnostics_and_preserves_late_block() { + let runner = ChainRunner::new(Arc::new(ResponseService { + script: Script::DiagnosticBudget, + })); + let mut preflight = runner + .preflight_http_response(&[entry(OnError::FailClosed)], input(200)) + .await + .expect("preflight response middleware"); + assert_eq!(preflight.findings.len(), 1); + let mut emitted_findings = preflight.findings.len(); + let mut emitted_routine_invocations = 0; + let mut truncation_signals = 0; + let mut session = preflight.session.take().expect("streaming session"); + + for sequence in 1..=MAX_MIDDLEWARE_CHAIN_FINDINGS + 1 { + match session.push_body(vec![b'x']).await { + Ok(body) => { + assert_eq!(body, vec![vec![b'x']]); + let diagnostics = session.take_diagnostics(); + emitted_findings += diagnostics.findings.len(); + emitted_routine_invocations += diagnostics.invocations.len(); + truncation_signals += usize::from(diagnostics.diagnostics_truncated); + } + Err(failure) => { + assert_eq!(sequence, MAX_MIDDLEWARE_CHAIN_FINDINGS + 1); + assert!(failure.denial.is_some()); + let diagnostics = session.take_diagnostics(); + emitted_findings += diagnostics.findings.len(); + truncation_signals += usize::from(diagnostics.diagnostics_truncated); + assert_eq!(diagnostics.invocations.len(), 1); + assert_eq!( + diagnostics.invocations[0].outcome, + HttpResponseInvocationOutcome::BlockDelivery + ); + } + } + } + + assert_eq!(emitted_findings, MAX_MIDDLEWARE_CHAIN_FINDINGS); + assert_eq!(emitted_routine_invocations, MAX_MIDDLEWARE_CHAIN_FINDINGS); + assert_eq!(truncation_signals, 1); + assert!(!session.take_diagnostics().diagnostics_truncated); + session + .end(MiddlewareSessionEndReason::MiddlewareDenial) + .await; + } } diff --git a/crates/openshell-supervisor-middleware/src/response/preflight.rs b/crates/openshell-supervisor-middleware/src/response/preflight.rs index e2e53e9bbe..95ea7ed7f1 100644 --- a/crates/openshell-supervisor-middleware/src/response/preflight.rs +++ b/crates/openshell-supervisor-middleware/src/response/preflight.rs @@ -227,6 +227,7 @@ impl ChainRunner { sequence: None, input_size: 0, output_size: None, + transformed: false, failed: false, stage_disabled: false, reason_code, @@ -293,6 +294,7 @@ impl ChainRunner { continue; } }; + let transformed = updated != headers; headers = updated; if mode == StageMode::Stream { strip_stale_integrity(&mut headers); @@ -315,6 +317,7 @@ impl ChainRunner { sequence: None, input_size: 0, output_size: None, + transformed, failed: false, stage_disabled: false, reason_code, @@ -342,6 +345,7 @@ impl ChainRunner { sequence: None, input_size: 0, output_size: None, + transformed: false, failed: false, stage_disabled: false, reason_code: reason_code.clone(), @@ -399,6 +403,7 @@ impl ChainRunner { let defer_output_until_finish = stages .iter() .any(|stage| stage.is_active() && stage.mode == StageMode::WholeBody); + let diagnostic_budget = HttpResponseDiagnosticBudget::from_preflight(&findings); Ok(HttpResponsePreflightOutcome { allowed: true, reason: String::new(), @@ -410,6 +415,7 @@ impl ChainRunner { findings: Vec::new(), metadata: BTreeMap::new(), invocations: Vec::new(), + diagnostic_budget, session_admission: Some(session_admission), body_transformed: false, retained_body_bytes: 0, diff --git a/crates/openshell-supervisor-network/README.md b/crates/openshell-supervisor-network/README.md index cd39c07c0a..29b183dfd3 100644 --- a/crates/openshell-supervisor-network/README.md +++ b/crates/openshell-supervisor-network/README.md @@ -83,3 +83,20 @@ the final head, a guard emits it with the available outcome instead. The byte observer does not buffer headers or alter transport behavior. A receipt confirms only observation of a response head, not completion of its body. Process crashes and requests that remain hung cannot be finalized by the guard. + +When selected, request and response middleware have separate child spans. Routine +invocations stay in sampled traces; security findings and notable failures remain +in OCSF output, linked by the same request ID. Each phase closes when its work +ends rather than waiting for the entire request. Cancelling an unfinished phase +marks its span as incomplete without inventing a middleware verdict. +Streaming observations record flag transitions, not one attribute update per +body unit, so long responses retain room for their final verdict. + +Middleware RPCs propagate the supervisor's current trace context, replacing +caller-supplied trace metadata without changing bearer authentication. Response +diagnostics have a session-wide budget; exceeding it preserves enforcement and +emits a bounded truncation finding. + +The middleware crate's Rust `ChainOutcome` now includes a typed termination +reason. Rust consumers constructing that result must supply the new field; +middleware protobuf messages and RPC signatures are unchanged. diff --git a/crates/openshell-supervisor-network/src/l7/correlation.rs b/crates/openshell-supervisor-network/src/l7/correlation.rs index 5df3fe338c..d723f8e738 100644 --- a/crates/openshell-supervisor-network/src/l7/correlation.rs +++ b/crates/openshell-supervisor-network/src/l7/correlation.rs @@ -5,6 +5,7 @@ use std::sync::Mutex; use std::{ + future::Future, pin::Pin, task::{Context, Poll}, }; @@ -14,6 +15,7 @@ use openshell_ocsf::{ ActionId, Endpoint, HttpActivityBuilder, HttpRequest, HttpResponse, OcsfEvent, SeverityId, StatusId, }; +use tracing::Instrument as _; use crate::l7::provider::L7Request; use crate::l7::relay::L7EvalContext; @@ -34,6 +36,95 @@ struct RequestState { upstream_status: Option, outcome: Option<&'static str>, policy_span: Option, + middleware: [Option; 2], +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MiddlewarePhase { + Request, + Response, +} + +impl MiddlewarePhase { + const fn index(self) -> usize { + match self { + Self::Request => 0, + Self::Response => 1, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MiddlewareOutcome { + Allowed, + Denied, + PlatformError, +} + +impl MiddlewareOutcome { + const fn as_str(self) -> &'static str { + match self { + Self::Allowed => "allowed", + Self::Denied => "denied", + Self::PlatformError => "platform_error", + } + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[allow(clippy::struct_excessive_bools)] +pub struct MiddlewareObservation { + pub mutated: bool, + pub fail_open: bool, + pub fail_closed: bool, + pub admission_exhausted: bool, + pub diagnostics_truncated: bool, + pub delivery_started: bool, + pub partial_delivery: bool, +} + +#[derive(Debug)] +struct MiddlewareState { + span: Option, + observation: MiddlewareObservation, + completed: bool, +} + +#[must_use] +pub struct MiddlewarePhaseGuard<'a> { + correlation: &'a RequestCorrelation, + phase: MiddlewarePhase, +} + +impl Drop for MiddlewarePhaseGuard<'_> { + fn drop(&mut self) { + let span = { + let mut state = self + .correlation + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(middleware) = &mut state.middleware[self.phase.index()] else { + return; + }; + let span = middleware.span.take(); + if !middleware.completed + && let Some(span) = &span + { + openshell_otel::mark_error(span); + } + span + }; + if let Some(span) = span { + close_span(span); + } + } +} + +fn close_span(span: tracing::Span) { + if let Some(dispatch) = span.with_subscriber(|(_, dispatch)| dispatch.clone()) { + tracing::dispatcher::with_default(&dispatch, || drop(span)); + } } impl RequestCorrelation { @@ -86,6 +177,7 @@ impl RequestCorrelation { upstream_status: None, outcome: None, policy_span: None, + middleware: [None, None], })), } } @@ -103,6 +195,21 @@ impl RequestCorrelation { } } + pub(crate) async fn within_span(&self, future: F) -> F::Output + where + F: Future, + { + future.instrument(self.span.clone()).await + } + + pub(crate) async fn within_middleware( + &self, + phase: MiddlewarePhase, + future: F, + ) -> F::Output { + future.instrument(self.middleware_span(phase)).await + } + pub(crate) fn record_policy_decision(&self, decision: &str) { self.span.record("openshell.policy.decision", decision); let policy_span = self @@ -165,6 +272,184 @@ impl RequestCorrelation { openshell_otel::mark_error(&self.span); } + pub(crate) fn middleware_span(&self, phase: MiddlewarePhase) -> tracing::Span { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.middleware[phase.index()] + .get_or_insert_with(|| MiddlewareState { + span: Some( + self.span + .with_subscriber(|(_, dispatch)| { + tracing::dispatcher::with_default(dispatch, || match phase { + MiddlewarePhase::Request => tracing::info_span!(parent: &self.span, + "openshell.middleware.request", otel.status_code = tracing::field::Empty, + "openshell.middleware.outcome" = "incomplete", + "openshell.middleware.mutated" = false, + "openshell.middleware.fail_open" = false, + "openshell.middleware.fail_closed" = false, + "openshell.middleware.admission_exhausted" = false, + "openshell.middleware.diagnostics_truncated" = false, + "openshell.response.delivery_started" = false, + "openshell.response.partial_delivery" = false), + MiddlewarePhase::Response => { + tracing::info_span!(parent: &self.span, + "openshell.middleware.response", otel.status_code = tracing::field::Empty, + "openshell.middleware.outcome" = "incomplete", + "openshell.middleware.mutated" = false, + "openshell.middleware.fail_open" = false, + "openshell.middleware.fail_closed" = false, + "openshell.middleware.admission_exhausted" = false, + "openshell.middleware.diagnostics_truncated" = false, + "openshell.response.delivery_started" = false, + "openshell.response.partial_delivery" = false) + } + }) + }) + .unwrap_or_else(tracing::Span::none), + ), + observation: MiddlewareObservation::default(), + completed: false, + }) + .span + .clone() + .unwrap_or_else(tracing::Span::none) + } + + pub(crate) fn begin_middleware(&self, phase: MiddlewarePhase) -> MiddlewarePhaseGuard<'_> { + drop(self.middleware_span(phase)); + MiddlewarePhaseGuard { + correlation: self, + phase, + } + } + + pub(crate) fn observe_middleware( + &self, + phase: MiddlewarePhase, + observation: MiddlewareObservation, + ) { + let span = self.middleware_span(phase); + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(middleware) = &mut state.middleware[phase.index()] else { + return; + }; + if middleware.completed { + return; + } + let merged = &mut middleware.observation; + let previous = *merged; + merged.mutated |= observation.mutated; + merged.fail_open |= observation.fail_open; + merged.fail_closed |= observation.fail_closed; + merged.admission_exhausted |= observation.admission_exhausted; + merged.diagnostics_truncated |= observation.diagnostics_truncated; + merged.delivery_started |= observation.delivery_started; + merged.partial_delivery |= observation.partial_delivery; + // Repeated SDK attribute writes consume the span limit, even for the same key. + for (field, before, after) in [ + ( + "openshell.middleware.mutated", + previous.mutated, + merged.mutated, + ), + ( + "openshell.middleware.fail_open", + previous.fail_open, + merged.fail_open, + ), + ( + "openshell.middleware.fail_closed", + previous.fail_closed, + merged.fail_closed, + ), + ( + "openshell.middleware.admission_exhausted", + previous.admission_exhausted, + merged.admission_exhausted, + ), + ( + "openshell.middleware.diagnostics_truncated", + previous.diagnostics_truncated, + merged.diagnostics_truncated, + ), + ( + "openshell.response.delivery_started", + previous.delivery_started, + merged.delivery_started, + ), + ( + "openshell.response.partial_delivery", + previous.partial_delivery, + merged.partial_delivery, + ), + ] { + if before != after { + span.record(field, after); + } + } + if let Some(primary) = &mut state.primary { + for (key, value) in [ + ("middleware_fail_open", observation.fail_open), + ("middleware_fail_closed", observation.fail_closed), + ( + "middleware_admission_exhausted", + observation.admission_exhausted, + ), + ] { + if value && let OcsfEvent::HttpActivity(primary) = primary { + primary + .base + .unmapped + .get_or_insert_with(|| serde_json::json!({}))[key] = true.into(); + } + } + } + } + + pub(crate) fn complete_middleware(&self, phase: MiddlewarePhase, outcome: MiddlewareOutcome) { + let span = self.middleware_span(phase); + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(middleware) = &mut state.middleware[phase.index()] { + if middleware.completed { + return; + } + middleware.completed = true; + span.record("openshell.middleware.outcome", outcome.as_str()); + if outcome != MiddlewareOutcome::Allowed { + openshell_otel::mark_error(&span); + } + } + } + + pub(crate) fn record_response_delivery_started(&self) { + self.observe_middleware( + MiddlewarePhase::Response, + MiddlewareObservation { + delivery_started: true, + ..Default::default() + }, + ); + } + + pub(crate) fn record_response_partial_delivery(&self) { + self.observe_middleware( + MiddlewarePhase::Response, + MiddlewareObservation { + delivery_started: true, + partial_delivery: true, + ..Default::default() + }, + ); + } + pub(crate) fn emit(&self, event: OcsfEvent) { if matches!(event, OcsfEvent::HttpActivity(_)) { let mut state = self @@ -217,6 +502,10 @@ impl RequestCorrelation { http.disposition = Some(openshell_ocsf::DispositionId::Blocked); http.base.status = Some(StatusId::Failure); http.base.severity = SeverityId::Medium; + } else if matches!(outcome, Some("platform_error" | "upstream_error")) { + http.base.status = Some(StatusId::Failure); + http.base.status_detail = outcome.map(str::to_owned); + http.base.severity = SeverityId::Low; } } self.emit_correlated(event); @@ -229,6 +518,43 @@ impl RequestCorrelation { } } +pub fn emit_ocsf(correlation: Option<&RequestCorrelation>, event: OcsfEvent) { + if let Some(correlation) = correlation { + correlation.emit_correlated(event); + } else { + crate::telemetry::egress_emit!(event); + } +} + +pub fn emit_phase_events( + correlation: Option<&RequestCorrelation>, + phase: MiddlewarePhase, + events: impl IntoIterator, +) { + let Some(correlation) = correlation else { + crate::telemetry::emit_decision(events, None); + return; + }; + let span = correlation.middleware_span(phase); + for mut event in events { + if let OcsfEvent::HttpActivity(http) = &event { + // Routine invocation records belong to the phase span. A late + // failure is separate from the already-written primary record. + let emitted = correlation + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .primary + .is_none(); + if !emitted || http.base.status != Some(StatusId::Failure) { + continue; + } + } + event.set_correlation_uid(correlation.request_id.clone()); + span.in_scope(|| crate::telemetry::emit_correlated(event)); + } +} + /// Observe accepted response bytes without buffering or changing delivery. pub struct AuditedClient<'a, C> { inner: &'a mut C, @@ -407,6 +733,140 @@ mod tests { assert!(uuid::Uuid::parse_str(second.request_id()).is_ok()); } + #[test] + fn request_span_stays_current_across_future_yields() { + tracing::subscriber::with_default(tracing_subscriber::registry(), || { + let correlation = RequestCorrelation::new( + &request(), + &L7EvalContext { + host: "example.test".into(), + port: 443, + ..Default::default() + }, + ); + let expected = correlation.span.id().expect("request span has an id"); + let mut yielded = false; + let current = futures::executor::block_on(correlation.within_span( + futures::future::poll_fn(|cx| { + if yielded { + Poll::Ready(tracing::Span::current().id()) + } else { + yielded = true; + cx.waker().wake_by_ref(); + Poll::Pending + } + }), + )); + + assert_eq!(current, Some(expected)); + }); + } + + #[test] + fn middleware_phases_end_before_the_request_and_report_cancellation() { + use tracing_subscriber::layer::SubscriberExt as _; + + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry() + .with(openshell_otel::layer(&provider, "middleware-lifetime")); + tracing::subscriber::with_default(subscriber, || { + let correlation = RequestCorrelation::new(&request(), &L7EvalContext::default()); + { + let _phase = correlation.begin_middleware(MiddlewarePhase::Request); + correlation + .complete_middleware(MiddlewarePhase::Request, MiddlewareOutcome::Allowed); + } + { + let _cancelled = correlation.begin_middleware(MiddlewarePhase::Response); + } + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 2, "only the phases have ended"); + let completed = spans + .iter() + .find(|span| span.name == "openshell.middleware.request") + .unwrap(); + assert!(!matches!( + completed.status, + opentelemetry::trace::Status::Error { .. } + )); + let cancelled = spans + .iter() + .find(|span| span.name == "openshell.middleware.response") + .unwrap(); + assert!(matches!( + cancelled.status, + opentelemetry::trace::Status::Error { .. } + )); + assert!( + correlation + .middleware_span(MiddlewarePhase::Response) + .is_disabled() + ); + }); + assert_eq!(exporter.get_finished_spans().unwrap().len(), 3); + } + + #[test] + fn repeated_middleware_observations_preserve_the_terminal_verdict() { + use tracing_subscriber::layer::SubscriberExt as _; + + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_max_attributes_per_span(48) + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry() + .with(openshell_otel::layer(&provider, "middleware-attributes")); + tracing::subscriber::with_default(subscriber, || { + let correlation = RequestCorrelation::new(&request(), &L7EvalContext::default()); + for phase in [MiddlewarePhase::Request, MiddlewarePhase::Response] { + let _phase = correlation.begin_middleware(phase); + // OCSF correlation and remote RPCs can materialize the SDK span early. + assert!( + correlation + .middleware_span(phase) + .in_scope(openshell_otel::current_ocsf_trace_correlation) + .is_some() + ); + for _ in 0..1000 { + correlation.observe_middleware(phase, MiddlewareObservation::default()); + } + correlation.observe_middleware( + phase, + MiddlewareObservation { + diagnostics_truncated: true, + partial_delivery: true, + ..Default::default() + }, + ); + correlation.complete_middleware(phase, MiddlewareOutcome::Denied); + } + }); + let spans = exporter.get_finished_spans().unwrap(); + let phases = spans + .iter() + .filter(|span| span.name.starts_with("openshell.middleware.")) + .collect::>(); + assert_eq!(phases.len(), 2); + for span in phases { + let attributes = span + .attributes + .iter() + .map(|attribute| (attribute.key.as_str(), attribute.value.to_string())) + .collect::>(); + assert_eq!(attributes["openshell.middleware.outcome"], "denied"); + assert_eq!( + attributes["openshell.middleware.diagnostics_truncated"], + "true" + ); + assert_eq!(attributes["openshell.response.partial_delivery"], "true"); + assert_eq!(span.dropped_attributes_count, 0); + } + } + #[test] fn extracts_only_traceparent_and_preserves_duplicates() { let headers = trace_headers( diff --git a/crates/openshell-supervisor-network/src/l7/middleware.rs b/crates/openshell-supervisor-network/src/l7/middleware.rs index 94efc3fd9c..f6b325f430 100644 --- a/crates/openshell-supervisor-network/src/l7/middleware.rs +++ b/crates/openshell-supervisor-network/src/l7/middleware.rs @@ -3,6 +3,10 @@ //! Supervisor middleware application for L7 requests. +use crate::l7::correlation::{ + MiddlewareObservation, MiddlewareOutcome, MiddlewarePhase, RequestCorrelation, + emit_phase_events, +}; use crate::l7::relay::L7EvalContext; use crate::opa::PolicyGenerationGuard; use crate::telemetry::egress_emit; @@ -18,6 +22,7 @@ pub enum MiddlewareApplyResult { Allowed(crate::l7::provider::L7Request), Denied { denial: Option, + termination: openshell_supervisor_middleware::ChainTermination, }, /// The platform's shared active-work and waiter capacities were both full. /// @@ -26,6 +31,23 @@ pub enum MiddlewareApplyResult { AdmissionExhausted, } +pub fn request_rejection_correlation( + termination: openshell_supervisor_middleware::ChainTermination, +) -> (&'static str, &'static str) { + use openshell_supervisor_middleware::ChainTermination; + + match termination { + ChainTermination::MiddlewareDenied => ("denied", "request_middleware"), + ChainTermination::TransformedBodyPolicyDenied => ("denied", "post_middleware_policy"), + ChainTermination::TransformedBodyPolicyEvaluationFailed => { + ("platform_error", "post_middleware_policy") + } + ChainTermination::Completed | ChainTermination::MiddlewareFailure => { + ("platform_error", "request_middleware") + } + } +} + /// One destination-selected middleware chain shared by an HTTP request and /// its matching response. The request and response phases filter bindings /// independently, so the full chain must remain available until relay ends. @@ -59,6 +81,7 @@ impl HttpMiddlewareExchange { ctx: &L7EvalContext, scheme: &str, transformed_body_policy: openshell_supervisor_middleware::TransformedBodyPolicy<'_>, + correlation: Option<&RequestCorrelation>, ) -> Result where C: AsyncRead + AsyncWrite + Unpin + Send, @@ -73,6 +96,7 @@ impl HttpMiddlewareExchange { &self.generation_guard, transformed_body_policy, &self.request_id, + correlation, ) .await } @@ -177,26 +201,37 @@ pub fn emit_middleware_uninspectable(ctx: &L7EvalContext, detail: &str, denied: pub fn emit_websocket_preflight_events( ctx: &L7EvalContext, outcome: &openshell_supervisor_middleware::WebSocketPreflightResult, + correlation: Option<&RequestCorrelation>, ) { - emit_websocket_invocations(ctx, &outcome.invocations); - emit_websocket_coverage(ctx, &outcome.coverage); - for event in websocket_preflight_finding_events(outcome) { - egress_emit!(event); + let mut events = websocket_invocation_events(ctx, &outcome.invocations); + events.extend(websocket_coverage_events(ctx, &outcome.coverage)); + events.extend(websocket_preflight_finding_events(outcome)); + if outcome.session_capacity_exhausted { + events.push(middleware_session_capacity_exhausted_event( + ctx, + !outcome.allowed, + )); } + emit_phase_events(correlation, MiddlewarePhase::Request, events); if outcome.saturated { - emit_websocket_saturation(ctx); - } - if outcome.session_capacity_exhausted { - emit_middleware_session_capacity_exhausted(ctx, !outcome.allowed); + emit_websocket_saturation(ctx, correlation); } } pub(super) fn emit_websocket_coverage( ctx: &L7EvalContext, coverage: &[openshell_supervisor_middleware::WebSocketCoverage], +) { + emit_websocket_coverage_correlated(ctx, coverage, None); +} + +fn emit_websocket_coverage_correlated( + ctx: &L7EvalContext, + coverage: &[openshell_supervisor_middleware::WebSocketCoverage], + correlation: Option<&RequestCorrelation>, ) { for event in websocket_coverage_events(ctx, coverage) { - egress_emit!(event); + emit_phase_events(correlation, MiddlewarePhase::Request, [event]); } } @@ -254,19 +289,18 @@ pub(super) fn emit_websocket_session_start_events( ctx: &L7EvalContext, outcome: &openshell_supervisor_middleware::WebSocketSessionStartOutcome, ) { - emit_websocket_invocations(ctx, &outcome.invocations); + crate::telemetry::emit_decision(websocket_invocation_events(ctx, &outcome.invocations), None); } pub(super) fn emit_websocket_message_events( ctx: &L7EvalContext, outcome: &openshell_supervisor_middleware::WebSocketMessageOutcome, ) { - emit_websocket_invocations(ctx, &outcome.invocations); + let mut events = websocket_invocation_events(ctx, &outcome.invocations); + events.extend(websocket_message_finding_events(outcome)); + crate::telemetry::emit_decision(events, None); if outcome.saturated { - emit_websocket_saturation(ctx); - } - for event in websocket_message_finding_events(outcome) { - egress_emit!(event); + emit_websocket_saturation(ctx, None); } } @@ -314,10 +348,11 @@ pub(super) fn middleware_finding_events( .collect() } -fn emit_websocket_invocations( +fn websocket_invocation_events( ctx: &L7EvalContext, invocations: &[openshell_supervisor_middleware::WebSocketInvocation], -) { +) -> Vec { + let mut events = Vec::new(); for invocation in invocations { use openshell_supervisor_middleware::WebSocketInvocationOutcome as Outcome; let (action, disposition, severity, status, outcome_name) = match invocation.outcome { @@ -388,7 +423,7 @@ fn emit_websocket_invocations( invocation.transformed, )) .build(); - egress_emit!(event); + events.push(event); if invocation.failed { let fail_closed = invocation.outcome == Outcome::FailClosed; let mut finding = DetectionFindingBuilder::new(openshell_ocsf::ctx::ctx()); @@ -414,7 +449,7 @@ fn emit_websocket_invocations( ]) .message("WebSocket middleware stage failed") .build(); - egress_emit!(event); + events.push(event); } if invocation.stage_disabled { let event = DetectionFindingBuilder::new(openshell_ocsf::ctx::ctx()) @@ -434,12 +469,13 @@ fn emit_websocket_invocations( "WebSocket middleware stage stream became unusable and was disabled for this session", ) .build(); - egress_emit!(event); + events.push(event); } } + events } -fn emit_websocket_saturation(ctx: &L7EvalContext) { +fn emit_websocket_saturation(ctx: &L7EvalContext, correlation: Option<&RequestCorrelation>) { let event = DetectionFindingBuilder::new(openshell_ocsf::ctx::ctx()) .severity(SeverityId::Medium) .finding_info(FindingInfo::new( @@ -453,10 +489,13 @@ fn emit_websocket_saturation(ctx: &L7EvalContext) { ]) .message("WebSocket middleware work waited for admission capacity") .build(); - egress_emit!(event); + emit_phase_events(correlation, MiddlewarePhase::Request, [event]); } -fn emit_middleware_session_capacity_exhausted(ctx: &L7EvalContext, denied: bool) { +fn middleware_session_capacity_exhausted_event( + ctx: &L7EvalContext, + denied: bool, +) -> openshell_ocsf::OcsfEvent { let mut event = DetectionFindingBuilder::new(openshell_ocsf::ctx::ctx()) .severity(SeverityId::Medium) .finding_info(FindingInfo::new( @@ -472,7 +511,7 @@ fn emit_middleware_session_capacity_exhausted(ctx: &L7EvalContext, denied: bool) if denied { event = event.action(ActionId::Denied); } - egress_emit!(event.build()); + event.build() } fn middleware_admission_exhausted_event(ctx: &L7EvalContext) -> openshell_ocsf::OcsfEvent { @@ -493,8 +532,15 @@ fn middleware_admission_exhausted_event(ctx: &L7EvalContext) -> openshell_ocsf:: .build() } -fn emit_middleware_admission_exhausted(ctx: &L7EvalContext) { - egress_emit!(middleware_admission_exhausted_event(ctx)); +fn emit_middleware_admission_exhausted( + ctx: &L7EvalContext, + correlation: Option<&RequestCorrelation>, +) { + emit_phase_events( + correlation, + MiddlewarePhase::Request, + [middleware_admission_exhausted_event(ctx)], + ); } /// Largest body-buffering limit across the entries that actually resolved to a @@ -526,6 +572,7 @@ pub async fn apply_middleware_chain_with_request_id, request_id: &str, + correlation: Option<&RequestCorrelation>, ) -> Result { apply_middleware_chain_for_scheme_with_request_id( req, @@ -537,6 +584,7 @@ pub async fn apply_middleware_chain_with_request_id, request_id: &str, + correlation: Option<&RequestCorrelation>, ) -> Result { if chain.is_empty() { return Ok(MiddlewareApplyResult::Allowed(req)); } let chain = runner.describe_chain(&chain).await?; + if chain.is_empty() { + return Ok(MiddlewareApplyResult::Allowed(req)); + } + let _phase = + correlation.map(|correlation| correlation.begin_middleware(MiddlewarePhase::Request)); + if let Some(correlation) = correlation { + correlation.observe_middleware(MiddlewarePhase::Request, MiddlewareObservation::default()); + } let admission = if chain.is_empty() { None } else { @@ -568,7 +625,20 @@ pub async fn apply_middleware_chain_for_scheme_with_request_id< admission, ) => Some(admission), openshell_supervisor_middleware::MiddlewareWorkAdmissionOutcome::QueueExhausted => { - emit_middleware_admission_exhausted(ctx); + emit_middleware_admission_exhausted(ctx, correlation); + if let Some(correlation) = correlation { + correlation.observe_middleware( + MiddlewarePhase::Request, + MiddlewareObservation { + admission_exhausted: true, + ..Default::default() + }, + ); + correlation.complete_middleware( + MiddlewarePhase::Request, + MiddlewareOutcome::PlatformError, + ); + } return Ok(MiddlewareApplyResult::AdmissionExhausted); } } @@ -589,20 +659,27 @@ pub async fn apply_middleware_chain_for_scheme_with_request_id< Vec::new(), request_id, ); - let outcome = runner - .evaluate_described_with_policy_admitted( - &chain, - input, - openshell_supervisor_middleware::TransformedBodyPolicy::NotPolicyRelevant, - admission, - ) - .await?; - emit_middleware_events(ctx, &req, &outcome); + let evaluation = runner.evaluate_described_with_policy_admitted( + &chain, + input, + openshell_supervisor_middleware::TransformedBodyPolicy::NotPolicyRelevant, + admission, + ); + let outcome = if let Some(correlation) = correlation { + correlation + .within_middleware(MiddlewarePhase::Request, evaluation) + .await? + } else { + evaluation.await? + }; + observe_request_chain_outcome(correlation, &outcome); + emit_middleware_events(ctx, &req, &outcome, correlation); return Ok(if outcome.allowed { MiddlewareApplyResult::Allowed(req) } else { MiddlewareApplyResult::Denied { denial: outcome.denial, + termination: outcome.termination, } }); }; @@ -620,7 +697,10 @@ pub async fn apply_middleware_chain_for_scheme_with_request_id< { crate::l7::rest::BufferResult::Buffered(buffered) => buffered, crate::l7::rest::BufferResult::OverCapacity { recoverable } => { - return Ok(resolve_unbuffered_body(ctx, req, &chain, recoverable)); + let result = + resolve_unbuffered_body_correlated(ctx, req, &chain, recoverable, correlation); + observe_unbuffered_request_outcome(correlation, &chain, &result); + return Ok(result); } }; let headers = safe_middleware_headers(&buffered.headers)?; @@ -639,18 +719,25 @@ pub async fn apply_middleware_chain_for_scheme_with_request_id< // The explicitly selected transformation policy either re-checks every // replacement or documents that this protocol's policy is body-independent. // An ALLOW outcome therefore means the final body is policy-compliant. - let outcome = runner - .evaluate_described_with_policy_admitted( - &chain, - input, - transformed_body_policy, - Some(admission), - ) - .await?; - emit_middleware_events(ctx, &req, &outcome); + let evaluation = runner.evaluate_described_with_policy_admitted( + &chain, + input, + transformed_body_policy, + Some(admission), + ); + let outcome = if let Some(correlation) = correlation { + correlation + .within_middleware(MiddlewarePhase::Request, evaluation) + .await? + } else { + evaluation.await? + }; + observe_request_chain_outcome(correlation, &outcome); + emit_middleware_events(ctx, &req, &outcome, correlation); if !outcome.allowed { return Ok(MiddlewareApplyResult::Denied { denial: outcome.denial, + termination: outcome.termination, }); } let rebuilt = crate::l7::rest::rebuild_request_with_buffered_body( @@ -662,6 +749,84 @@ pub async fn apply_middleware_chain_for_scheme_with_request_id< Ok(MiddlewareApplyResult::Allowed(rebuilt)) } +fn observe_request_chain_outcome( + correlation: Option<&RequestCorrelation>, + outcome: &openshell_supervisor_middleware::ChainOutcome, +) { + let Some(correlation) = correlation else { + return; + }; + let (observation, middleware_outcome) = request_chain_observation(outcome); + correlation.observe_middleware(MiddlewarePhase::Request, observation); + correlation.complete_middleware(MiddlewarePhase::Request, middleware_outcome); +} + +fn request_chain_observation( + outcome: &openshell_supervisor_middleware::ChainOutcome, +) -> (MiddlewareObservation, MiddlewareOutcome) { + let observation = MiddlewareObservation { + mutated: outcome + .applied + .iter() + .any(|invocation| invocation.transformed), + fail_open: outcome.applied.iter().any(|invocation| { + invocation.failed && invocation.decision == openshell_core::proto::Decision::Allow + }), + fail_closed: outcome.applied.iter().any(|invocation| { + invocation.failed && invocation.decision == openshell_core::proto::Decision::Deny + }), + admission_exhausted: false, + ..Default::default() + }; + let middleware_outcome = match outcome.termination { + openshell_supervisor_middleware::ChainTermination::Completed + | openshell_supervisor_middleware::ChainTermination::TransformedBodyPolicyDenied + | openshell_supervisor_middleware::ChainTermination::TransformedBodyPolicyEvaluationFailed => { + MiddlewareOutcome::Allowed + } + openshell_supervisor_middleware::ChainTermination::MiddlewareDenied => { + MiddlewareOutcome::Denied + } + openshell_supervisor_middleware::ChainTermination::MiddlewareFailure => { + MiddlewareOutcome::PlatformError + } + }; + (observation, middleware_outcome) +} + +fn observe_unbuffered_request_outcome( + correlation: Option<&RequestCorrelation>, + chain: &[openshell_supervisor_middleware::DescribedChainEntry], + result: &MiddlewareApplyResult, +) { + let Some(correlation) = correlation else { + return; + }; + let fail_open = chain + .iter() + .any(|entry| entry.on_error() == openshell_supervisor_middleware::OnError::FailOpen); + let fail_closed = chain + .iter() + .any(|entry| entry.on_error() == openshell_supervisor_middleware::OnError::FailClosed); + correlation.observe_middleware( + MiddlewarePhase::Request, + MiddlewareObservation { + fail_open, + fail_closed, + ..Default::default() + }, + ); + let outcome = match result { + MiddlewareApplyResult::Allowed(_) => MiddlewareOutcome::Allowed, + MiddlewareApplyResult::Denied { + denial: Some(_), .. + } => MiddlewareOutcome::Denied, + MiddlewareApplyResult::Denied { denial: None, .. } + | MiddlewareApplyResult::AdmissionExhausted => MiddlewareOutcome::PlatformError, + }; + correlation.complete_middleware(MiddlewarePhase::Request, outcome); +} + pub async fn send_middleware_rejection_response( req: &crate::l7::provider::L7Request, client: &mut C, @@ -789,24 +954,42 @@ pub(super) fn raw_query_from_request_headers(headers: &[u8]) -> Result { /// would individually fail with `request_body_over_capacity`; the aggregate is /// a deny unless every attached middleware is `fail_open`, and passing the /// body through is only safe when no bytes were consumed. +#[cfg(test)] pub(super) fn resolve_unbuffered_body( ctx: &L7EvalContext, req: crate::l7::provider::L7Request, chain: &[openshell_supervisor_middleware::DescribedChainEntry], recoverable: bool, +) -> MiddlewareApplyResult { + resolve_unbuffered_body_correlated(ctx, req, chain, recoverable, None) +} + +fn resolve_unbuffered_body_correlated( + ctx: &L7EvalContext, + req: crate::l7::provider::L7Request, + chain: &[openshell_supervisor_middleware::DescribedChainEntry], + recoverable: bool, + correlation: Option<&RequestCorrelation>, ) -> MiddlewareApplyResult { let all_fail_open = chain .iter() .all(|entry| entry.on_error() == openshell_supervisor_middleware::OnError::FailOpen); if recoverable && all_fail_open { - emit_middleware_body_unavailable(ctx, false); + emit_middleware_body_unavailable(ctx, false, correlation); return MiddlewareApplyResult::Allowed(req); } - emit_middleware_body_unavailable(ctx, true); - MiddlewareApplyResult::Denied { denial: None } + emit_middleware_body_unavailable(ctx, true, correlation); + MiddlewareApplyResult::Denied { + denial: None, + termination: openshell_supervisor_middleware::ChainTermination::MiddlewareFailure, + } } -fn emit_middleware_body_unavailable(ctx: &L7EvalContext, denied: bool) { +fn emit_middleware_body_unavailable( + ctx: &L7EvalContext, + denied: bool, + correlation: Option<&RequestCorrelation>, +) { let mut event = DetectionFindingBuilder::new(openshell_ocsf::ctx::ctx()) .severity(if denied { SeverityId::High @@ -830,7 +1013,7 @@ fn emit_middleware_body_unavailable(ctx: &L7EvalContext, denied: bool) { if denied { event = event.action(ActionId::Denied); } - egress_emit!(event.build()); + emit_phase_events(correlation, MiddlewarePhase::Request, [event.build()]); } /// Parse the raw header block into middleware-visible headers, preserving @@ -916,6 +1099,23 @@ pub(super) fn middleware_events( let mut events = Vec::new(); for invocation in &outcome.applied { let allowed = invocation.decision == openshell_core::proto::Decision::Allow; + let failed = invocation.failed; + let status_detail = if failed { + Some(if allowed { + "middleware_failed_open".to_string() + } else { + "middleware_failed_closed".to_string() + }) + } else { + outcome.denial.as_ref().and_then(|denial| { + (denial.config_name == invocation.name).then(|| { + denial.reason_code.as_ref().map_or_else( + || format!("middleware_denied:{}", denial.config_name), + |code| format!("middleware_denied:{}:{code}", denial.config_name), + ) + }) + }) + }; let mut event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(if allowed { @@ -928,10 +1128,10 @@ pub(super) fn middleware_events( } else { DispositionId::Blocked }) - .severity(if allowed { - SeverityId::Informational - } else { + .severity(if failed || !allowed { SeverityId::Medium + } else { + SeverityId::Informational }) .http_request(HttpRequest::new(&req.action, ctx.ocsf_url(&req.target))) .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) @@ -942,10 +1142,13 @@ pub(super) fn middleware_events( "MIDDLEWARE {} {} decision={:?}", invocation.name, invocation.implementation, invocation.decision )); - if !allowed && !outcome.reason.is_empty() { - event = event - .status(StatusId::Failure) - .status_detail(&outcome.reason); + event = event.status(if failed || !allowed { + StatusId::Failure + } else { + StatusId::Success + }); + if let Some(status_detail) = status_detail { + event = event.status_detail(status_detail); } let event = event.build(); events.push(event); @@ -1019,8 +1222,13 @@ fn emit_middleware_events( ctx: &L7EvalContext, req: &crate::l7::provider::L7Request, outcome: &openshell_supervisor_middleware::ChainOutcome, + correlation: Option<&RequestCorrelation>, ) { - crate::telemetry::emit_decision(middleware_events(ctx, req, outcome), None); + emit_phase_events( + correlation, + MiddlewarePhase::Request, + middleware_events(ctx, req, outcome), + ); } #[cfg(test)] @@ -1057,6 +1265,7 @@ mod trace_tests { metadata: std::collections::BTreeMap::new(), applied, denial: None, + termination: openshell_supervisor_middleware::ChainTermination::Completed, } } @@ -1077,20 +1286,23 @@ mod trace_tests { }; for (outcome, action_id) in [(Outcome::FailClosed, Some(2)), (Outcome::FailOpen, None)] { let (events, _) = exported("warn", || { - emit_websocket_invocations( - &L7EvalContext::default(), - &[WebSocketInvocation { - config_name: "guard".to_string(), - implementation: "test/guard".to_string(), - outcome, - sequence: None, - original_size: 0, - replacement_size: None, - transformed: false, - failed: true, - stage_disabled: false, - reason_code: None, - }], + crate::telemetry::emit_decision( + websocket_invocation_events( + &L7EvalContext::default(), + &[WebSocketInvocation { + config_name: "guard".to_string(), + implementation: "test/guard".to_string(), + outcome, + sequence: None, + original_size: 0, + replacement_size: None, + transformed: false, + failed: true, + stage_disabled: false, + reason_code: None, + }], + ), + None, ); }) .await; @@ -1128,7 +1340,7 @@ mod trace_tests { let (events, received) = exported("warn", || { let ctx = L7EvalContext::default(); emit_middleware_uninspectable(&ctx, "unknown protocol", denied); - emit_middleware_body_unavailable(&ctx, denied); + emit_middleware_body_unavailable(&ctx, denied, None); }) .await; assert_eq!(events.len(), 2); @@ -1169,6 +1381,7 @@ mod trace_tests { saturated: false, session_capacity_exhausted: true, }, + None, ); }) .await; @@ -1211,6 +1424,7 @@ mod trace_tests { &L7EvalContext::default(), &request(), &outcome(allowed, reason, vec![invocation], vec![finding()]), + None, ); }) .await; @@ -1228,10 +1442,13 @@ mod trace_tests { #[cfg(test)] mod tests { use super::{ - middleware_admission_exhausted_event, safe_middleware_headers, - send_middleware_rejection_response, websocket_coverage_events, - websocket_message_finding_events, websocket_preflight_finding_events, + emit_middleware_events, emit_websocket_message_events, emit_websocket_preflight_events, + middleware_admission_exhausted_event, request_chain_observation, + request_rejection_correlation, safe_middleware_headers, send_middleware_rejection_response, + websocket_coverage_events, websocket_message_finding_events, + websocket_preflight_finding_events, }; + use crate::l7::correlation::{MiddlewareObservation, MiddlewareOutcome, RequestCorrelation}; use crate::l7::relay::L7EvalContext; use tokio::io::AsyncReadExt; @@ -1247,10 +1464,6 @@ mod tests { let serialized = serde_json::to_string(&middleware_admission_exhausted_event(&ctx)) .expect("serialize admission event"); - let event: serde_json::Value = serde_json::from_str(&serialized).unwrap(); - assert_eq!(event["class_uid"], 2004); - assert_eq!(event["action_id"], 2); - assert_eq!(event["action"], "Denied"); assert!(serialized.contains("openshell.middleware.admission_exhausted")); assert!(serialized.contains("api-policy")); assert!(serialized.contains("api.example.test")); @@ -1617,4 +1830,343 @@ mod tests { ); } } + + #[test] + fn request_chain_observation_maps_only_bounded_aggregate_state() { + use openshell_core::proto::Decision; + use openshell_supervisor_middleware::{ + ChainOutcome, ChainTermination, MiddlewareDenial, MiddlewareInvocation, + }; + + let outcome = |allowed, denial, termination, applied| ChainOutcome { + allowed, + reason: "SENTINEL_UNTRUSTED_REASON".into(), + body: b"SENTINEL_UNTRUSTED_BODY".to_vec(), + header_mutations: Vec::new(), + findings: Vec::new(), + metadata: std::collections::BTreeMap::new(), + applied, + denial, + termination, + }; + let invocation = |decision, transformed, failed| MiddlewareInvocation { + name: "SENTINEL_UNTRUSTED_NAME".into(), + implementation: "SENTINEL_UNTRUSTED_IMPLEMENTATION".into(), + decision, + transformed, + failed, + }; + + assert_eq!( + request_chain_observation(&outcome( + true, + None, + ChainTermination::Completed, + vec![invocation(Decision::Allow, true, false)], + )), + ( + MiddlewareObservation { + mutated: true, + ..Default::default() + }, + MiddlewareOutcome::Allowed, + ) + ); + assert_eq!( + request_rejection_correlation(ChainTermination::TransformedBodyPolicyDenied), + ("denied", "post_middleware_policy") + ); + assert_eq!( + request_rejection_correlation(ChainTermination::TransformedBodyPolicyEvaluationFailed), + ("platform_error", "post_middleware_policy") + ); + assert_eq!( + request_chain_observation(&outcome( + true, + None, + ChainTermination::Completed, + vec![invocation(Decision::Allow, false, true)], + )), + ( + MiddlewareObservation { + fail_open: true, + ..Default::default() + }, + MiddlewareOutcome::Allowed, + ) + ); + assert_eq!( + request_chain_observation(&outcome( + false, + Some(MiddlewareDenial { + config_name: "SENTINEL_UNTRUSTED_NAME".into(), + reason_code: Some("SENTINEL_UNTRUSTED_REASON".into()), + }), + ChainTermination::MiddlewareDenied, + vec![invocation(Decision::Deny, false, false)], + )), + (MiddlewareObservation::default(), MiddlewareOutcome::Denied) + ); + assert_eq!( + request_chain_observation(&outcome( + false, + None, + ChainTermination::MiddlewareFailure, + vec![invocation(Decision::Deny, false, true)], + )), + ( + MiddlewareObservation { + fail_closed: true, + ..Default::default() + }, + MiddlewareOutcome::PlatformError, + ) + ); + assert_eq!( + request_chain_observation(&outcome( + false, + None, + ChainTermination::TransformedBodyPolicyDenied, + vec![invocation(Decision::Allow, true, false)], + )), + ( + MiddlewareObservation { + mutated: true, + ..Default::default() + }, + MiddlewareOutcome::Allowed, + ) + ); + assert_eq!( + request_chain_observation(&outcome( + false, + None, + ChainTermination::TransformedBodyPolicyEvaluationFailed, + vec![invocation(Decision::Allow, true, false)], + )), + ( + MiddlewareObservation { + mutated: true, + ..Default::default() + }, + MiddlewareOutcome::Allowed, + ) + ); + } + + #[test] + fn correlated_middleware_keeps_one_primary_without_payload_or_reason() { + use openshell_core::proto::Decision; + use openshell_supervisor_middleware::{ + ChainOutcome, ChainTermination, MiddlewareDenial, MiddlewareInvocation, + }; + use tracing_subscriber::layer::SubscriberExt as _; + + const REQUEST_ID: &str = "00000000-0000-4000-8000-000000000080"; + const SENTINEL: &str = "SENTINEL_PRIVATE_MIDDLEWARE_VALUE"; + let ctx = L7EvalContext { + host: "api.example.test".into(), + port: 443, + policy_name: "api-policy".into(), + ..Default::default() + }; + let request = crate::l7::provider::L7Request { + action: "POST".into(), + target: "/v1/search".into(), + query_params: std::collections::HashMap::new(), + raw_header: Vec::new(), + body_length: crate::l7::provider::BodyLength::ContentLength(1), + }; + let outcome = ChainOutcome { + allowed: false, + reason: format!("middleware_denied:{SENTINEL}"), + body: SENTINEL.as_bytes().to_vec(), + header_mutations: Vec::new(), + findings: Vec::new(), + metadata: std::collections::BTreeMap::new(), + applied: vec![ + MiddlewareInvocation { + name: "request-redactor".into(), + implementation: "test/redactor".into(), + decision: Decision::Allow, + transformed: true, + failed: false, + }, + MiddlewareInvocation { + name: "request-guard".into(), + implementation: "test/guard".into(), + decision: Decision::Deny, + transformed: false, + failed: false, + }, + ], + denial: Some(MiddlewareDenial { + config_name: "request-guard".into(), + reason_code: Some("content_match".into()), + }), + termination: ChainTermination::MiddlewareDenied, + }; + let ocsf = openshell_ocsf::test_support::OcsfCapture::default(); + let subscriber = tracing_subscriber::registry().with(ocsf.clone()); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let correlation = + RequestCorrelation::with_request_id(&request, &ctx, REQUEST_ID.to_string()); + emit_middleware_events(&ctx, &request, &outcome, Some(&correlation)); + super::observe_request_chain_outcome(Some(&correlation), &outcome); + correlation.record_final_http_outcome("denied"); + correlation.response_selected(403); + }); + + let events = ocsf + .json_events() + .into_iter() + .filter(|event| event["class_uid"] == 4002) + .collect::>(); + assert_eq!(events.len(), 1); + assert!( + events + .iter() + .all(|event| { event["metadata"]["correlation_uid"] == REQUEST_ID }) + ); + assert!(events.iter().any(|event| event["action"] == "Denied")); + assert!(!serde_json::to_string(&events).unwrap().contains(SENTINEL)); + } + + #[test] + fn branch08_detailed_middleware_event_without_request_correlation_has_no_receipt() { + use openshell_core::proto::Decision; + use openshell_supervisor_middleware::{ + ChainOutcome, ChainTermination, MiddlewareInvocation, + }; + use tracing_subscriber::layer::SubscriberExt as _; + + let ctx = L7EvalContext { + host: "api.example.test".into(), + port: 443, + policy_name: "api-policy".into(), + ..Default::default() + }; + let request = crate::l7::provider::L7Request { + action: "GET".into(), + target: "/health".into(), + query_params: std::collections::HashMap::new(), + raw_header: Vec::new(), + body_length: crate::l7::provider::BodyLength::None, + }; + let outcome = ChainOutcome { + allowed: true, + reason: "allowed".into(), + body: Vec::new(), + header_mutations: Vec::new(), + findings: Vec::new(), + metadata: std::collections::BTreeMap::new(), + applied: vec![MiddlewareInvocation { + name: "request-guard".into(), + implementation: "test/guard".into(), + decision: Decision::Allow, + transformed: false, + failed: false, + }], + denial: None, + termination: ChainTermination::Completed, + }; + let ocsf = openshell_ocsf::test_support::OcsfCapture::default(); + let subscriber = tracing_subscriber::registry().with(ocsf.clone()); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + emit_middleware_events(&ctx, &request, &outcome, None); + }); + + let event = ocsf + .json_events() + .into_iter() + .find(|event| event["firewall_rule"]["type"] == "middleware") + .expect("detailed middleware event"); + assert!(event["metadata"].get("correlation_uid").is_none()); + } + + #[test] + fn branch08_websocket_only_correlates_upgrade_preflight_events() { + use openshell_supervisor_middleware::{ + WebSocketInvocation, WebSocketInvocationOutcome, WebSocketMessageOutcome, + WebSocketPreflightResult, + }; + use tracing_subscriber::layer::SubscriberExt as _; + + const REQUEST_ID: &str = "00000000-0000-4000-8000-000000000081"; + let ctx = L7EvalContext { + host: "stream.example.test".into(), + port: 443, + policy_name: "stream-policy".into(), + ..Default::default() + }; + let request = crate::l7::provider::L7Request { + action: "GET".into(), + target: "/socket".into(), + query_params: std::collections::HashMap::new(), + raw_header: Vec::new(), + body_length: crate::l7::provider::BodyLength::None, + }; + let invocation = |outcome| WebSocketInvocation { + config_name: "socket-guard".into(), + implementation: "test/socket-guard".into(), + outcome, + sequence: None, + original_size: 0, + replacement_size: None, + transformed: false, + failed: false, + stage_disabled: false, + reason_code: None, + }; + let preflight = WebSocketPreflightResult { + allowed: false, + terminal_reason: Some( + openshell_core::proto::MiddlewareSessionEndReason::MiddlewareDenial, + ), + reason: "SENTINEL_PRIVATE_PREFLIGHT_REASON".into(), + denial: None, + session: None, + findings: Vec::new(), + metadata: std::collections::BTreeMap::new(), + invocations: vec![invocation(WebSocketInvocationOutcome::Deny)], + coverage: Vec::new(), + saturated: false, + session_capacity_exhausted: false, + }; + let message = WebSocketMessageOutcome { + allowed: true, + reason: "allowed".into(), + payload: "SENTINEL_PRIVATE_MESSAGE".into(), + findings: Vec::new(), + metadata: std::collections::BTreeMap::new(), + invocations: vec![invocation(WebSocketInvocationOutcome::Allow)], + denial: None, + saturated: false, + platform_oversize: false, + }; + let ocsf = openshell_ocsf::test_support::OcsfCapture::default(); + let subscriber = tracing_subscriber::registry().with(ocsf.clone()); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let correlation = + RequestCorrelation::with_request_id(&request, &ctx, REQUEST_ID.to_string()); + emit_websocket_preflight_events(&ctx, &preflight, Some(&correlation)); + emit_websocket_message_events(&ctx, &message); + }); + + let events = ocsf + .json_events() + .into_iter() + .filter(|event| event["firewall_rule"]["type"] == "supervisor-middleware") + .collect::>(); + assert_eq!(events.len(), 2); + assert_eq!(events[0]["metadata"]["correlation_uid"], REQUEST_ID); + assert!(events[1]["metadata"].get("correlation_uid").is_none()); + assert!(!serde_json::to_string(&events).unwrap().contains("SENTINEL")); + } } diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index ad2ef43b33..87d95c2d66 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -980,30 +980,6 @@ where .await } -async fn relay_http_request_with_credential_rejection( - request: &crate::l7::provider::L7Request, - client: &mut C, - upstream: &mut U, - options: crate::l7::rest::RelayRequestOptions<'_>, - ctx: &L7EvalContext, - response_middleware: Option>, -) -> Result> -where - C: AsyncRead + AsyncWrite + Unpin, - U: AsyncRead + AsyncWrite + Unpin, -{ - relay_http_request_with_credential_rejection_observed( - request, - client, - upstream, - options, - ctx, - response_middleware, - RelayHttpInstrumentation::default(), - ) - .await -} - #[derive(Clone, Copy, Default)] struct RelayHttpInstrumentation<'a> { observer: Option<&'a EndpointObserver>, @@ -1811,23 +1787,25 @@ where engine.generation_guard(), openshell_supervisor_middleware::TransformedBodyPolicy::Reevaluate(&validate), request_id, + Some(&correlation), ) .await; let req = match middleware_result? { MiddlewareApplyResult::Allowed(request) => request, - MiddlewareApplyResult::Denied { denial, .. } => { + MiddlewareApplyResult::Denied { + denial, + termination, + } => { if let Some(observer) = observer.as_ref() { observer.observe(EndpointResult::PolicyDenied); } { correlation.record_enforcement_action("block"); correlation.record_response_status(403); - correlation.record_final_http_outcome(if denial.is_some() { - "denied" - } else { - "platform_error" - }); - correlation.record_error_stage("request_middleware"); + let (outcome, stage) = + crate::l7::middleware::request_rejection_correlation(termination); + correlation.record_final_http_outcome(outcome); + correlation.record_error_stage(stage); } let denied_request = crate::l7::provider::L7Request { action: request_info.action.clone(), @@ -1879,6 +1857,7 @@ where engine.middleware_runner(), ctx, ctx.websocket_scheme(), + Some(&correlation), ) .await; let preflight = match preflight { @@ -1895,7 +1874,11 @@ where return Ok(()); } }; - crate::l7::middleware::emit_websocket_preflight_events(ctx, &preflight); + crate::l7::middleware::emit_websocket_preflight_events( + ctx, + &preflight, + Some(&correlation), + ); if preflight.terminal_reason.is_some() { { correlation.record_enforcement_action("block"); @@ -2158,6 +2141,7 @@ pub(crate) async fn websocket_middleware_preflight( runner: &openshell_supervisor_middleware::ChainRunner, ctx: &L7EvalContext, scheme: &str, + correlation: Option<&RequestCorrelation>, ) -> Result { let header_end = req .raw_header @@ -2172,8 +2156,14 @@ pub(crate) async fn websocket_middleware_preflight( req, scheme, requested_subprotocols, + correlation.map(RequestCorrelation::request_id), ); - runner.preflight_websocket(chain, input).await + let preflight = runner.preflight_websocket(chain, input); + if let Some(correlation) = correlation { + correlation.within_span(preflight).await + } else { + preflight.await + } } /// Build the WebSocket preflight input from the sandbox and evaluation @@ -2186,10 +2176,11 @@ fn websocket_preflight_input( req: &crate::l7::provider::L7Request, scheme: &str, requested_subprotocols: Vec, + request_id: Option<&str>, ) -> openshell_supervisor_middleware::WebSocketPreflightInput { openshell_supervisor_middleware::WebSocketPreflightInput { session_id: uuid::Uuid::new_v4().to_string(), - request_id: uuid::Uuid::new_v4().to_string(), + request_id: request_id.map_or_else(|| uuid::Uuid::new_v4().to_string(), str::to_string), sandbox_id: sandbox.sandbox_id.clone(), sandbox_name: sandbox.sandbox_name.clone(), workspace: ctx.workspace.clone(), @@ -2653,19 +2644,21 @@ where engine.generation_guard(), openshell_supervisor_middleware::TransformedBodyPolicy::NotPolicyRelevant, request_id, + Some(&correlation), ) .await; let req = match middleware_result? { MiddlewareApplyResult::Allowed(request) => request, - MiddlewareApplyResult::Denied { denial, .. } => { + MiddlewareApplyResult::Denied { + denial, + termination, + } => { correlation.record_enforcement_action("block"); correlation.record_response_status(403); - correlation.record_final_http_outcome(if denial.is_some() { - "denied" - } else { - "platform_error" - }); - correlation.record_error_stage("request_middleware"); + let (outcome, stage) = + crate::l7::middleware::request_rejection_correlation(termination); + correlation.record_final_http_outcome(outcome); + correlation.record_error_stage(stage); let denied_request = crate::l7::provider::L7Request { action: request_info.action.clone(), target: redacted_target.clone(), @@ -2714,6 +2707,7 @@ where engine.middleware_runner(), ctx, ctx.websocket_scheme(), + Some(&correlation), ) .await; let preflight = match preflight { @@ -2728,7 +2722,11 @@ where return Ok(()); } }; - crate::l7::middleware::emit_websocket_preflight_events(ctx, &preflight); + crate::l7::middleware::emit_websocket_preflight_events( + ctx, + &preflight, + Some(&correlation), + ); if preflight.terminal_reason.is_some() { correlation.record_enforcement_action("block"); correlation.record_response_status(403); @@ -3136,22 +3134,24 @@ where engine.generation_guard(), openshell_supervisor_middleware::TransformedBodyPolicy::Reevaluate(&validate), request_id, + Some(&correlation), ) .await? { MiddlewareApplyResult::Allowed(request) => request, - MiddlewareApplyResult::Denied { denial, .. } => { + MiddlewareApplyResult::Denied { + denial, + termination, + } => { if let Some(observer) = observer.as_ref() { observer.observe(EndpointResult::PolicyDenied); } correlation.record_enforcement_action("block"); correlation.record_response_status(403); - correlation.record_final_http_outcome(if denial.is_some() { - "denied" - } else { - "platform_error" - }); - correlation.record_error_stage("request_middleware"); + let (outcome, stage) = + crate::l7::middleware::request_rejection_correlation(termination); + correlation.record_final_http_outcome(outcome); + correlation.record_error_stage(stage); let denied_request = crate::l7::provider::L7Request { action: request_info.action.clone(), target: redacted_target.clone(), @@ -3463,19 +3463,21 @@ where engine.generation_guard(), openshell_supervisor_middleware::TransformedBodyPolicy::Reevaluate(&validate), request_id, + Some(&correlation), ) .await? { MiddlewareApplyResult::Allowed(request) => request, - MiddlewareApplyResult::Denied { denial, .. } => { + MiddlewareApplyResult::Denied { + denial, + termination, + } => { correlation.record_enforcement_action("block"); correlation.record_response_status(403); - correlation.record_final_http_outcome(if denial.is_some() { - "denied" - } else { - "platform_error" - }); - correlation.record_error_stage("request_middleware"); + let (outcome, stage) = + crate::l7::middleware::request_rejection_correlation(termination); + correlation.record_final_http_outcome(outcome); + correlation.record_error_stage(stage); let denied_request = crate::l7::provider::L7Request { action: request_info.action.clone(), target: redacted_target.clone(), @@ -4169,7 +4171,8 @@ where egress_emit!(event); } - let request_id = uuid::Uuid::new_v4().to_string(); + let fallback_request_id = uuid::Uuid::new_v4().to_string(); + let mut correlation = None; let mut response_selection = None; let req = if let Some(engine) = middleware_engine { let input = middleware_network_input(ctx); @@ -4177,9 +4180,17 @@ where if generation != generation_guard.captured_generation() { return Ok(()); } + if !chain.is_empty() { + let selected = RequestCorrelation::new(&req, ctx); + selected.record_policy_decision("not_applicable"); + correlation = Some(selected); + } + let request_id = correlation + .as_ref() + .map_or(fallback_request_id.as_str(), RequestCorrelation::request_id); let runner = engine.middleware_runner()?; let exchange = crate::l7::middleware::HttpMiddlewareExchange::new( - request_id.clone(), + request_id.to_string(), chain, runner, generation_guard.clone(), @@ -4193,11 +4204,23 @@ where ctx, ctx.request_scheme(), openshell_supervisor_middleware::TransformedBodyPolicy::NotPolicyRelevant, + correlation.as_ref(), ) .await?; let request = match result { MiddlewareApplyResult::Allowed(request) => request, - MiddlewareApplyResult::Denied { denial, .. } => { + MiddlewareApplyResult::Denied { + denial, + termination, + } => { + if let Some(correlation) = correlation.as_ref() { + correlation.record_enforcement_action("block"); + correlation.record_response_status(403); + let (outcome, stage) = + crate::l7::middleware::request_rejection_correlation(termination); + correlation.record_final_http_outcome(outcome); + correlation.record_error_stage(stage); + } let denied_request = crate::l7::provider::L7Request { action: "HTTP".into(), target: redacted_target.clone(), @@ -4211,12 +4234,18 @@ where ctx, denial.as_ref(), &redacted_target, - None, + correlation.as_ref().map(RequestCorrelation::request_id), ) .await?; return Ok(()); } MiddlewareApplyResult::AdmissionExhausted => { + if let Some(correlation) = correlation.as_ref() { + correlation.record_enforcement_action("block"); + correlation.record_response_status(503); + correlation.record_final_http_outcome("platform_error"); + correlation.record_error_stage("middleware_admission"); + } let unavailable_request = crate::l7::provider::L7Request { action: "HTTP".into(), target: redacted_target.clone(), @@ -4229,7 +4258,7 @@ where client, ctx, &redacted_target, - None, + correlation.as_ref().map(RequestCorrelation::request_id), ) .await?; return Ok(()); @@ -4240,6 +4269,9 @@ where } else { req }; + let request_id = correlation + .as_ref() + .map_or(fallback_request_id.as_str(), RequestCorrelation::request_id); let injected = match crate::l7::token_grant_injection::pin_live_credentials(&mut grant_ctx) { @@ -4248,7 +4280,7 @@ where req, &grant_ctx, &snapshot, - &request_id, + request_id, observation_context.as_ref(), ) .await @@ -4258,7 +4290,13 @@ where let Ok(req_with_auth) = injected else { // The refusal event names the credential and request ID. // The sandbox-level grant record does not require an MCP endpoint identity. - write_bad_gateway_response(client, Some(&request_id)).await?; + if let Some(correlation) = correlation.as_ref() { + correlation.record_enforcement_action("block"); + correlation.record_response_status(502); + correlation.record_final_http_outcome("platform_error"); + correlation.record_error_stage("credential_preparation"); + } + write_bad_gateway_response(client, Some(request_id)).await?; return Ok(()); }; let ctx = &grant_ctx; @@ -4272,7 +4310,7 @@ where // Forward request with credential rewriting and relay the response. // relay_http_request_with_resolver handles both directions: it sends // the request upstream and reads the response back to the client. - let Some(outcome) = relay_http_request_with_credential_rejection( + let Some(outcome) = relay_http_request_with_credential_rejection_observed( &req_with_auth, client, upstream, @@ -4284,6 +4322,10 @@ where }, ctx, response_middleware, + RelayHttpInstrumentation { + observer: None, + correlation: correlation.as_ref(), + }, ) .await? else { @@ -4730,12 +4772,20 @@ mod tests { body_length: crate::l7::provider::BodyLength::None, }; - let input = - websocket_preflight_input(&sandbox, &eval, &req, "wss", vec!["chat".to_string()]); + let input = websocket_preflight_input( + &sandbox, + &eval, + &req, + "wss", + vec!["chat".to_string()], + Some("request-123"), + ); assert_eq!(input.sandbox_id, "sbx-123"); assert_eq!(input.sandbox_name, "nightly-build"); assert_eq!(input.workspace, "team-a"); + assert_eq!(input.request_id, "request-123"); + assert_ne!(input.session_id, input.request_id); } #[test] @@ -5231,7 +5281,7 @@ mod tests { port: 443, ..Default::default() }; - let outcome = relay_http_request_with_credential_rejection( + let outcome = relay_http_request_with_credential_rejection_observed( &request, &mut client, &mut upstream, @@ -5242,6 +5292,7 @@ mod tests { }, &ctx, None, + RelayHttpInstrumentation::default(), ) .await .unwrap(); @@ -8690,6 +8741,7 @@ network_policies: tunnel_engine.generation_guard(), openshell_supervisor_middleware::TransformedBodyPolicy::NotPolicyRelevant, "test-request-id", + None, ) .await .expect("apply middleware chain"); @@ -8806,6 +8858,7 @@ network_policies: tunnel_engine.generation_guard(), openshell_supervisor_middleware::TransformedBodyPolicy::NotPolicyRelevant, "test-request-id", + None, ) .await .expect("apply middleware chain"); @@ -8967,7 +9020,7 @@ network_policies: #[test] fn middleware_ocsf_events_are_audit_safe() { use openshell_supervisor_middleware::{ - ChainOutcome, MiddlewareInvocation, NamespacedFinding, + ChainOutcome, ChainTermination, MiddlewareInvocation, NamespacedFinding, }; const RAW_SECRET: &str = "sk-RAWSECRETVALUE0123456789"; @@ -9016,6 +9069,7 @@ network_policies: failed: false, }], denial: None, + termination: ChainTermination::Completed, }; // Build the events directly rather than routing through the global @@ -9101,6 +9155,7 @@ network_policies: config_name: "content-guard".into(), reason_code: Some("content_match".into()), }), + termination: ChainTermination::MiddlewareDenied, }; let denied_events = middleware_events(&ctx, &req, &denied_outcome); let denied_http = denied_events @@ -9125,6 +9180,7 @@ network_policies: let external_failure_outcome = ChainOutcome { reason: "middleware_failed: header_mutation_invalid_name".into(), denial: None, + termination: ChainTermination::MiddlewareFailure, applied: vec![MiddlewareInvocation { failed: true, ..denied_outcome.applied[0].clone() @@ -9133,8 +9189,44 @@ network_policies: }; let failure_events = middleware_events(&ctx, &req, &external_failure_outcome); let serialized = serde_json::to_string(&failure_events).expect("serialize failure events"); - assert!(serialized.contains("header_mutation_invalid_name")); + assert!(serialized.contains("middleware_failed_closed")); + assert!(!serialized.contains("header_mutation_invalid_name")); assert!(!serialized.contains(RAW_SECRET)); + + let fail_open_outcome = ChainOutcome { + allowed: true, + reason: format!("middleware_failed:{RAW_SECRET}"), + body: Vec::new(), + header_mutations: Vec::new(), + findings: Vec::new(), + metadata: BTreeMap::new(), + applied: vec![MiddlewareInvocation { + name: "content-guard".into(), + implementation: "example/content-guard".into(), + decision: openshell_core::proto::Decision::Allow, + transformed: false, + failed: true, + }], + denial: None, + termination: ChainTermination::Completed, + }; + let fail_open_events = middleware_events(&ctx, &req, &fail_open_outcome); + let fail_open_http = fail_open_events + .iter() + .find(|event| event.class_uid() == 4002) + .expect("expected fail-open HTTP Activity event"); + assert_eq!(fail_open_http.base().status, Some(StatusId::Failure)); + assert_eq!(fail_open_http.base().severity, SeverityId::Medium); + assert_eq!( + fail_open_http.base().status_detail.as_deref(), + Some("middleware_failed_open") + ); + assert_eq!(fail_open_http.to_json().unwrap()["action"], "Allowed"); + assert!( + !serde_json::to_string(&fail_open_events) + .unwrap() + .contains(RAW_SECRET) + ); } #[tokio::test] diff --git a/crates/openshell-supervisor-network/src/l7/rest.rs b/crates/openshell-supervisor-network/src/l7/rest.rs index b4644e44cf..299013dd1c 100644 --- a/crates/openshell-supervisor-network/src/l7/rest.rs +++ b/crates/openshell-supervisor-network/src/l7/rest.rs @@ -8333,6 +8333,7 @@ mod tests { sequence: Some(1), input_size: 19, output_size: None, + transformed: false, failed: true, stage_disabled: true, reason_code: Some("stable_reason".into()), @@ -8380,6 +8381,7 @@ mod tests { sequence: Some(1), input_size: 19, output_size: None, + transformed: false, failed: true, stage_disabled: true, reason_code: None, diff --git a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs index 606767d3fc..1b152b6a99 100644 --- a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs +++ b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs @@ -4,7 +4,10 @@ //! HTTP response relay and pre-return middleware integration. use super::*; -use crate::l7::correlation::RequestCorrelation; +use crate::l7::correlation::{ + MiddlewareObservation, MiddlewareOutcome, MiddlewarePhase, RequestCorrelation, emit_ocsf, + emit_phase_events, +}; /// Default wall-clock bound shared by whole-body stages in one response. pub const DEFAULT_HTTP_RESPONSE_WHOLE_BODY_TIMEOUT: std::time::Duration = @@ -584,6 +587,12 @@ where { Ok(described) if described.is_empty() => return Ok(None), Ok(described) => { + if let Some(correlation) = response_correlation { + correlation.observe_middleware( + MiddlewarePhase::Response, + MiddlewareObservation::default(), + ); + } parse_response_head_for_middleware(header_bytes).map(|parsed| (described, parsed)) } Err(error) => Err(error), @@ -594,12 +603,16 @@ where let (described, parsed) = match parsed { Ok(parsed) => parsed, Err(error) => { + let _phase = response_correlation + .map(|correlation| correlation.begin_middleware(MiddlewarePhase::Response)); + complete_response_middleware(response_correlation, MiddlewareOutcome::PlatformError); debug!(error = %error, "HTTP response head normalization failed"); emit_http_response_middleware_failure( middleware.policy_name, &middleware.target, status_code, false, + response_correlation, ); send_response_delivery_failure( client, @@ -613,6 +626,8 @@ where return Ok(Some(RelayOutcome::Consumed)); } }; + let _phase = response_correlation + .map(|correlation| correlation.begin_middleware(MiddlewarePhase::Response)); let original_headers = parsed.headers.clone(); let preserved_credential_headers = parsed.preserved_credential_headers; let upstream_declared_trailers = parsed.declared_trailers.clone(); @@ -629,10 +644,16 @@ where connection_nominated_headers: parsed.connection_nominated, }; let preflight_result = if parsed.representable { - middleware + let preflight = middleware .runner - .preflight_described_http_response(described, input) - .await + .preflight_described_http_response(described, input); + if let Some(correlation) = response_correlation { + correlation + .within_middleware(MiddlewarePhase::Response, preflight) + .await + } else { + preflight.await + } } else { Ok(middleware .runner @@ -641,12 +662,20 @@ where let mut preflight = match preflight_result { Ok(preflight) => preflight, Err(error) => { + observe_response_admission_exhaustion( + response_correlation, + error + .downcast_ref::() + .is_some(), + ); + complete_response_middleware(response_correlation, MiddlewareOutcome::PlatformError); debug!(error = %error, "HTTP response middleware preflight failed"); emit_http_response_middleware_failure( middleware.policy_name, &middleware.target, status_code, false, + response_correlation, ); send_response_delivery_failure( client, @@ -660,6 +689,13 @@ where return Ok(Some(RelayOutcome::Consumed)); } }; + observe_response_middleware( + response_correlation, + &preflight.invocations, + preflight.headers != original_headers, + preflight.session_capacity_exhausted, + false, + ); if let Some(guard) = middleware.generation_guard && let Err(error) = guard.ensure_current() { @@ -695,8 +731,22 @@ where )); } } - crate::telemetry::emit_decision(preflight_events, None); + emit_phase_events( + response_correlation, + MiddlewarePhase::Response, + preflight_events, + ); if !preflight.allowed { + let denied = + preflight.denial.is_some() || response_invocations_blocked(&preflight.invocations); + complete_response_middleware( + response_correlation, + if denied { + MiddlewareOutcome::Denied + } else { + MiddlewareOutcome::PlatformError + }, + ); if let Some(denial) = preflight.denial.as_ref() { send_response_middleware_denial( client, @@ -721,21 +771,21 @@ where } return Ok(Some(RelayOutcome::Consumed)); } - crate::telemetry::emit_decision( - http_response_decision_events( - middleware.policy_name, - &middleware.target, - status_code, - &preflight.invocations, - ), - None, + emit_http_response_middleware_invocations( + middleware.policy_name, + &middleware.target, + status_code, + &preflight.invocations, + response_correlation, ); let Some(mut session) = preflight.session else { if preflight.headers == original_headers { + complete_response_middleware(response_correlation, MiddlewareOutcome::Allowed); return Ok(None); } let status_line = response_status_line(header_bytes)?; + complete_response_middleware(response_correlation, MiddlewareOutcome::Allowed); let outcome = relay_headers_only_response( request_method, upstream, @@ -763,22 +813,25 @@ where let finish = match session.finish(Vec::new()).await { Ok(finish) => finish, Err(error) => { - debug!(error = %error, "HTTP response middleware finalization failed"); - let mut events = http_response_diagnostic_events( - middleware.policy_name, - &middleware.target, - status_code, - &error.diagnostics.invocations, - &error.diagnostics.findings, + observe_response_diagnostics(response_correlation, &error.diagnostics); + observe_response_admission_exhaustion( + response_correlation, + error.admission_exhausted(), + ); + complete_response_middleware( + response_correlation, + middleware_failure_outcome(error.denial.is_some(), &error.diagnostics), ); - events.push(http_response_middleware_failure_event( + debug!(error = %error, "HTTP response middleware finalization failed"); + emit_http_response_diagnostics( middleware.policy_name, &middleware.target, status_code, - false, + &error.diagnostics, true, - )); - crate::telemetry::emit_decision(events, None); + Some(false), + response_correlation, + ); send_response_delivery_failure( client, request_method, @@ -791,19 +844,32 @@ where return Ok(Some(RelayOutcome::Consumed)); } }; + observe_response_middleware( + response_correlation, + &finish.invocations, + finish.strip_stale_integrity_headers, + false, + finish.diagnostics_truncated, + ); if let Some(guard) = middleware.generation_guard { guard.ensure_current()?; } let mut headers = preflight.headers; - crate::telemetry::emit_decision( - http_response_diagnostic_events( - middleware.policy_name, - &middleware.target, - status_code, - &finish.invocations, - &finish.findings, - ), - None, + emit_http_response_middleware_invocations( + middleware.policy_name, + &middleware.target, + status_code, + &finish.invocations, + response_correlation, + ); + for event in crate::l7::middleware::middleware_finding_events(&finish.findings) { + crate::telemetry::egress_emit!(event); + } + emit_http_response_diagnostic_truncation( + middleware.policy_name, + &middleware.target, + finish.diagnostics_truncated, + response_correlation, ); if finish.strip_stale_integrity_headers { strip_response_integrity_headers(&mut headers); @@ -817,6 +883,7 @@ where &[], ); let head = final_response_head(&head, response_correlation); + complete_response_middleware(response_correlation, MiddlewareOutcome::Allowed); client.write_all(&head).await.into_diagnostic()?; client.flush().await.into_diagnostic()?; record_allowed_response(response_correlation); @@ -861,6 +928,7 @@ where .await; return Err(error).into_diagnostic(); } + record_response_delivery_started(response_correlation); if let Err(error) = client.flush().await { session .end(openshell_core::proto::MiddlewareSessionEndReason::DownstreamDisconnect) @@ -886,6 +954,7 @@ where &middleware.target, status_code, &connection_nominated_headers, + response_correlation, ) .await; let trailers = match body_result { @@ -910,26 +979,38 @@ where // Diagnostics from a middleware stop and its terminal finding are // one decision. Transport/framing errors still record delivery // failure, but do not imply an enforced middleware refusal. - let mut events = middleware_stop.map_or_else(Vec::new, |stop| { - http_response_diagnostic_events( - middleware.policy_name, - &middleware.target, - status_code, - &stop.diagnostics.invocations, - &stop.diagnostics.findings, - ) - }); - // A denial sent before commitment is reported by its invocation. - if committed || middleware_stop.is_none_or(|stop| stop.failure.denial.is_none()) { - events.push(http_response_middleware_failure_event( - middleware.policy_name, - &middleware.target, - status_code, - committed, - middleware_stop.is_some(), - )); + let empty_diagnostics = + openshell_supervisor_middleware::HttpResponseDiagnostics::default(); + emit_http_response_diagnostics( + middleware.policy_name, + &middleware.target, + status_code, + middleware_stop.map_or(&empty_diagnostics, |stop| &stop.diagnostics), + middleware_stop.is_some(), + (committed || middleware_stop.is_none_or(|stop| stop.failure.denial.is_none())) + .then_some(committed), + response_correlation, + ); + if committed { + record_response_partial_delivery(response_correlation); + } + if let Some(stop) = middleware_stop { + observe_response_diagnostics(response_correlation, &stop.failure.diagnostics); + observe_response_admission_exhaustion( + response_correlation, + stop.failure.admission_exhausted(), + ); + let denied = stop.failure.denial.is_some() + || response_invocations_blocked(&stop.failure.diagnostics.invocations); + complete_response_middleware( + response_correlation, + if denied { + MiddlewareOutcome::Denied + } else { + MiddlewareOutcome::PlatformError + }, + ); } - crate::telemetry::emit_decision(events, None); if committed { if middleware_stop.is_some_and(|stop| stop.failure.denial.is_some()) { record_committed_response_error( @@ -976,32 +1057,40 @@ where if let Some(guard) = middleware.generation_guard && let Err(error) = guard.ensure_current() { + if committed { + record_response_partial_delivery(response_correlation); + } session .end(openshell_core::proto::MiddlewareSessionEndReason::PolicyReload) .await; return Err(error); } + let input_trailers = trailers.clone(); let finish = match session.finish(trailers).await { Ok(finish) => finish, Err(error) => { - let mut events = http_response_diagnostic_events( + observe_response_diagnostics(response_correlation, &error.diagnostics); + observe_response_admission_exhaustion( + response_correlation, + error.admission_exhausted(), + ); + if committed { + record_response_partial_delivery(response_correlation); + } + complete_response_middleware( + response_correlation, + middleware_failure_outcome(error.denial.is_some(), &error.diagnostics), + ); + emit_http_response_diagnostics( middleware.policy_name, &middleware.target, status_code, - &error.diagnostics.invocations, - &error.diagnostics.findings, + &error.diagnostics, + true, + (committed || error.denial.is_none()).then_some(committed), + response_correlation, ); - if committed || error.denial.is_none() { - events.push(http_response_middleware_failure_event( - middleware.policy_name, - &middleware.target, - status_code, - committed, - true, - )); - } - crate::telemetry::emit_decision(events, None); if committed { if error.denial.is_some() { record_committed_response_error( @@ -1046,18 +1135,36 @@ where return Ok(Some(RelayOutcome::Consumed)); } }; - if let Some(guard) = middleware.generation_guard { - guard.ensure_current()?; + observe_response_middleware( + response_correlation, + &finish.invocations, + finish.strip_stale_integrity_headers || finish.trailers != input_trailers, + false, + finish.diagnostics_truncated, + ); + if let Some(guard) = middleware.generation_guard + && let Err(error) = guard.ensure_current() + { + if committed { + record_response_partial_delivery(response_correlation); + } + return Err(error); } - crate::telemetry::emit_decision( - http_response_diagnostic_events( - middleware.policy_name, - &middleware.target, - status_code, - &finish.invocations, - &finish.findings, - ), - None, + emit_http_response_middleware_invocations( + middleware.policy_name, + &middleware.target, + status_code, + &finish.invocations, + response_correlation, + ); + for event in crate::l7::middleware::middleware_finding_events(&finish.findings) { + crate::telemetry::egress_emit!(event); + } + emit_http_response_diagnostic_truncation( + middleware.policy_name, + &middleware.target, + finish.diagnostics_truncated, + response_correlation, ); if whole_body && !committed { @@ -1065,11 +1172,14 @@ where if finish.strip_stale_integrity_headers { strip_response_integrity_headers(&mut headers); } - let output_length = finish + let Some(output_length) = finish .body_units .iter() .try_fold(0usize, |total, unit| total.checked_add(unit.len())) - .ok_or_else(|| miette!("HTTP response middleware output length overflow"))?; + else { + complete_response_middleware(response_correlation, MiddlewareOutcome::PlatformError); + return Err(miette!("HTTP response middleware output length overflow")); + }; let framing = if finish.trailers.is_empty() || !supports_chunked_response { ResponseFraming::ContentLength(output_length as u64) } else { @@ -1093,6 +1203,7 @@ where &trailer_names, ); let head = final_response_head(&head, response_correlation); + complete_response_middleware(response_correlation, MiddlewareOutcome::Allowed); client.write_all(&head).await.into_diagnostic()?; if matches!(framing, ResponseFraming::Chunked) { for unit in &finish.body_units { @@ -1105,6 +1216,7 @@ where } } } else { + complete_response_middleware(response_correlation, MiddlewareOutcome::Allowed); for unit in &finish.body_units { if chunked_output { write_downstream_response_chunk(client, unit).await?; @@ -1230,16 +1342,134 @@ fn http_response_decision_events( events } -fn http_response_diagnostic_events( +fn observe_response_middleware( + correlation: Option<&RequestCorrelation>, + invocations: &[openshell_supervisor_middleware::HttpResponseInvocation], + mutated: bool, + admission_exhausted: bool, + diagnostics_truncated: bool, +) { + let Some(correlation) = correlation else { + return; + }; + correlation.observe_middleware( + MiddlewarePhase::Response, + MiddlewareObservation { + diagnostics_truncated, + ..response_middleware_observation(invocations, mutated, admission_exhausted) + }, + ); +} + +fn response_middleware_observation( + invocations: &[openshell_supervisor_middleware::HttpResponseInvocation], + mutated: bool, + admission_exhausted: bool, +) -> MiddlewareObservation { + let mut observation = MiddlewareObservation { + mutated, + admission_exhausted, + ..Default::default() + }; + for invocation in invocations { + observation.mutated |= invocation.transformed; + match invocation.outcome { + openshell_supervisor_middleware::HttpResponseInvocationOutcome::Transform => { + observation.mutated = true; + } + openshell_supervisor_middleware::HttpResponseInvocationOutcome::FailOpen => { + observation.fail_open = true; + } + openshell_supervisor_middleware::HttpResponseInvocationOutcome::FailClosed => { + observation.fail_closed = true; + } + _ => {} + } + } + observation +} + +fn observe_response_diagnostics( + correlation: Option<&RequestCorrelation>, + diagnostics: &openshell_supervisor_middleware::HttpResponseDiagnostics, +) { + observe_response_middleware( + correlation, + &diagnostics.invocations, + false, + false, + diagnostics.diagnostics_truncated, + ); +} + +fn observe_response_admission_exhaustion( + correlation: Option<&RequestCorrelation>, + admission_exhausted: bool, +) { + if admission_exhausted && let Some(correlation) = correlation { + correlation.observe_middleware( + MiddlewarePhase::Response, + MiddlewareObservation { + admission_exhausted: true, + ..Default::default() + }, + ); + } +} + +fn record_response_delivery_started(correlation: Option<&RequestCorrelation>) { + if let Some(correlation) = correlation { + correlation.record_response_delivery_started(); + } +} + +fn record_response_partial_delivery(correlation: Option<&RequestCorrelation>) { + if let Some(correlation) = correlation { + correlation.record_response_partial_delivery(); + } +} + +fn complete_response_middleware( + correlation: Option<&RequestCorrelation>, + outcome: MiddlewareOutcome, +) { + if let Some(correlation) = correlation { + correlation.complete_middleware(MiddlewarePhase::Response, outcome); + } +} + +fn response_invocations_blocked( + invocations: &[openshell_supervisor_middleware::HttpResponseInvocation], +) -> bool { + invocations.iter().any(|invocation| { + invocation.outcome + == openshell_supervisor_middleware::HttpResponseInvocationOutcome::BlockDelivery + }) +} + +fn middleware_failure_outcome( + explicit_denial: bool, + diagnostics: &openshell_supervisor_middleware::HttpResponseDiagnostics, +) -> MiddlewareOutcome { + if explicit_denial || response_invocations_blocked(&diagnostics.invocations) { + MiddlewareOutcome::Denied + } else { + MiddlewareOutcome::PlatformError + } +} + +fn emit_http_response_middleware_invocations( policy_name: &str, target: &HttpRequestTarget, status_code: u16, invocations: &[openshell_supervisor_middleware::HttpResponseInvocation], - findings: &[openshell_supervisor_middleware::NamespacedFinding], -) -> Vec { - let mut events = http_response_decision_events(policy_name, target, status_code, invocations); - events.extend(crate::l7::middleware::middleware_finding_events(findings)); - events + correlation: Option<&RequestCorrelation>, +) { + emit_phase_events( + correlation, + MiddlewarePhase::Response, + http_response_decision_events(policy_name, target, status_code, invocations), + ); } fn emit_http_response_diagnostics( @@ -1247,19 +1477,68 @@ fn emit_http_response_diagnostics( target: &HttpRequestTarget, status_code: u16, diagnostics: &openshell_supervisor_middleware::HttpResponseDiagnostics, + denied: bool, + terminal_failure: Option, + correlation: Option<&RequestCorrelation>, ) { - crate::telemetry::emit_decision( - http_response_diagnostic_events( + let mut events = + http_response_decision_events(policy_name, target, status_code, &diagnostics.invocations); + events.extend(crate::l7::middleware::middleware_finding_events( + &diagnostics.findings, + )); + if let Some(committed) = terminal_failure { + events.push(http_response_middleware_failure_event( policy_name, target, status_code, - &diagnostics.invocations, - &diagnostics.findings, - ), - None, + committed, + denied, + )); + } + emit_phase_events(correlation, MiddlewarePhase::Response, events); + emit_http_response_diagnostic_truncation( + policy_name, + target, + diagnostics.diagnostics_truncated, + correlation, + ); +} + +fn emit_http_response_diagnostic_truncation( + policy_name: &str, + target: &HttpRequestTarget, + diagnostics_truncated: bool, + correlation: Option<&RequestCorrelation>, +) { + if !diagnostics_truncated { + return; + } + emit_ocsf( + correlation, + http_response_diagnostic_truncation_event(policy_name, target), ); } +fn http_response_diagnostic_truncation_event( + policy_name: &str, + target: &HttpRequestTarget, +) -> openshell_ocsf::OcsfEvent { + openshell_ocsf::DetectionFindingBuilder::new(ocsf_ctx()) + .severity(openshell_ocsf::SeverityId::Low) + .finding_info(openshell_ocsf::FindingInfo::new( + "openshell.middleware.http_response_diagnostics_truncated", + "HTTP response middleware diagnostics truncated", + )) + .evidence_pairs(&[ + ("policy", policy_name), + ("host", target.host.as_str()), + ("phase", "pre_return"), + ]) + .unmapped("phase", "pre_return") + .message("HTTP response middleware diagnostic evidence exceeded the session budget") + .build() +} + pub(super) fn http_response_middleware_invocation_events( policy_name: &str, target: &HttpRequestTarget, @@ -1428,14 +1707,19 @@ fn emit_http_response_middleware_failure( target: &HttpRequestTarget, status_code: u16, committed: bool, + correlation: Option<&RequestCorrelation>, ) { - crate::telemetry::egress_emit!(http_response_middleware_failure_event( - policy_name, - target, - status_code, - committed, - true, - )); + emit_phase_events( + correlation, + MiddlewarePhase::Response, + [http_response_middleware_failure_event( + policy_name, + target, + status_code, + committed, + true, + )], + ); } fn http_response_middleware_failure_event( @@ -1786,6 +2070,7 @@ async fn relay_normalized_response_body( target: &HttpRequestTarget, status_code: u16, connection_nominated_headers: &[String], + response_correlation: Option<&RequestCorrelation>, ) -> Result> where R: AsyncRead + Unpin, @@ -1800,6 +2085,7 @@ where policy_name, target, status_code, + response_correlation, }; match body_length { BodyLength::ContentLength(mut remaining) => { @@ -2003,6 +2289,7 @@ struct ResponseOutputState<'a> { policy_name: &'a str, target: &'a HttpRequestTarget, status_code: u16, + response_correlation: Option<&'a RequestCorrelation>, } #[derive(Debug)] @@ -2060,16 +2347,26 @@ async fn expire_whole_body_deadline( framing: &mut ResponseOutputState<'_>, ) -> Result<()> { let output = session.expire_whole_body_deadline().await; + let diagnostics = session.take_diagnostics(); + observe_response_diagnostics(framing.response_correlation, &diagnostics); + if let Err(failure) = &output { + observe_response_admission_exhaustion( + framing.response_correlation, + failure.admission_exhausted(), + ); + } if let Some(guard) = framing.generation_guard { guard.ensure_current()?; } - let diagnostics = session.take_diagnostics(); if output.is_ok() { emit_http_response_diagnostics( framing.policy_name, framing.target, framing.status_code, &diagnostics, + false, + None, + framing.response_correlation, ); } let output = output.map_err(|failure| ResponseMiddlewareStop::new(failure, diagnostics))?; @@ -2204,16 +2501,26 @@ async fn process_response_unit( framing: &mut ResponseOutputState<'_>, ) -> Result<()> { let output = session.push_body(unit).await; + let diagnostics = session.take_diagnostics(); + observe_response_diagnostics(framing.response_correlation, &diagnostics); + if let Err(failure) = &output { + observe_response_admission_exhaustion( + framing.response_correlation, + failure.admission_exhausted(), + ); + } if let Some(guard) = framing.generation_guard { guard.ensure_current()?; } - let diagnostics = session.take_diagnostics(); if output.is_ok() { emit_http_response_diagnostics( framing.policy_name, framing.target, framing.status_code, &diagnostics, + false, + None, + framing.response_correlation, ); } let output = output.map_err(|failure| ResponseMiddlewareStop::new(failure, diagnostics))?; @@ -2240,6 +2547,7 @@ async fn deliver_response_units( .write_all(framing.commit_head) .await .map_err(ResponseDownstreamWrite)?; + record_response_delivery_started(framing.response_correlation); client.flush().await.map_err(ResponseDownstreamWrite)?; } if *framing.committed { @@ -2494,6 +2802,7 @@ mod tests { input_size: 0, output_size: None, failed, + transformed: false, stage_disabled: false, reason_code: None, failure_category: None, @@ -2526,6 +2835,7 @@ mod tests { &HttpRequestTarget::default(), 200, committed, + None, ); }) .await; @@ -2560,6 +2870,9 @@ mod tests { &HttpRequestTarget::default(), 200, &diagnostics, + denied, + None, + None, ); }) .await; @@ -2614,20 +2927,24 @@ mod tests { use super::{ DEFAULT_HTTP_RESPONSE_WHOLE_BODY_TIMEOUT, HttpResponseMiddlewareRelay, - RelayResponseOptions, ResponseFraming, parse_connection_keep_alive, - parse_response_head_for_middleware, relay_response, response_is_http_10, - serialize_response_head, validate_http_response_header_block, + RelayResponseOptions, ResponseFraming, http_response_diagnostic_truncation_event, + parse_connection_keep_alive, parse_response_head_for_middleware, relay_response, + response_is_http_10, response_middleware_observation, serialize_response_head, + validate_http_response_header_block, }; + use crate::l7::correlation::MiddlewareObservation; use crate::l7::provider::{BodyLength, L7Request, RelayOutcome}; use crate::l7::relay::L7EvalContext; use openshell_core::proto::{ - Decision, HttpHeader, HttpRequestResult, HttpRequestTarget, HttpResponseBlockDelivery, - HttpResponseBodyMode, HttpResponseBodyResult, HttpResponseBodyTransform, HttpResponseEvent, + Decision, ExistingHeaderAction, HeaderMutation, HttpHeader, HttpRequestResult, + HttpRequestTarget, HttpResponseBlockDelivery, HttpResponseBodyMode, HttpResponseBodyResult, + HttpResponseBodySkipRemaining, HttpResponseBodyTransform, HttpResponseEvent, HttpResponseEventResult, HttpResponsePreflightInspect, HttpResponsePreflightResult, HttpResponseTrailersResult, MiddlewareBinding, MiddlewareManifest, RequestContext, - SupervisorMiddlewareOperation, SupervisorMiddlewarePhase, http_response_body_result, - http_response_body_transform, http_response_body_unit, http_response_event, - http_response_event_result, http_response_preflight_result, + SupervisorMiddlewareOperation, SupervisorMiddlewarePhase, WriteHeader, header_mutation, + http_response_body_result, http_response_body_skip_remaining, http_response_body_transform, + http_response_body_unit, http_response_event, http_response_event_result, + http_response_preflight_result, }; use std::collections::HashMap; use std::sync::Arc; @@ -2638,47 +2955,155 @@ mod tests { struct BlockingResponseMiddleware; + struct RequestOnlyMiddleware; + #[tonic::async_trait] - impl openshell_supervisor_middleware::InProcessMiddleware for BlockingResponseMiddleware { + impl openshell_supervisor_middleware::InProcessMiddleware for RequestOnlyMiddleware { async fn describe(&self) -> MiddlewareManifest { - MiddlewareManifest { - name: "test/blocking-response".into(), - service_version: "test".into(), - bindings: vec![MiddlewareBinding { - operation: SupervisorMiddlewareOperation::HttpResponse as i32, - phase: SupervisorMiddlewarePhase::PreReturn as i32, - max_payload_bytes: 4096, - request_timeout: None, - }], - expected_audience: String::new(), - extension: Some(openshell_core::extension_protocol::extension_metadata( - openshell_core::extension_protocol::ExtensionFamily::SupervisorMiddleware, - "openshell/test-blocking-response", - "test", - [], - )), - } + let mut manifest = openshell_supervisor_middleware::InProcessMiddleware::describe( + &BlockingResponseMiddleware, + ) + .await; + manifest.name = "test/request-only".into(); + manifest.bindings[0].operation = SupervisorMiddlewareOperation::HttpRequest as i32; + manifest.bindings[0].phase = SupervisorMiddlewarePhase::PreCredentials as i32; + manifest } - async fn validate_config( - &self, - _middleware_name: &str, - _config: &prost_types::Struct, - ) -> miette::Result<()> { + async fn validate_config(&self, _: &str, _: &prost_types::Struct) -> miette::Result<()> { Ok(()) } async fn evaluate_http_request( &self, - _request: openshell_supervisor_middleware::HttpRequestView<'_>, + _: openshell_supervisor_middleware::HttpRequestView<'_>, ) -> miette::Result { - Ok(HttpRequestResult { - decision: Decision::Allow as i32, - ..Default::default() - }) + panic!("request hook must not run on a response") } + } - async fn open_http_response_pre_return( + #[tokio::test] + async fn an_absent_response_binding_does_not_create_a_cancelled_phase() { + use tracing_subscriber::layer::SubscriberExt as _; + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry() + .with(openshell_otel::layer(&provider, "no-response-hook")); + let _subscriber = tracing::subscriber::set_default(subscriber); + let correlation = test_correlation("request-only"); + let runner = + openshell_supervisor_middleware::ChainRunner::new(Arc::new(RequestOnlyMiddleware)); + let chain = vec![openshell_supervisor_middleware::ChainEntry { + name: "request".into(), + implementation: "test/request-only".into(), + order: 0, + config: prost_types::Struct::default(), + on_error: openshell_supervisor_middleware::OnError::FailClosed, + }]; + let mut upstream = &b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nbody"[..]; + let mut output = Vec::new(); + relay_response( + "GET", + &mut upstream, + &mut output, + RelayResponseOptions { + correlation: Some(&correlation), + ..Default::default() + }, + Some(test_response_middleware(&runner, &chain, "request-only")), + ) + .await + .unwrap(); + drop(correlation); + assert!(output.ends_with(b"\r\n\r\nbody")); + assert!( + !exporter + .get_finished_spans() + .unwrap() + .iter() + .any(|span| span.name == "openshell.middleware.response") + ); + } + + fn test_response_middleware<'a>( + runner: &'a openshell_supervisor_middleware::ChainRunner, + chain: &'a [openshell_supervisor_middleware::ChainEntry], + request_id: &str, + ) -> HttpResponseMiddlewareRelay<'a> { + HttpResponseMiddlewareRelay { + chain, + runner, + request_context: RequestContext { + request_id: request_id.into(), + sandbox_id: "sandbox-1".into(), + ..Default::default() + }, + target: HttpRequestTarget { + scheme: "https".into(), + host: "example.test".into(), + port: 443, + method: "GET".into(), + path: "/data".into(), + query: String::new(), + }, + policy_name: "test-policy", + generation_guard: None, + whole_body_timeout: DEFAULT_HTTP_RESPONSE_WHOLE_BODY_TIMEOUT, + } + } + + enum BodyTestAction { + Block, + ReloadAndSkipTransform(Arc), + } + + struct BodyResponseMiddleware { + action: BodyTestAction, + } + + #[tonic::async_trait] + impl openshell_supervisor_middleware::InProcessMiddleware for BlockingResponseMiddleware { + async fn describe(&self) -> MiddlewareManifest { + MiddlewareManifest { + name: "test/blocking-response".into(), + service_version: "test".into(), + bindings: vec![MiddlewareBinding { + operation: SupervisorMiddlewareOperation::HttpResponse as i32, + phase: SupervisorMiddlewarePhase::PreReturn as i32, + max_payload_bytes: 4096, + request_timeout: None, + }], + expected_audience: String::new(), + extension: Some(openshell_core::extension_protocol::extension_metadata( + openshell_core::extension_protocol::ExtensionFamily::SupervisorMiddleware, + "openshell/test-blocking-response", + "test", + [], + )), + } + } + + async fn validate_config( + &self, + _middleware_name: &str, + _config: &prost_types::Struct, + ) -> miette::Result<()> { + Ok(()) + } + + async fn evaluate_http_request( + &self, + _request: openshell_supervisor_middleware::HttpRequestView<'_>, + ) -> miette::Result { + Ok(HttpRequestResult { + decision: Decision::Allow as i32, + ..Default::default() + }) + } + + async fn open_http_response_pre_return( &self, mut requests: mpsc::Receiver, ) -> Result @@ -2765,7 +3190,16 @@ mod tests { action: Some(http_response_preflight_result::Action::Inspect( HttpResponsePreflightInspect { body_mode: HttpResponseBodyMode::StreamBytes as i32, - header_mutations: Vec::new(), + header_mutations: vec![HeaderMutation { + operation: Some(header_mutation::Operation::Write( + WriteHeader { + name: "x-inspected".into(), + value: "true".into(), + on_existing: ExistingHeaderAction::Overwrite + as i32, + }, + )), + }], }, )), ..Default::default() @@ -2811,6 +3245,137 @@ mod tests { } } + #[tonic::async_trait] + impl openshell_supervisor_middleware::InProcessMiddleware for BodyResponseMiddleware { + async fn describe(&self) -> MiddlewareManifest { + MiddlewareManifest { + name: "test/body-response".into(), + service_version: "test".into(), + bindings: vec![MiddlewareBinding { + operation: SupervisorMiddlewareOperation::HttpResponse as i32, + phase: SupervisorMiddlewarePhase::PreReturn as i32, + max_payload_bytes: 4096, + request_timeout: None, + }], + expected_audience: String::new(), + extension: Some(openshell_core::extension_protocol::extension_metadata( + openshell_core::extension_protocol::ExtensionFamily::SupervisorMiddleware, + "openshell/test-body-response", + "test", + [], + )), + } + } + + async fn validate_config( + &self, + _middleware_name: &str, + _config: &prost_types::Struct, + ) -> miette::Result<()> { + Ok(()) + } + + async fn evaluate_http_request( + &self, + _request: openshell_supervisor_middleware::HttpRequestView<'_>, + ) -> miette::Result { + Ok(HttpRequestResult { + decision: Decision::Allow as i32, + ..Default::default() + }) + } + + async fn open_http_response_pre_return( + &self, + mut requests: mpsc::Receiver, + ) -> Result + { + let (sender, receiver) = mpsc::channel(4); + let action = match &self.action { + BodyTestAction::Block => BodyTestAction::Block, + BodyTestAction::ReloadAndSkipTransform(engine) => { + BodyTestAction::ReloadAndSkipTransform(Arc::clone(engine)) + } + }; + tokio::spawn(async move { + while let Some(event) = requests.recv().await { + let Some(event) = event.event else { + break; + }; + let result = match event { + http_response_event::Event::Preflight(_) => HttpResponseEventResult { + result: Some(http_response_event_result::Result::PreflightResult( + HttpResponsePreflightResult { + action: Some(http_response_preflight_result::Action::Inspect( + HttpResponsePreflightInspect { + body_mode: HttpResponseBodyMode::StreamBytes as i32, + header_mutations: Vec::new(), + }, + )), + ..Default::default() + }, + )), + }, + http_response_event::Event::Body(body) => { + let Some(http_response_body_unit::Payload::Data(data)) = body.payload + else { + break; + }; + let action = match &action { + BodyTestAction::Block => { + http_response_body_result::Action::BlockDelivery( + HttpResponseBlockDelivery {}, + ) + } + BodyTestAction::ReloadAndSkipTransform(engine) => { + engine + .reload( + include_str!("../../../data/sandbox-policy.rego"), + "network_policies: {}\n", + ) + .expect("reload test policy"); + http_response_body_result::Action::SkipRemaining( + HttpResponseBodySkipRemaining { + current: Some( + http_response_body_skip_remaining::Current::Transform( + HttpResponseBodyTransform { + replacement: Some( + http_response_body_transform::Replacement::Data(data), + ), + }, + ), + ), + }, + ) + } + }; + HttpResponseEventResult { + result: Some(http_response_event_result::Result::BodyResult( + HttpResponseBodyResult { + sequence: body.sequence, + action: Some(action), + reason_code: "test_body_action".into(), + ..Default::default() + }, + )), + } + } + http_response_event::Event::Trailers(_) => HttpResponseEventResult { + result: Some(http_response_event_result::Result::TrailersResult( + HttpResponseTrailersResult::default(), + )), + }, + http_response_event::Event::SessionEnd(_) => break, + }; + if sender.send(Ok(result)).await.is_err() { + break; + } + } + }); + Ok(Box::pin(ReceiverStream::new(receiver))) + } + } + fn test_correlation_for_method( request_id: &str, method: &str, @@ -2875,20 +3440,23 @@ mod tests { } #[tokio::test] - async fn response_middleware_denial_records_that_upstream_was_reached() { + async fn branch08_response_middleware_denial_records_that_upstream_was_reached() { use tracing_subscriber::layer::SubscriberExt as _; let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() .with_simple_exporter(exporter.clone()) .build(); - let subscriber = tracing_subscriber::registry().with(openshell_otel::layer( - &provider, - "response-correlation-test", - )); - let correlation = tracing::subscriber::with_default(subscriber, || { - test_correlation_for_method("response-denied", "POST") - }); + let ocsf = openshell_ocsf::test_support::OcsfCapture::default(); + let subscriber = tracing_subscriber::registry() + .with(openshell_otel::layer( + &provider, + "response-correlation-test", + )) + .with(ocsf.clone()); + let _subscriber = tracing::subscriber::set_default(subscriber); + tracing::callsite::rebuild_interest_cache(); + let correlation = test_correlation_for_method("response-denied", "POST"); let runner = openshell_supervisor_middleware::ChainRunner::new(Arc::new(BlockingResponseMiddleware)); let chain = vec![openshell_supervisor_middleware::ChainEntry { @@ -2968,6 +3536,365 @@ mod tests { assert!(!attributes.contains_key("openshell.error.stage")); } + #[tokio::test] + async fn streaming_response_middleware_aggregates_header_and_body_mutations() { + use tracing_subscriber::layer::SubscriberExt as _; + + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry().with(openshell_otel::layer( + &provider, + "response-correlation-test", + )); + let correlation = tracing::subscriber::with_default(subscriber, || { + test_correlation_for_method("response-streaming", "GET") + }); + let runner = openshell_supervisor_middleware::ChainRunner::new(Arc::new( + StreamingResponseMiddleware, + )); + let chain = vec![openshell_supervisor_middleware::ChainEntry { + name: "response".into(), + implementation: "test/streaming-response".into(), + order: 0, + config: prost_types::Struct::default(), + on_error: openshell_supervisor_middleware::OnError::FailClosed, + }]; + let middleware = HttpResponseMiddlewareRelay { + chain: &chain, + runner: &runner, + request_context: RequestContext { + request_id: "response-streaming".into(), + sandbox_id: "sandbox-1".into(), + ..Default::default() + }, + target: HttpRequestTarget { + scheme: "https".into(), + host: "example.test".into(), + port: 443, + method: "GET".into(), + path: "/data".into(), + query: String::new(), + }, + policy_name: "test-policy", + generation_guard: None, + whole_body_timeout: DEFAULT_HTTP_RESPONSE_WHOLE_BODY_TIMEOUT, + }; + let body = vec![b'a'; 9_000]; + let mut response = + format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).into_bytes(); + response.extend_from_slice(&body); + let mut upstream = response.as_slice(); + let mut client = Vec::new(); + + let outcome = relay_response( + "GET", + &mut upstream, + &mut client, + RelayResponseOptions { + correlation: Some(&correlation), + ..Default::default() + }, + Some(middleware), + ) + .await + .expect("streaming middleware response"); + + assert!(matches!(outcome, RelayOutcome::Reusable)); + assert!(String::from_utf8_lossy(&client).contains("x-inspected: true\r\n")); + assert!( + String::from_utf8_lossy(&client) + .contains("x-openshell-request-id: response-streaming\r\n") + ); + drop(correlation); + + let span = exporter + .get_finished_spans() + .expect("finished spans") + .into_iter() + .find(|span| span.name == "openshell.middleware.response") + .expect("HTTP enforcement span"); + let attributes = span + .attributes + .into_iter() + .map(|attribute| { + ( + attribute.key.as_str().to_string(), + attribute.value.to_string(), + ) + }) + .collect::>(); + assert_eq!(attributes["openshell.middleware.outcome"], "allowed"); + assert_eq!(attributes["openshell.middleware.mutated"], "true"); + assert_eq!(attributes["openshell.middleware.fail_open"], "false"); + assert!(!attributes.contains_key("openshell.request.id")); + } + + #[tokio::test] + async fn streaming_body_denial_records_partial_delivery() { + use tracing_subscriber::layer::SubscriberExt as _; + + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let ocsf = openshell_ocsf::test_support::OcsfCapture::default(); + let subscriber = tracing_subscriber::registry() + .with(openshell_otel::layer( + &provider, + "response-correlation-test", + )) + .with(ocsf.clone()); + let _subscriber = tracing::subscriber::set_default(subscriber); + tracing::callsite::rebuild_interest_cache(); + let correlation = test_correlation_for_method("response-partial-denial", "GET"); + let runner = + openshell_supervisor_middleware::ChainRunner::new(Arc::new(BodyResponseMiddleware { + action: BodyTestAction::Block, + })); + let chain = vec![openshell_supervisor_middleware::ChainEntry { + name: "response".into(), + implementation: "test/body-response".into(), + order: 0, + config: prost_types::Struct::default(), + on_error: openshell_supervisor_middleware::OnError::FailClosed, + }]; + let middleware = HttpResponseMiddlewareRelay { + chain: &chain, + runner: &runner, + request_context: RequestContext { + request_id: "response-partial-denial".into(), + sandbox_id: "sandbox-1".into(), + ..Default::default() + }, + target: HttpRequestTarget { + scheme: "https".into(), + host: "example.test".into(), + port: 443, + method: "GET".into(), + path: "/data".into(), + query: String::new(), + }, + policy_name: "test-policy", + generation_guard: None, + whole_body_timeout: DEFAULT_HTTP_RESPONSE_WHOLE_BODY_TIMEOUT, + }; + let mut upstream = &b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nbody"[..]; + let mut client = Vec::new(); + + let error = relay_response( + "GET", + &mut upstream, + &mut client, + RelayResponseOptions { + correlation: Some(&correlation), + ..Default::default() + }, + Some(middleware), + ) + .await + .expect_err("streaming body denial follows response commitment"); + assert!(error.to_string().contains("stopped delivery"), "{error}"); + assert!(client.starts_with(b"HTTP/1.1 200 OK\r\n")); + drop(correlation); + provider.force_flush().expect("flush spans"); + + let span = exporter + .get_finished_spans() + .expect("finished spans") + .into_iter() + .find(|span| span.name == "openshell.middleware.response") + .expect("HTTP enforcement span"); + let attributes = span + .attributes + .into_iter() + .map(|attribute| { + ( + attribute.key.as_str().to_string(), + attribute.value.to_string(), + ) + }) + .collect::>(); + assert_eq!(attributes["openshell.middleware.outcome"], "denied"); + assert_eq!(attributes["openshell.response.delivery_started"], "true"); + assert_eq!(attributes["openshell.response.partial_delivery"], "true"); + assert!( + !ocsf + .json_events() + .iter() + .any(|event| event["firewall_rule"]["type"] == "middleware-correlation"), + "routine middleware aggregates are not separate audit records" + ); + } + + #[tokio::test] + async fn stale_generation_preserves_skip_transform_observation() { + use tracing_subscriber::layer::SubscriberExt as _; + + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry().with(openshell_otel::layer( + &provider, + "response-correlation-test", + )); + let correlation = tracing::subscriber::with_default(subscriber, || { + test_correlation_for_method("response-stale-transform", "GET") + }); + let engine = Arc::new( + crate::opa::OpaEngine::from_strings( + include_str!("../../../data/sandbox-policy.rego"), + "network_policies: {}\n", + ) + .expect("test policy"), + ); + let generation_guard = engine + .generation_guard(engine.current_generation()) + .expect("generation guard"); + let runner = + openshell_supervisor_middleware::ChainRunner::new(Arc::new(BodyResponseMiddleware { + action: BodyTestAction::ReloadAndSkipTransform(Arc::clone(&engine)), + })); + let chain = vec![openshell_supervisor_middleware::ChainEntry { + name: "response".into(), + implementation: "test/body-response".into(), + order: 0, + config: prost_types::Struct::default(), + on_error: openshell_supervisor_middleware::OnError::FailClosed, + }]; + let middleware = HttpResponseMiddlewareRelay { + chain: &chain, + runner: &runner, + request_context: RequestContext { + request_id: "response-stale-transform".into(), + sandbox_id: "sandbox-1".into(), + ..Default::default() + }, + target: HttpRequestTarget { + scheme: "https".into(), + host: "example.test".into(), + port: 443, + method: "GET".into(), + path: "/data".into(), + query: String::new(), + }, + policy_name: "test-policy", + generation_guard: Some(&generation_guard), + whole_body_timeout: DEFAULT_HTTP_RESPONSE_WHOLE_BODY_TIMEOUT, + }; + let mut upstream = &b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nbody"[..]; + let mut client = Vec::new(); + + let error = relay_response( + "GET", + &mut upstream, + &mut client, + RelayResponseOptions { + correlation: Some(&correlation), + ..Default::default() + }, + Some(middleware), + ) + .await + .expect_err("policy reload must stop the stale response"); + assert!( + error.to_string().contains("policy generation is stale"), + "{error}" + ); + drop(correlation); + provider.force_flush().expect("flush spans"); + + let span = exporter + .get_finished_spans() + .expect("finished spans") + .into_iter() + .find(|span| span.name == "openshell.middleware.response") + .expect("HTTP enforcement span"); + let attributes = span + .attributes + .into_iter() + .map(|attribute| { + ( + attribute.key.as_str().to_string(), + attribute.value.to_string(), + ) + }) + .collect::>(); + assert_eq!(attributes["openshell.middleware.outcome"], "incomplete"); + assert_eq!(attributes["openshell.middleware.mutated"], "true"); + assert_eq!(attributes["openshell.response.delivery_started"], "true"); + assert_eq!(attributes["openshell.response.partial_delivery"], "true"); + } + + #[test] + fn response_middleware_observation_maps_bounded_outcomes() { + let invocation = |outcome| HttpResponseInvocation { + config_name: "not-exported".into(), + implementation: "not-exported".into(), + outcome, + sequence: None, + input_size: 17, + output_size: None, + transformed: outcome + == openshell_supervisor_middleware::HttpResponseInvocationOutcome::Transform, + failed: false, + stage_disabled: false, + reason_code: None, + failure_category: None, + }; + let mut invocations = vec![ + invocation(openshell_supervisor_middleware::HttpResponseInvocationOutcome::Transform), + invocation(openshell_supervisor_middleware::HttpResponseInvocationOutcome::FailOpen), + invocation(openshell_supervisor_middleware::HttpResponseInvocationOutcome::FailClosed), + ]; + let mut skip_transform = invocation( + openshell_supervisor_middleware::HttpResponseInvocationOutcome::SkipRemaining, + ); + skip_transform.transformed = true; + invocations.push(skip_transform); + let observation = response_middleware_observation(&invocations, false, true); + + assert_eq!( + observation, + MiddlewareObservation { + mutated: true, + fail_open: true, + fail_closed: true, + admission_exhausted: true, + ..Default::default() + } + ); + } + + #[test] + fn response_diagnostic_truncation_event_is_bounded() { + let event = http_response_diagnostic_truncation_event( + "test-policy", + &HttpRequestTarget { + scheme: "https".into(), + host: "example.test".into(), + port: 443, + method: "GET".into(), + path: "/private".into(), + query: "secret=value".into(), + }, + ) + .to_json() + .expect("serialize truncation event"); + + assert_eq!(event["severity"], "Low"); + assert_eq!( + event["finding_info"]["uid"], + "openshell.middleware.http_response_diagnostics_truncated" + ); + assert_eq!(event["unmapped"]["phase"], "pre_return"); + let encoded = serde_json::to_string(&event).expect("encode truncation event"); + assert!(!encoded.contains("/private")); + assert!(!encoded.contains("secret=value")); + } + #[tokio::test] async fn upstream_receipt_is_replaced_by_platform_receipt() { let correlation = test_correlation("trusted"); diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index b79c0caa60..fa47354e14 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -1919,6 +1919,7 @@ struct ForwardMiddlewarePipeline<'a> { scheme: &'a str, exchange: &'a crate::l7::middleware::HttpMiddlewareExchange, l7_reevaluation: Option>, + correlation: Option<&'a crate::l7::correlation::RequestCorrelation>, } impl ForwardMiddlewarePipeline<'_> { @@ -1955,6 +1956,7 @@ impl ForwardMiddlewarePipeline<'_> { self.ctx, self.scheme, transformed_body_policy, + self.correlation, ) .await } @@ -6646,6 +6648,7 @@ async fn handle_forward_proxy( scheme: &scheme, exchange: &middleware_exchange, l7_reevaluation, + correlation: Some(&correlation), }; // Middleware denials join the connect trace, like other FORWARD denials. forward_request_bytes = match pipeline @@ -6654,16 +6657,17 @@ async fn handle_forward_proxy( .await? { crate::l7::middleware::MiddlewareApplyResult::Allowed(request) => request.raw_header, - crate::l7::middleware::MiddlewareApplyResult::Denied { denial, .. } => { + crate::l7::middleware::MiddlewareApplyResult::Denied { + denial, + termination, + } => { emit_activity_simple(activity_tx, true, "middleware"); correlation.record_enforcement_action("block"); correlation.record_response_status(403); - correlation.record_final_http_outcome(if denial.is_some() { - "denied" - } else { - "platform_error" - }); - correlation.record_error_stage("request_middleware"); + let (outcome, stage) = + crate::l7::middleware::request_rejection_correlation(termination); + correlation.record_final_http_outcome(outcome); + correlation.record_error_stage(stage); let response = denial.as_ref().map_or_else( || { build_middleware_failure_response( @@ -6712,6 +6716,7 @@ async fn handle_forward_proxy( &middleware_runner, &l7_ctx, "ws", + Some(&correlation), ) .await; let preflight = match preflight { @@ -6736,7 +6741,11 @@ async fn handle_forward_proxy( return Ok(()); } }; - crate::l7::middleware::emit_websocket_preflight_events(&l7_ctx, &preflight); + crate::l7::middleware::emit_websocket_preflight_events( + &l7_ctx, + &preflight, + Some(&correlation), + ); if preflight.terminal_reason.is_some() { correlation.record_enforcement_action("block"); correlation.record_response_status(403); @@ -11624,6 +11633,7 @@ network_policies: engine: &tunnel_engine, request_info: &request_info, }), + correlation: None, }; let (_app, mut client) = tokio::io::duplex(8192); @@ -11633,8 +11643,19 @@ network_policies: .expect("forward middleware pipeline"); match outcome { - crate::l7::middleware::MiddlewareApplyResult::Denied { denial } => { + crate::l7::middleware::MiddlewareApplyResult::Denied { + denial, + termination, + } => { assert!(denial.is_none()); + assert_eq!( + termination, + openshell_supervisor_middleware::ChainTermination::TransformedBodyPolicyDenied + ); + assert_eq!( + crate::l7::middleware::request_rejection_correlation(termination), + ("denied", "post_middleware_policy") + ); } crate::l7::middleware::MiddlewareApplyResult::Allowed(_) => { panic!("policy-invalid transformed request must be denied") @@ -11723,6 +11744,7 @@ network_policies: scheme: "http", exchange: &exchange, l7_reevaluation: None, + correlation: None, }; let (_app, mut client) = tokio::io::duplex(8192); let revoke = async { @@ -14628,6 +14650,7 @@ network_policies: scheme: "http", exchange: &exchange, l7_reevaluation: None, + correlation: None, }; let (_app, mut client) = tokio::io::duplex(8192); diff --git a/docs/observability/traces.mdx b/docs/observability/traces.mdx index 0cf8ca3c27..1b89c6666d 100644 --- a/docs/observability/traces.mdx +++ b/docs/observability/traces.mdx @@ -73,6 +73,27 @@ The gateway continues a valid W3C `traceparent` sent by the caller, so a client' The gateway also passes its trace context to compute drivers, so driver spans appear in the same trace as the gateway request that caused them. Built-in drivers export their spans through the gateway's process. Managed external drivers receive the trace endpoint, gateway name, and compute driver from the gateway. Drivers that you run yourself own their telemetry configuration. +### HTTP Enforcement Correlation + +Inspected HTTP requests create an `openshell.http.request` span. Policy evaluation +and selected request and response middleware phases appear as children. A valid, +sampled W3C `traceparent` links the request to the caller's trace. Malformed or +duplicate headers are ignored; an unsampled context becomes a link rather than +disabling local enforcement telemetry. OpenShell does not adopt `tracestate`. + +OpenShell assigns its own request ID, includes it in the span and related OCSF +records, and returns it as `X-OpenShell-Request-ID`. This header is a correlation +handle, not authorization or proof of delivery. Match it to trusted OpenShell +output for the same sandbox. L4 and uninspected passthrough traffic do not provide +this receipt guarantee. + +The primary HTTP Activity record describes the final response headers accepted +by the client connection. Response-body completion is recorded separately in the +trace. A +successful header write does not establish that the caller received a complete +body or that the external service committed an operation. Correlation attributes +exclude credentials, query values, bodies and MCP arguments. + ## Next Steps - Configure [OTLP export](/how-it-works/gateways/configuration#otlp-export), including sampling and the `OTEL_*` variables the gateway honors.