From 5416e48cb7add756cbea89539c99365c77ef483f Mon Sep 17 00:00:00 2001 From: Alex Fournier Date: Fri, 6 Nov 2026 10:50:23 -0700 Subject: [PATCH] feat(observability): correlate MCP inspection as request child spans Signed-off-by: Alex Fournier --- crates/openshell-supervisor-network/README.md | 5 + .../src/l7/correlation.rs | 278 +++++++++++- .../src/l7/jsonrpc.rs | 24 +- .../src/l7/relay.rs | 400 ++++++++++++++++-- .../openshell-supervisor-network/src/proxy.rs | 65 ++- docs/observability/traces.mdx | 7 +- 6 files changed, 727 insertions(+), 52 deletions(-) diff --git a/crates/openshell-supervisor-network/README.md b/crates/openshell-supervisor-network/README.md index 105519cd44..1c68428645 100644 --- a/crates/openshell-supervisor-network/README.md +++ b/crates/openshell-supervisor-network/README.md @@ -113,3 +113,8 @@ not whichever revision is current when an operator later reads its records. Revision values remain decimal strings so full-width values survive export. OCSF stores these request configuration identities in `unmapped`; they are not firewall-rule versions or claims about the currently active configuration. + +JSON-RPC and MCP inspection use child spans for bounded message classification, +counts, protocol revision and inspection result. Reinspection after middleware +records the effective view without exposing arguments, tool names or raw errors. +The HTTP request owns the correlation ID before dedicated MCP body inspection. diff --git a/crates/openshell-supervisor-network/src/l7/correlation.rs b/crates/openshell-supervisor-network/src/l7/correlation.rs index 506e7469f0..c3efdb327b 100644 --- a/crates/openshell-supervisor-network/src/l7/correlation.rs +++ b/crates/openshell-supervisor-network/src/l7/correlation.rs @@ -20,6 +20,8 @@ use tracing::Instrument as _; use crate::l7::provider::L7Request; use crate::l7::relay::L7EvalContext; +use crate::l7::{L7Protocol, jsonrpc}; +const JSONRPC_CORRELATED_COUNT_LIMIT: usize = 64; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; #[derive(Debug)] @@ -38,11 +40,33 @@ struct RequestState { outcome: Option<&'static str>, policy_span: Option, middleware: [Option; 2], + jsonrpc: JsonRpcCorrelationState, + mcp_span: Option, resources: ResourceCorrelationState, credential: CredentialState, credential_span: Option, } +#[derive(Debug, Default)] +struct JsonRpcCorrelationState { + effective: Option, + selected_mcp_revision: Option, + mcp_tools_call_observed: bool, +} + +#[derive(Debug, Clone, Copy)] +struct JsonRpcEffectiveSummary { + protocol: &'static str, + message_kind: &'static str, + inspection_outcome: &'static str, + call_count: i64, + notification_count: i64, + counts_truncated: bool, + has_response: bool, + mcp_revision: Option, + mcp_method_classification: Option<&'static str>, +} + #[derive(Debug, Default, PartialEq, Eq)] struct ResourceCorrelationState { policy_runtime_generation: Option, @@ -385,6 +409,8 @@ impl RequestCorrelation { outcome: None, policy_span: None, middleware: [None, None], + jsonrpc: JsonRpcCorrelationState::default(), + mcp_span: None, resources: ResourceCorrelationState::default(), credential: CredentialState::default(), credential_span: None, @@ -543,6 +569,187 @@ impl RequestCorrelation { ); } + pub(crate) fn record_effective_jsonrpc_summary( + &self, + protocol: L7Protocol, + info: &jsonrpc::JsonRpcRequestInfo, + ) { + let protocol_name = match protocol { + L7Protocol::JsonRpc => "jsonrpc", + L7Protocol::Mcp => "mcp", + _ => return, + }; + let call_count = info + .calls + .iter() + .filter(|call| !call.is_notification) + .count(); + let notification_count = info + .calls + .iter() + .filter(|call| call.is_notification) + .count(); + let counts_truncated = call_count > JSONRPC_CORRELATED_COUNT_LIMIT + || notification_count > JSONRPC_CORRELATED_COUNT_LIMIT; + + { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let state = &mut state.jsonrpc; + if protocol == L7Protocol::Mcp { + if let Some(revision) = info.mcp_revision { + state.selected_mcp_revision = Some(revision); + } + state.mcp_tools_call_observed |= + info.calls.iter().any(|call| call.method == "tools/call"); + } + state.effective = Some(JsonRpcEffectiveSummary { + protocol: protocol_name, + message_kind: jsonrpc_message_kind(info, call_count, notification_count), + inspection_outcome: jsonrpc_inspection_outcome(info), + call_count: i64::try_from(call_count.min(JSONRPC_CORRELATED_COUNT_LIMIT)) + .expect("bounded JSON-RPC call count fits i64"), + notification_count: i64::try_from( + notification_count.min(JSONRPC_CORRELATED_COUNT_LIMIT), + ) + .expect("bounded JSON-RPC notification count fits i64"), + counts_truncated, + has_response: info.has_response, + mcp_revision: if protocol == L7Protocol::Mcp { + info.mcp_revision + } else { + None + }, + mcp_method_classification: (protocol == L7Protocol::Mcp) + .then(|| mcp_method_classification(info)), + }); + } + self.record_jsonrpc_attributes(); + } + + fn record_jsonrpc_attributes(&self) { + let span = self.take_mcp_span(); + let (summary, selected_revision, tools_call_observed) = { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let state = &state.jsonrpc; + let Some(summary) = state.effective else { + return; + }; + ( + summary, + state.selected_mcp_revision, + state.mcp_tools_call_observed, + ) + }; + + span.record("openshell.l7.protocol", summary.protocol); + span.record( + "openshell.jsonrpc.effective.message_kind", + summary.message_kind, + ); + span.record( + "openshell.jsonrpc.effective.inspection_outcome", + summary.inspection_outcome, + ); + span.record("openshell.jsonrpc.effective.call_count", summary.call_count); + span.record( + "openshell.jsonrpc.effective.notification_count", + summary.notification_count, + ); + span.record( + "openshell.jsonrpc.effective.counts_truncated", + summary.counts_truncated, + ); + span.record( + "openshell.jsonrpc.effective.has_response", + summary.has_response, + ); + + if summary.protocol == "mcp" { + if let Some(revision) = selected_revision { + span.record( + "openshell.mcp.selected_protocol_revision", + revision.as_str(), + ); + } + if let Some(revision) = summary.mcp_revision { + span.record( + "openshell.mcp.effective.protocol_revision", + revision.as_str(), + ); + } + if let Some(classification) = summary.mcp_method_classification { + span.record( + "openshell.mcp.effective.method_classification", + classification, + ); + } + span.record("openshell.mcp.tools_call_observed", tools_call_observed); + } + } + + fn new_mcp_span(&self) -> tracing::Span { + self.span + .with_subscriber(|(_, dispatch)| { + tracing::dispatcher::with_default(dispatch, || { + tracing::info_span!(parent: &self.span, "openshell.mcp.inspect", + otel.status_code = tracing::field::Empty, + "openshell.l7.protocol" = tracing::field::Empty, + "openshell.jsonrpc.effective.message_kind" = tracing::field::Empty, + "openshell.jsonrpc.effective.inspection_outcome" = tracing::field::Empty, + "openshell.jsonrpc.effective.call_count" = tracing::field::Empty, + "openshell.jsonrpc.effective.notification_count" = tracing::field::Empty, + "openshell.jsonrpc.effective.counts_truncated" = tracing::field::Empty, + "openshell.jsonrpc.effective.has_response" = tracing::field::Empty, + "openshell.mcp.selected_protocol_revision" = tracing::field::Empty, + "openshell.mcp.effective.protocol_revision" = tracing::field::Empty, + "openshell.mcp.effective.method_classification" = tracing::field::Empty, + "openshell.mcp.tools_call_observed" = tracing::field::Empty, + ) + }) + }) + .unwrap_or_else(tracing::Span::none) + } + + fn take_mcp_span(&self) -> tracing::Span { + let span = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .mcp_span + .take(); + span.unwrap_or_else(|| self.new_mcp_span()) + } + + pub(crate) fn within_mcp(&self, future: F) -> impl Future> + where + F: Future>, + { + let future = Box::pin(future); + async move { + let span = self.new_mcp_span(); + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .mcp_span = Some(span.clone()); + let result = future.instrument(span.clone()).await; + if result.is_err() { + openshell_otel::mark_error(&span); + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .mcp_span + .take(); + } + result + } + } + pub(crate) fn record_target_reached(&self) { self.span.record("openshell.target.reached", true); } @@ -966,8 +1173,10 @@ pub fn test_request_attributes( .map(|attribute| (attribute.key.to_string(), attribute.value.to_string())) .collect::>(); for child in spans.iter().filter(|span| { - span.name == "openshell.credential.prepare" - && span.parent_span_id == root.span_context.span_id() + matches!( + span.name.as_ref(), + "openshell.credential.prepare" | "openshell.mcp.inspect" + ) && span.parent_span_id == root.span_context.span_id() }) { assert_eq!(child.span_context.trace_id(), root.span_context.trace_id()); attributes.extend( @@ -1131,6 +1340,71 @@ fn add_resource_fields(event: &mut OcsfEvent, resources: &ResourceCorrelationSta } } +fn jsonrpc_message_kind( + info: &jsonrpc::JsonRpcRequestInfo, + call_count: usize, + notification_count: usize, +) -> &'static str { + if info.error.is_some() { + return "invalid"; + } + if info.receive_stream { + return "receive_stream"; + } + if info.is_batch { + return match (call_count > 0, notification_count > 0, info.has_response) { + (true, false, false) => "batch_calls", + (false, true, false) => "batch_notifications", + (false, false, true) => "batch_responses", + _ => "batch_mixed", + }; + } + match (call_count > 0, notification_count > 0, info.has_response) { + (true, false, false) => "call", + (false, true, false) => "notification", + (false, false, true) => "response", + (false, false, false) => "empty", + _ => "mixed", + } +} + +fn jsonrpc_inspection_outcome(info: &jsonrpc::JsonRpcRequestInfo) -> &'static str { + match info + .error + .as_ref() + .map(jsonrpc::JsonRpcInspectionError::kind) + { + None => "accepted", + Some(jsonrpc::JsonRpcInspectionErrorKind::InvalidJson) => "invalid_json", + Some(jsonrpc::JsonRpcInspectionErrorKind::InvalidMessage) => "invalid_message", + Some(jsonrpc::JsonRpcInspectionErrorKind::RevisionNotSelected) => "revision_not_selected", + Some(jsonrpc::JsonRpcInspectionErrorKind::McpProfileViolation) => "mcp_profile_violation", + Some(jsonrpc::JsonRpcInspectionErrorKind::McpLifecycleViolation) => { + "mcp_lifecycle_violation" + } + } +} + +fn mcp_method_classification(info: &jsonrpc::JsonRpcRequestInfo) -> &'static str { + let mut core = false; + let mut extension = false; + let mut unclassified = false; + for call in &info.calls { + match call.mcp_classification { + Some(jsonrpc::McpMethodClassification::Available) => core = true, + Some(jsonrpc::McpMethodClassification::Extension) => extension = true, + None => unclassified = true, + } + } + match (core, extension, unclassified) { + (false, false, false) => "none", + (true, false, false) => "core", + (false, true, false) => "extension", + (true, true, false) => "mixed", + _ => "unclassified", + } +} + fn bounded_http_method(method: &str) -> &'static str { [ "CONNECT", "DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT", "TRACE", diff --git a/crates/openshell-supervisor-network/src/l7/jsonrpc.rs b/crates/openshell-supervisor-network/src/l7/jsonrpc.rs index 6d767191fd..86ff70a9b5 100644 --- a/crates/openshell-supervisor-network/src/l7/jsonrpc.rs +++ b/crates/openshell-supervisor-network/src/l7/jsonrpc.rs @@ -18,7 +18,7 @@ use tower_mcp_types::{ protocol::{InitializeParams, JSONRPC_VERSION, McpRequest, RequestMeta, validate_meta_object}, }; -use crate::l7::provider::{L7Provider, L7Request}; +use crate::l7::provider::L7Request; pub const DEFAULT_MAX_BODY_BYTES: usize = 64 * 1024; @@ -114,26 +114,18 @@ pub struct JsonRpcHttpRequest { pub info: JsonRpcRequestInfo, } -pub(crate) async fn parse_jsonrpc_http_request( +pub(crate) async fn inspect_jsonrpc_http_request( client: &mut C, + request: &mut L7Request, max_body_bytes: usize, - canonicalize_options: crate::l7::path::CanonicalizeOptions, inspection_options: JsonRpcInspectionOptions, -) -> Result> { - let provider = crate::l7::rest::RestProvider::with_options(canonicalize_options); - let Some(mut request) = provider.parse_request(client).await? else { - return Ok(None); - }; - if jsonrpc_receive_stream_request(&request) { - return Ok(Some(JsonRpcHttpRequest { - request, - info: JsonRpcRequestInfo::receive_stream(), - })); +) -> Result { + if jsonrpc_receive_stream_request(request) { + return Ok(JsonRpcRequestInfo::receive_stream()); } - let body = - crate::l7::http::read_body_for_inspection(client, &mut request, max_body_bytes).await?; + let body = crate::l7::http::read_body_for_inspection(client, request, max_body_bytes).await?; let info = parse_jsonrpc_body_with_options(&body, inspection_options); - Ok(Some(JsonRpcHttpRequest { request, info })) + Ok(info) } /// Reinspect a fully buffered JSON-RPC-family HTTP request after middleware. diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index 75a5c18a30..45d022c961 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -343,9 +343,16 @@ where W: AsyncWrite + Unpin, { if config.protocol != L7Protocol::Mcp { + if let Some(correlation) = correlation { + correlation.record_effective_jsonrpc_summary(config.protocol, &info); + } return Ok(Some(info)); } + if let Some(correlation) = correlation { + correlation.record_effective_jsonrpc_summary(config.protocol, &info); + } + match crate::l7::mcp::select_request_protocol_version(request, &info, &config.mcp_versions) { Ok(crate::l7::mcp::McpRequestProtocolVersion::Initialization) => Ok(Some(info)), Ok(crate::l7::mcp::McpRequestProtocolVersion::Selected(version)) => { @@ -361,6 +368,9 @@ where // classify, but later middleware re-evaluation must still retain // the exact transport-selected revision. info.mcp_revision = Some(version); + if let Some(correlation) = correlation { + correlation.record_effective_jsonrpc_summary(config.protocol, &info); + } if let Some(error) = info.error.as_ref() { if let Some(observer) = observer { @@ -1657,17 +1667,16 @@ where if crate::l7::jsonrpc::jsonrpc_receive_stream_request(&req) { Some(crate::l7::jsonrpc::JsonRpcRequestInfo::receive_stream()) } else { - match crate::l7::http::read_body_for_inspection( - client, - &mut req, - config.json_rpc_max_body_bytes, - ) - .await - { - Ok(body) => Some(crate::l7::jsonrpc::parse_jsonrpc_body_with_options( - &body, + match correlation + .within_mcp(crate::l7::jsonrpc::inspect_jsonrpc_http_request( + client, + &mut req, + config.json_rpc_max_body_bytes, crate::l7::jsonrpc::JsonRpcInspectionOptions::for_config(config), - )), + )) + .await + { + Ok(info) => Some(info), Err(e) => { if is_benign_connection_error(&e) { debug!( @@ -1830,7 +1839,8 @@ where // body against that protocol's policy after every transforming // stage (a no-op for REST and websocket, whose policy inputs the // chain cannot mutate). - let validate = transformed_body_validator(config, &engine, ctx, &request_info); + let validate = + transformed_body_validator(config, &engine, ctx, &request_info, Some(&correlation)); let middleware_result = apply_middleware_chain_with_request_id( req, client, @@ -3008,17 +3018,12 @@ where .endpoint_observation_tx .as_ref() .and_then(EndpointObservationSender::capture); - let parsed = match crate::l7::jsonrpc::parse_jsonrpc_http_request( - client, - config.json_rpc_max_body_bytes, - crate::l7::path::CanonicalizeOptions { + let provider = + crate::l7::rest::RestProvider::with_options(crate::l7::path::CanonicalizeOptions { allow_encoded_slash: config.allow_encoded_slash, ..Default::default() - }, - crate::l7::jsonrpc::JsonRpcInspectionOptions::for_config(config), - ) - .await - { + }); + let mut req = match provider.parse_request(client).await { Ok(Some(parsed)) => parsed, Ok(None) => return Ok(()), Err(e) => { @@ -3038,14 +3043,32 @@ where } }; - let req = parsed.request; - let jsonrpc_info = parsed.info; let correlation = RequestCorrelation::new(&req, ctx); let mut audited_client = correlation.observe_response(client); let client = &mut audited_client; correlation .record_policy_runtime_generation(engine.generation_guard().captured_generation()); correlation.record_selected_policy_endpoint(&config.policy_hash, &config.endpoint_id); + let jsonrpc_info = match correlation + .within_mcp(crate::l7::jsonrpc::inspect_jsonrpc_http_request( + client, + &mut req, + config.json_rpc_max_body_bytes, + crate::l7::jsonrpc::JsonRpcInspectionOptions::for_config(config), + )) + .await + { + Ok(info) => info, + Err(error) => { + let detail = + parse_rejection_detail(&error.to_string(), ParseRejectionMode::L7Endpoint); + emit_parse_rejection(ctx, &detail, jsonrpc_engine_type(config.protocol)); + correlation.record_enforcement_action("block"); + correlation.record_final_http_outcome("denied"); + correlation.record_error_stage("request_inspection"); + return Ok(()); + } + }; let observer = EndpointObserver::begin_captured( ctx.endpoint_observation_tx.as_ref(), config, @@ -3183,7 +3206,8 @@ where // against the same body-aware policy after every transforming // stage so a middleware cannot smuggle a denied operation to the // upstream or the next stage. - let validate = transformed_body_validator(config, engine, ctx, &request_info); + let validate = + transformed_body_validator(config, engine, ctx, &request_info, Some(&correlation)); let req = match apply_middleware_chain_with_request_id( req, client, @@ -3515,7 +3539,8 @@ where // against the same body-aware policy after every transforming // stage so a middleware cannot smuggle a denied operation to the // upstream or the next stage. - let validate = transformed_body_validator(config, engine, ctx, &request_info); + let validate = + transformed_body_validator(config, engine, ctx, &request_info, Some(&correlation)); let req = match apply_middleware_chain_with_request_id( req, client, @@ -3887,6 +3912,7 @@ fn reevaluate_transformed_body( ctx: &L7EvalContext, request_info: &L7RequestInfo, body: &[u8], + correlation: Option<&RequestCorrelation>, ) -> Result> { let (engine_type, transformed_info) = match config.protocol { // REST and websocket-upgrade policy evaluates only the method, path, @@ -3910,6 +3936,9 @@ fn reevaluate_transformed_body( } let info = crate::l7::jsonrpc::parse_jsonrpc_body_with_options(body, inspection_options); + if let Some(correlation) = correlation { + correlation.record_effective_jsonrpc_summary(config.protocol, &info); + } let mut transformed_info = request_info.clone(); transformed_info.jsonrpc = Some(info); (jsonrpc_engine_type(config.protocol), transformed_info) @@ -3934,7 +3963,14 @@ fn reevaluate_transformed_body( if let Some(reason) = l7_request_hard_deny_reason(config.protocol, &transformed_info) { let reason = format!("middleware transformation rejected: {reason}"); - emit_transformed_body_decision(ctx, request_info, engine_type, "deny", &reason); + emit_transformed_body_decision( + ctx, + request_info, + engine_type, + "deny", + &reason, + correlation, + ); return Ok(Some(reason)); } @@ -3944,10 +3980,17 @@ fn reevaluate_transformed_body( } let reason = format!("middleware transformation denied by policy: {reason}"); if config.enforcement == EnforcementMode::Audit { - emit_transformed_body_decision(ctx, request_info, engine_type, "audit", &reason); + emit_transformed_body_decision( + ctx, + request_info, + engine_type, + "audit", + &reason, + correlation, + ); return Ok(None); } - emit_transformed_body_decision(ctx, request_info, engine_type, "deny", &reason); + emit_transformed_body_decision(ctx, request_info, engine_type, "deny", &reason, correlation); Ok(Some(reason)) } @@ -3959,8 +4002,11 @@ pub(crate) fn transformed_body_validator<'a>( engine: &'a TunnelPolicyEngine, ctx: &'a L7EvalContext, request_info: &'a L7RequestInfo, + correlation: Option<&'a RequestCorrelation>, ) -> impl Fn(&[u8]) -> Result> + Send + Sync + 'a { - move |body: &[u8]| reevaluate_transformed_body(config, engine, ctx, request_info, body) + move |body: &[u8]| { + reevaluate_transformed_body(config, engine, ctx, request_info, body, correlation) + } } /// Log the post-transformation policy decision as an OCSF HTTP Activity @@ -3972,6 +4018,7 @@ fn emit_transformed_body_decision( engine_type: &str, decision_str: &str, reason: &str, + correlation: Option<&RequestCorrelation>, ) { let (action_id, disposition_id, severity) = match decision_str { "deny" => (ActionId::Denied, DispositionId::Blocked, SeverityId::Medium), @@ -3997,7 +4044,11 @@ fn emit_transformed_body_decision( request_info.action, ctx.host, ctx.port, request_info.target, reason )) .build(); - egress_emit!(event); + if let Some(correlation) = correlation { + correlation.emit(event); + } else { + egress_emit!(event); + } } fn jsonrpc_policy_input(info: &crate::l7::jsonrpc::JsonRpcRequestInfo) -> serde_json::Value { @@ -4535,7 +4586,7 @@ mod tests { use openshell_core::token_grant_failure::{ TokenGrantFailure, TokenGrantFailureKind, TokenGrantStage, }; - use std::collections::{BTreeMap, HashMap as TestHashMap}; + use std::collections::{BTreeMap, BTreeSet, HashMap as TestHashMap}; use std::fmt::Write as _; use std::path::PathBuf; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; @@ -8783,6 +8834,113 @@ network_policies: assert!(upstream_seen.is_none(), "upstream must not see the request"); } + #[test] + fn transformed_jsonrpc_decision_reuses_correlation_without_body_data() { + const SENTINEL: &str = "SENTINEL_PRIVATE_TRANSFORMED_BODY"; + const REQUEST_ID: &str = "00000000-0000-4000-8000-000000000011"; + + let (config, engine, ctx) = jsonrpc_transforming_relay_parts("enforce", b"not used"); + let original_body = br#"{"jsonrpc":"2.0","id":1,"method":"reports.list"}"#; + let original_info = crate::l7::jsonrpc::parse_jsonrpc_body( + original_body, + crate::l7::jsonrpc::JsonRpcInspectionMode::JsonRpc, + ); + let request_info = L7RequestInfo { + action: "POST".into(), + target: "/rpc".into(), + query_params: std::collections::HashMap::new(), + graphql: None, + jsonrpc: Some(original_info.clone()), + }; + let request = crate::l7::provider::L7Request { + action: "POST".into(), + target: "/rpc".into(), + query_params: std::collections::HashMap::new(), + raw_header: format!( + "POST /rpc HTTP/1.1\r\nHost: api.example.test\r\nContent-Length: {}\r\n\r\n{}", + original_body.len(), + std::str::from_utf8(original_body).unwrap() + ) + .into_bytes(), + body_length: crate::l7::provider::BodyLength::ContentLength(original_body.len() as u64), + }; + 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, "correlation-test")) + .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()); + correlation.record_effective_jsonrpc_summary(config.protocol, &original_info); + let validate = transformed_body_validator( + &config, + &engine, + &ctx, + &request_info, + Some(&correlation), + ); + let rejection = validate(format!("{SENTINEL} is not JSON").as_bytes()) + .expect("transformed-body evaluation"); + assert!(rejection.is_some()); + correlation.record_final_http_outcome("denied"); + }); + provider.force_flush().expect("flush spans"); + + let span = exporter + .get_finished_spans() + .expect("finished spans") + .into_iter() + .rev() + .find(|span| span.name == "openshell.mcp.inspect") + .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.l7.protocol"], "jsonrpc"); + assert_eq!( + attributes["openshell.jsonrpc.effective.inspection_outcome"], + "invalid_json" + ); + assert_eq!( + attributes["openshell.jsonrpc.effective.message_kind"], + "invalid" + ); + assert!(!format!("{attributes:?}").contains(SENTINEL)); + + let transformed_events = ocsf + .json_events() + .into_iter() + .filter(|event| { + event["message"] + .as_str() + .is_some_and(|message| message.starts_with("L7_REQUEST_TRANSFORMED")) + }) + .collect::>(); + assert_eq!(transformed_events.len(), 1); + assert_eq!( + transformed_events[0]["metadata"]["correlation_uid"], + REQUEST_ID + ); + assert!( + !serde_json::to_string(&transformed_events) + .unwrap() + .contains(SENTINEL) + ); + } + #[tokio::test] async fn transformed_graphql_body_is_reevaluated_and_denied() { // GraphQL counterpart: policy allows query { viewer }; the middleware @@ -13220,6 +13378,188 @@ network_policies: serde_json::json!({"jsonrpc":"2.0", "id":1, "method":method, "params":params}).to_string() } + async fn read_test_http_message(reader: &mut R) -> Vec { + let mut message = Vec::new(); + let mut byte = [0_u8; 1]; + while !message.ends_with(b"\r\n\r\n") { + reader + .read_exact(&mut byte) + .await + .expect("complete HTTP headers"); + message.push(byte[0]); + } + let headers = std::str::from_utf8(&message).expect("ASCII test headers"); + let body_len = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length").then(|| { + value + .trim() + .parse::() + .expect("numeric content length") + }) + }) + .unwrap_or_default(); + let header_len = message.len(); + message.resize(header_len + body_len, 0); + reader + .read_exact(&mut message[header_len..]) + .await + .expect("complete HTTP body"); + message + } + + #[tokio::test] + async fn production_mcp_relays_export_one_privacy_safe_effective_summary() { + const SENTINEL: &str = "SENTINEL_PRIVATE_PRODUCTION_TOOL"; + + for route_selected in [false, true] { + 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, + "mcp-production-correlation-test", + )); + let (config, tunnel_engine, ctx) = mcp_sessionless_test_relay_context(); + let body = sessionless_mcp_body( + "tools/call", + serde_json::json!({"name": SENTINEL, "arguments": {"secret": SENTINEL}}), + ); + let request = format!( + "POST /mcp HTTP/1.1\r\nHost: mcp.example.test:8000\r\nContent-Type: application/json\r\nAccept: application/json, text/event-stream\r\nMCP-Protocol-Version: 2026-07-28\r\nMcp-Method: tools/call\r\nMcp-Name: {SENTINEL}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let (mut app, mut relay_client) = tokio::io::duplex(32 * 1024); + let (mut relay_upstream, mut upstream) = tokio::io::duplex(32 * 1024); + + let exchange = async { + tracing::callsite::rebuild_interest_cache(); + tokio::join!( + async { + if route_selected { + relay_with_route_selection( + &[config], + tunnel_engine, + &mut relay_client, + &mut relay_upstream, + &ctx, + ) + .await + } else { + relay_with_inspection( + &config, + tunnel_engine, + &mut relay_client, + &mut relay_upstream, + &ctx, + ) + .await + } + }, + async { + app.write_all(request.as_bytes()) + .await + .expect("send MCP request"); + app.flush().await.expect("flush MCP request"); + read_test_http_message(&mut app).await + }, + async { + let forwarded = read_test_http_message(&mut upstream).await; + upstream + .write_all( + b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ) + .await + .expect("send upstream response"); + upstream.flush().await.expect("flush upstream response"); + forwarded + }, + ) + } + .with_subscriber(subscriber); + let (relay, response, forwarded) = + tokio::time::timeout(std::time::Duration::from_secs(5), Box::pin(exchange)) + .await + .expect("MCP correlation exchange completes"); + relay.expect("MCP relay succeeds"); + assert!(response.starts_with(b"HTTP/1.1 204 No Content")); + assert!(forwarded.starts_with(b"POST /mcp HTTP/1.1")); + provider.force_flush().expect("flush spans"); + + let span = exporter + .get_finished_spans() + .expect("finished spans") + .into_iter() + .rev() + .find(|span| span.name == "openshell.mcp.inspect") + .expect("HTTP enforcement span"); + assert_eq!(span.dropped_attributes_count, 0); + let raw_summary_keys = span + .attributes + .iter() + .map(|attribute| attribute.key.as_str()) + .filter(|key| { + *key == "openshell.l7.protocol" + || key.starts_with("openshell.jsonrpc.") + || key.starts_with("openshell.mcp.") + }) + .collect::>(); + assert_eq!( + raw_summary_keys.len(), + raw_summary_keys + .iter() + .copied() + .collect::>() + .len(), + "route_selected={route_selected}: unique correlation summary keys" + ); + let call_count = span + .attributes + .iter() + .find(|attribute| { + attribute.key.as_str() == "openshell.jsonrpc.effective.call_count" + }) + .expect("call count attribute"); + assert_eq!(format!("{:?}", call_count.value), "I64(1)"); + let attributes = span + .attributes + .into_iter() + .map(|attribute| { + ( + attribute.key.as_str().to_string(), + attribute.value.to_string(), + ) + }) + .collect::>(); + assert_eq!(attributes["openshell.l7.protocol"], "mcp"); + assert_eq!( + attributes["openshell.jsonrpc.effective.message_kind"], + "call" + ); + assert_eq!( + attributes["openshell.jsonrpc.effective.inspection_outcome"], + "accepted" + ); + assert_eq!( + attributes["openshell.mcp.selected_protocol_revision"], + "2026-07-28" + ); + assert_eq!( + attributes["openshell.mcp.effective.protocol_revision"], + "2026-07-28" + ); + assert_eq!( + attributes["openshell.mcp.effective.method_classification"], + "core" + ); + assert_eq!(attributes["openshell.mcp.tools_call_observed"], "true"); + assert!(!format!("{attributes:?}").contains(SENTINEL)); + } + } + async fn run_sessionless_mcp_relay( route_selected: bool, headers: &str, diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index eb5a93d7f8..54e98c801c 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -1943,6 +1943,7 @@ impl ForwardMiddlewarePipeline<'_> { l7.engine, self.ctx, l7.request_info, + self.correlation, ); openshell_supervisor_middleware::TransformedBodyPolicy::Reevaluate(&validate) } @@ -11631,7 +11632,11 @@ network_policies: #[tokio::test] async fn forward_middleware_pipeline_denies_policy_invalid_transformation() { + use tracing_subscriber::layer::SubscriberExt as _; + const TEST_POLICY: &str = include_str!("../data/sandbox-policy.rego"); + const REQUEST_ID: &str = "00000000-0000-4000-8000-000000000031"; + const SENSITIVE_METHOD: &str = "sk-ABCDEFGHIJKLMNOP"; let data = r#" network_policies: jsonrpc_api: @@ -11712,6 +11717,24 @@ network_policies: runner, tunnel_engine.generation_guard().clone(), ); + 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, + "forward-pipeline-correlation-test", + )) + .with(ocsf.clone()); + let _subscriber = tracing::subscriber::set_default(subscriber); + tracing::callsite::rebuild_interest_cache(); + let correlation = crate::l7::correlation::RequestCorrelation::with_request_id( + &request, + &ctx, + REQUEST_ID.into(), + ); let pipeline = ForwardMiddlewarePipeline { ctx: &ctx, scheme: "http", @@ -11721,7 +11744,7 @@ network_policies: engine: &tunnel_engine, request_info: &request_info, }), - correlation: None, + correlation: Some(&correlation), }; let (_app, mut client) = tokio::io::duplex(8192); @@ -11752,6 +11775,46 @@ network_policies: panic!("test middleware work admission must be available") } } + + correlation.record_final_http_outcome("denied"); + drop(correlation); + provider.force_flush().expect("flush spans"); + + let attributes = crate::l7::correlation::test_request_attributes( + &exporter.get_finished_spans().expect("finished spans"), + Some(REQUEST_ID), + ); + assert_eq!(attributes["openshell.l7.protocol"], "jsonrpc"); + assert_eq!( + attributes["openshell.jsonrpc.effective.inspection_outcome"], + "accepted" + ); + assert_eq!( + attributes["openshell.jsonrpc.effective.message_kind"], + "call" + ); + assert_eq!(attributes["openshell.jsonrpc.effective.call_count"], "1"); + assert!(!format!("{attributes:?}").contains(SENSITIVE_METHOD)); + + let transformed_events = ocsf + .json_events() + .into_iter() + .filter(|event| { + event["message"] + .as_str() + .is_some_and(|message| message.starts_with("L7_REQUEST_TRANSFORMED")) + }) + .collect::>(); + assert_eq!(transformed_events.len(), 1); + assert_eq!( + transformed_events[0]["metadata"]["correlation_uid"], + REQUEST_ID + ); + assert!( + !serde_json::to_string(&transformed_events) + .unwrap() + .contains(SENSITIVE_METHOD) + ); } #[tokio::test] diff --git a/docs/observability/traces.mdx b/docs/observability/traces.mdx index 8dd7cf7fad..d153a01d25 100644 --- a/docs/observability/traces.mdx +++ b/docs/observability/traces.mdx @@ -76,7 +76,9 @@ The gateway also passes its trace context to compute drivers, so driver spans ap ### 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, +and selected request and response middleware phases appear as children. +MCP and JSON-RPC inspection adds an `openshell.mcp.inspect` child with bounded +counts and outcomes, not arguments or payloads. 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`. @@ -93,8 +95,7 @@ OCSF record includes notable withholding, denial, unavailability or stale state. 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 +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.