From 10d08e8d8215ba3ffec56f55b339ab71163a1c7d Mon Sep 17 00:00:00 2001 From: Kris Hicks Date: Fri, 28 Aug 2026 13:02:49 -0700 Subject: [PATCH 1/3] fix(server): restore accurate sandbox logs `openshell logs` now includes sandbox-scoped gateway OCSF events, including policy and service-routing activity. Service request failures that occur before endpoint lookup completes are attributed to the affected sandbox instead of disappearing from its logs. Gateway-wide events no longer create a phantom empty-sandbox log, and a deleted sandbox's log stream stays closed so late activity cannot recreate stale log state. Refs #1055 Signed-off-by: Kris Hicks --- crates/openshell-server/src/grpc/policy.rs | 39 ++- .../openshell-server/src/service_routing.rs | 202 +++++++++++++-- crates/openshell-server/src/tls.rs | 16 +- crates/openshell-server/src/tracing_bus.rs | 241 ++++++++++++++++-- 4 files changed, 414 insertions(+), 84 deletions(-) diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 0e4c74af95..e46b2ae8ae 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -56,7 +56,7 @@ use openshell_core::{ settings::{self, SettingValueKind}, }; use openshell_ocsf::{ - ConfigStateChangeBuilder, OCSF_TARGET, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, + ConfigStateChangeBuilder, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, }; use openshell_policy::{ PolicyMergeOp, ProviderPolicyLayer, canonicalize_advisor_add_rule, compose_effective_policy, @@ -162,7 +162,7 @@ fn emit_gateway_policy_audit_log( version: i64, policy_hash: &str, ) { - let message = build_gateway_policy_audit_message( + let event = build_gateway_policy_audit_event( sandbox_id, sandbox_name, state_label, @@ -171,11 +171,7 @@ fn emit_gateway_policy_audit_log( policy_hash, &[], ); - info!( - target: OCSF_TARGET, - sandbox_id = %sandbox_id, - message = %message - ); + openshell_ocsf::ocsf_emit!(event); } /// Emit a `CONFIG:APPROVED` audit event for an auto-approval — same event @@ -200,7 +196,7 @@ fn emit_gateway_policy_auto_approve_audit_log( ("prover_delta", "empty".to_string()), ("resolved_from", resolved_from.to_string()), ]; - let message = build_gateway_policy_audit_message( + let event = build_gateway_policy_audit_event( sandbox_id, sandbox_name, "approved", @@ -209,14 +205,10 @@ fn emit_gateway_policy_auto_approve_audit_log( policy_hash, &extra, ); - info!( - target: OCSF_TARGET, - sandbox_id = %sandbox_id, - message = %message - ); + openshell_ocsf::ocsf_emit!(event); } -fn build_gateway_policy_audit_message( +fn build_gateway_policy_audit_event( sandbox_id: &str, sandbox_name: &str, state_label: &str, @@ -224,7 +216,7 @@ fn build_gateway_policy_audit_message( version: i64, policy_hash: &str, extra_fields: &[(&str, String)], -) -> String { +) -> OcsfEvent { let ctx = SandboxContext { sandbox_id: sandbox_id.to_string(), sandbox_name: sandbox_name.to_string(), @@ -248,8 +240,7 @@ fn build_gateway_policy_audit_message( for (key, value) in extra_fields { builder = builder.unmapped(key, value.clone()); } - let event: OcsfEvent = builder.build(); - event.format_shorthand() + builder.build() } fn summarize_cli_policy_merge_op(operation: &PolicyMergeOp) -> String { @@ -15486,8 +15477,8 @@ mod tests { } #[test] - fn build_gateway_policy_audit_message_formats_ocsf_config_line() { - let message = build_gateway_policy_audit_message( + fn build_gateway_policy_audit_event_formats_ocsf_config_line() { + let message = build_gateway_policy_audit_event( "sb-123", "demo-sandbox", "merged", @@ -15495,7 +15486,8 @@ mod tests { 7, "sha256:testhash", &[], - ); + ) + .format_shorthand(); assert_eq!( message, @@ -15510,13 +15502,13 @@ mod tests { /// findings" — never "safe" — because the claim is about the prover's /// reasoning, not the world. #[test] - fn build_gateway_policy_audit_message_carries_auto_approve_provenance() { + fn build_gateway_policy_audit_event_carries_auto_approve_provenance() { let extra = [ ("auto", "true".to_string()), ("source", "agent_authored".to_string()), ("prover_delta", "empty".to_string()), ]; - let message = build_gateway_policy_audit_message( + let message = build_gateway_policy_audit_event( "sb-123", "demo-sandbox", "approved", @@ -15524,7 +15516,8 @@ mod tests { 12, "sha256:autohash", &extra, - ); + ) + .format_shorthand(); assert!( message.contains("CONFIG:APPROVED"), "auto-approval reuses CONFIG:APPROVED; got: {message}" diff --git a/crates/openshell-server/src/service_routing.rs b/crates/openshell-server/src/service_routing.rs index 3e80bc26f5..1ca575c059 100644 --- a/crates/openshell-server/src/service_routing.rs +++ b/crates/openshell-server/src/service_routing.rs @@ -14,17 +14,18 @@ use openshell_core::proto::{Sandbox, SandboxPhase, ServiceEndpoint, TcpRelayTarg use openshell_core::{ObjectId, VERSION}; use openshell_ocsf::{ ActionId, ActivityId, ConfigStateChangeBuilder, DispositionId, Endpoint, HttpActivityBuilder, - HttpRequest, HttpResponse as OcsfHttpResponse, NetworkActivityBuilder, OCSF_TARGET, OcsfEvent, + HttpRequest, HttpResponse as OcsfHttpResponse, NetworkActivityBuilder, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, Url as OcsfUrl, }; use std::net::{IpAddr, Ipv4Addr}; use std::sync::Arc; use std::time::Duration; use tokio::io::AsyncWriteExt; -use tracing::{info, warn}; +use tracing::warn; use crate::ServerState; use crate::persistence::{ObjectType, Store}; +use crate::sandbox_index::SandboxIndex; const ENDPOINT_OBJECT_TYPE: &str = "service_endpoint"; const ROUTING_RULE_NAME: &str = "sandbox_service_routing"; @@ -236,7 +237,15 @@ async fn proxy_to_endpoint( { Ok(endpoint) => endpoint, Err(err) => { - emit_service_http_failure(&state, &req, &sandbox_name, &service_name, None, &err); + emit_service_http_failure( + &state, + &req, + workspace, + &sandbox_name, + &service_name, + None, + &err, + ); return Err(err); } }; @@ -245,6 +254,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -264,6 +274,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -277,6 +288,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -290,6 +302,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -302,6 +315,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -314,6 +328,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -582,19 +597,19 @@ fn is_gateway_auth_cookie(name: &str) -> bool { pub fn emit_service_endpoint_config_event(endpoint: &ServiceEndpoint, url: &str, created: bool) { let event = build_service_endpoint_config_event(endpoint, url, created); - emit_gateway_ocsf_event(&endpoint.sandbox_id, event); + emit_gateway_ocsf_event(event); } pub fn emit_service_endpoint_delete_event(endpoint: &ServiceEndpoint) { let event = build_service_endpoint_delete_event(endpoint); - emit_gateway_ocsf_event(&endpoint.sandbox_id, event); + emit_gateway_ocsf_event(event); } pub fn emit_cross_origin_service_http_rejection(state: &ServerState, req: &Request) { let Some(host) = request_host(req) else { return; }; - let Some((_workspace, sandbox_name, service_name)) = + let Some((workspace, sandbox_name, service_name)) = parse_host(host, &state.config.service_routing) else { return; @@ -604,32 +619,42 @@ pub fn emit_cross_origin_service_http_rejection(state: &ServerState, req: &Reque "Cross-origin service request rejected", "cross-origin service request rejected", ); - emit_service_http_failure(state, req, &sandbox_name, &service_name, None, &err); + emit_service_http_failure( + state, + req, + &workspace, + &sandbox_name, + &service_name, + None, + &err, + ); } fn emit_service_http_failure( state: &ServerState, req: &Request, + workspace: &str, sandbox_name: &str, service_name: &str, endpoint: Option<&ServiceEndpoint>, err: &ServiceRouteError, ) { + let sandbox_id = + http_failure_sandbox_id(&state.sandbox_index, workspace, sandbox_name, endpoint); let event = build_service_http_failure_event( state.config.bind_address.port(), req, + &sandbox_id, sandbox_name, service_name, - endpoint, err, ); - let sandbox_id = endpoint.map_or("", |endpoint| endpoint.sandbox_id.as_str()); - emit_gateway_ocsf_event(sandbox_id, event); + emit_gateway_ocsf_event(event); } fn emit_service_relay_failure(endpoint: &ServiceEndpoint, target_port: u16, reason: &str) { let event = build_service_relay_failure_event(endpoint, target_port, reason); - emit_gateway_ocsf_event(&endpoint.sandbox_id, event); + emit_gateway_ocsf_event(event); } fn build_service_endpoint_config_event( @@ -679,20 +704,34 @@ fn build_service_endpoint_delete_event(endpoint: &ServiceEndpoint) -> OcsfEvent .build() } +/// Resolve an endpoint's sandbox id, falling back to the workspace/name index. +fn http_failure_sandbox_id( + index: &SandboxIndex, + workspace: &str, + sandbox_name: &str, + endpoint: Option<&ServiceEndpoint>, +) -> String { + endpoint.map_or_else( + || { + index + .sandbox_id_for_sandbox_name(workspace, sandbox_name) + .unwrap_or_default() + }, + |endpoint| endpoint.sandbox_id.clone(), + ) +} + fn build_service_http_failure_event( bind_port: u16, req: &Request, + sandbox_id: &str, sandbox_name: &str, service_name: &str, - endpoint: Option<&ServiceEndpoint>, err: &ServiceRouteError, ) -> OcsfEvent { let host = request_host(req).unwrap_or("unknown"); let (hostname, port) = split_authority_for_event(host, bind_port); - let ctx = gateway_ocsf_ctx( - endpoint.map_or("", |endpoint| endpoint.sandbox_id.as_str()), - sandbox_name, - ); + let ctx = gateway_ocsf_ctx(sandbox_id, sandbox_name); HttpActivityBuilder::new(&ctx) .activity(http_activity_for_method(req.method())) .action(ActionId::Denied) @@ -751,13 +790,9 @@ fn build_service_relay_failure_event( .build() } -fn emit_gateway_ocsf_event(sandbox_id: &str, event: OcsfEvent) { - let message = event.format_shorthand(); - info!( - target: OCSF_TARGET, - sandbox_id = %sandbox_id, - message = %message - ); +/// Emit through the structured OCSF tracing bridge. +fn emit_gateway_ocsf_event(event: OcsfEvent) { + openshell_ocsf::ocsf_emit!(event); } fn gateway_ocsf_ctx(sandbox_id: &str, sandbox_name: &str) -> SandboxContext { @@ -1120,8 +1155,14 @@ mod tests { "Cross-origin service request rejected", "cross-origin service request rejected", ); - let event = - build_service_http_failure_event(18080, &request, "my-sandbox", "web", None, &err); + let event = build_service_http_failure_event( + 18080, + &request, + "sandbox-1", + "my-sandbox", + "web", + &err, + ); let json = event.to_json().unwrap(); assert_eq!(json["class_uid"], 4002); @@ -1234,4 +1275,115 @@ mod tests { "should not find endpoint in wrong workspace" ); } + + fn indexed_sandbox() -> SandboxIndex { + let index = SandboxIndex::new(); + index.update_from_sandbox(&Sandbox { + metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { + id: "sandbox-1".to_string(), + name: "my-sandbox".to_string(), + workspace: "default".to_string(), + ..Default::default() + }), + ..Default::default() + }); + index + } + + #[test] + fn http_failure_resolves_the_sandbox_id_from_the_index_without_an_endpoint() { + let index = indexed_sandbox(); + + assert_eq!( + http_failure_sandbox_id(&index, "default", "my-sandbox", None), + "sandbox-1" + ); + } + + #[test] + fn http_failure_sandbox_id_prefers_the_endpoint_and_tolerates_unknown_names() { + let index = indexed_sandbox(); + let endpoint = endpoint(); + + assert_eq!( + http_failure_sandbox_id(&index, "default", "my-sandbox", Some(&endpoint)), + endpoint.sandbox_id + ); + assert_eq!( + http_failure_sandbox_id(&index, "default", "does-not-exist", None), + "" + ); + assert_eq!( + http_failure_sandbox_id(&index, "other-workspace", "my-sandbox", None), + "" + ); + } + + #[test] + fn http_failure_event_carries_the_resolved_sandbox_id() { + let req = Request::builder() + .method(Method::GET) + .uri("/") + .header(header::HOST, "default--my-sandbox--web.example.test") + .body(Body::empty()) + .unwrap(); + let event = build_service_http_failure_event( + 8443, + &req, + "sandbox-1", + "my-sandbox", + "web", + &ServiceRouteError::endpoint_not_found(), + ); + + assert_eq!( + event.base().metadata.uid.as_deref(), + Some("sandbox-1"), + "resolved sandbox id should reach the event" + ); + } + + /// Captures structured OCSF events during tracing dispatch. + #[derive(Clone, Default)] + struct ProbeLayer { + seen: Arc>>>, + } + + impl tracing_subscriber::Layer for ProbeLayer { + fn on_event( + &self, + event: &tracing::Event<'_>, + _ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + if event.metadata().target() == openshell_ocsf::OCSF_TARGET { + self.seen + .lock() + .unwrap() + .push(openshell_ocsf::clone_current_event()); + } + } + } + + #[test] + fn gateway_ocsf_events_expose_the_structured_event_to_layers() { + use tracing_subscriber::layer::SubscriberExt; + + let probe = ProbeLayer::default(); + let subscriber = tracing_subscriber::registry().with(probe.clone()); + + let endpoint = endpoint(); + let expected = build_service_endpoint_config_event(&endpoint, "https://example.test", true); + let expected_shorthand = expected.format_shorthand(); + + tracing::subscriber::with_default(subscriber, || { + emit_service_endpoint_config_event(&endpoint, "https://example.test", true); + }); + + let seen = probe.seen.lock().unwrap(); + assert_eq!(seen.len(), 1, "expected exactly one OCSF tracing event"); + let event = seen[0] + .as_ref() + .expect("structured OCSF event should be reachable from the layer"); + assert_eq!(event.format_shorthand(), expected_shorthand); + } } diff --git a/crates/openshell-server/src/tls.rs b/crates/openshell-server/src/tls.rs index aa627746ce..e3b48b9fbb 100644 --- a/crates/openshell-server/src/tls.rs +++ b/crates/openshell-server/src/tls.rs @@ -14,9 +14,7 @@ use arc_swap::ArcSwap; use notify::event::EventKind; use notify::{Event, RecursiveMode, Watcher}; use openshell_core::{Error, Result}; -use openshell_ocsf::{ - ConfigStateChangeBuilder, OCSF_TARGET, SandboxContext, SeverityId, StateId, StatusId, -}; +use openshell_ocsf::{ConfigStateChangeBuilder, SandboxContext, SeverityId, StateId, StatusId}; use rustls::ServerConfig; use rustls::crypto::ring::sign; use rustls::pki_types::{CertificateDer, PrivateKeyDer}; @@ -117,11 +115,7 @@ impl TlsAcceptor { .state(StateId::Enabled, "reloaded") .message("TLS certificate config reloaded successfully") .build(); - info!( - target: OCSF_TARGET, - sandbox_id = "", - message = %event.format_shorthand() - ); + openshell_ocsf::ocsf_emit!(event); Ok(()) } @@ -247,11 +241,7 @@ impl TlsAcceptor { "TLS certificate reload failed: {e}" )) .build(); - info!( - target: OCSF_TARGET, - sandbox_id = "", - message = %event.format_shorthand() - ); + openshell_ocsf::ocsf_emit!(event); warn!(error = %e, "TLS certificate reload failed, keeping existing config"); } break; diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index a91a5fd877..15c7908e34 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -3,7 +3,7 @@ //! Capture openshell-server tracing logs for streaming over gRPC. -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{Arc, Mutex}; use openshell_core::proto::{SandboxLogLine, SandboxStreamEvent}; @@ -24,6 +24,9 @@ pub struct TracingLogBus { struct Inner { per_id: HashMap>, tails: HashMap>, + /// Recently removed sandbox ids, in eviction order. + removed: VecDeque, + removed_set: HashSet, } impl Default for TracingLogBus { @@ -39,6 +42,8 @@ impl TracingLogBus { inner: Arc::new(Mutex::new(Inner { per_id: HashMap::new(), tails: HashMap::new(), + removed: VecDeque::new(), + removed_set: HashSet::new(), })), platform_event_bus: PlatformEventBus::new(), } @@ -51,8 +56,13 @@ impl TracingLogBus { } } - fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { + pub fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + if inner.removed_set.contains(sandbox_id) { + let (tx, rx) = broadcast::channel(1); + drop(tx); + return rx; + } inner .per_id .entry(sandbox_id.to_string()) @@ -60,11 +70,7 @@ impl TracingLogBus { let (tx, _rx) = broadcast::channel(1024); tx }) - .clone() - } - - pub fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { - self.sender_for(sandbox_id).subscribe() + .subscribe() } /// Remove all bus entries for the given sandbox id. @@ -75,6 +81,15 @@ impl TracingLogBus { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); inner.per_id.remove(sandbox_id); inner.tails.remove(sandbox_id); + + if inner.removed_set.insert(sandbox_id.to_string()) { + inner.removed.push_back(sandbox_id.to_string()); + while inner.removed.len() > Self::MAX_REMEMBERED_REMOVALS { + if let Some(evicted) = inner.removed.pop_front() { + inner.removed_set.remove(&evicted); + } + } + } } pub fn tail(&self, sandbox_id: &str, max: usize) -> Vec { @@ -95,6 +110,9 @@ impl TracingLogBus { /// used by the tracing layer, so it appears in `WatchSandbox` and /// `GetSandboxLogs` transparently. pub fn publish_external(&self, log: SandboxLogLine) { + if log.sandbox_id.is_empty() { + return; + } let evt = SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( log.clone(), @@ -106,16 +124,42 @@ impl TracingLogBus { /// Default tail buffer capacity (lines per sandbox). const DEFAULT_TAIL: usize = 2000; - fn publish(&self, sandbox_id: &str, event: SandboxStreamEvent, tail_cap: usize) { - let tx = self.sender_for(sandbox_id); - let _ = tx.send(event.clone()); + /// Number of `(sender, tail)` entries currently held, for leak assertions. + #[cfg(test)] + fn entry_counts(&self) -> (usize, usize) { + let inner = self.inner.lock().expect("tracing bus lock poisoned"); + (inner.per_id.len(), inner.tails.len()) + } + + /// Maximum number of removed sandbox ids to retain. + /// + /// This bounds memory; after eviction, a very late publisher may create a + /// fresh entry for that id. + const MAX_REMEMBERED_REMOVALS: usize = 1024; + fn publish(&self, sandbox_id: &str, event: SandboxStreamEvent, tail_cap: usize) { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + if inner.removed_set.contains(sandbox_id) { + return; + } + + let tx = inner + .per_id + .entry(sandbox_id.to_string()) + .or_insert_with(|| { + let (tx, _rx) = broadcast::channel(1024); + tx + }) + .clone(); + let deque = inner.tails.entry(sandbox_id.to_string()).or_default(); - deque.push_back(event); + deque.push_back(event.clone()); while deque.len() > tail_cap { deque.pop_front(); } + drop(inner); + + let _ = tx.send(event); } } @@ -131,14 +175,28 @@ where { fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { let meta = event.metadata(); - let mut visitor = LogVisitor::default(); - event.record(&mut visitor); + // OCSF tracing events carry no fields; the payload arrives out of band + // through a thread-local. + let (visitor_sandbox_id, visitor_message) = if meta.target() == OCSF_TARGET { + openshell_ocsf::clone_current_event().map_or((None, None), |ocsf_event| { + ( + ocsf_event.base().metadata.uid.clone(), + Some(ocsf_event.format_shorthand()), + ) + }) + } else { + let mut visitor = LogVisitor::default(); + event.record(&mut visitor); + (visitor.sandbox_id, visitor.message) + }; - let Some(sandbox_id) = visitor.sandbox_id else { + // An empty id means no sandbox association; publishing would allocate a + // bucket nothing can subscribe to. + let Some(sandbox_id) = visitor_sandbox_id.filter(|id| !id.is_empty()) else { return; }; - let msg = visitor.message.unwrap_or_else(|| meta.name().to_string()); + let msg = visitor_message.unwrap_or_else(|| meta.name().to_string()); let level = display_level(meta.target(), &meta.level().to_string()); let ts = openshell_core::time::now_ms(); @@ -228,22 +286,22 @@ mod tests { } #[test] - fn tracing_log_bus_subscribe_after_remove_creates_fresh_channel() { + fn subscribe_after_remove_does_not_reactivate_the_bus() { let bus = TracingLogBus::new(); let sandbox_id = "sb-2"; - // Create and remove bus.publish_external(make_log_event(sandbox_id, "old message")); bus.remove(sandbox_id); - // Subscribe again — should get a fresh channel with no history let mut rx = bus.subscribe(sandbox_id); - assert!(bus.tail(sandbox_id, 10).is_empty()); + bus.publish_external(make_log_event(sandbox_id, "late message")); - // New publish should reach the new subscriber - bus.publish_external(make_log_event(sandbox_id, "new message")); - let evt = rx.try_recv().expect("should receive new event"); - assert!(evt.payload.is_some()); + assert_eq!(bus.entry_counts(), (0, 0)); + assert!(bus.tail(sandbox_id, 10).is_empty()); + assert!(matches!( + rx.try_recv(), + Err(broadcast::error::TryRecvError::Closed) + )); } #[test] @@ -270,6 +328,143 @@ mod tests { bus.remove("nonexistent"); } + #[test] + fn publish_after_remove_does_not_resurrect_the_bus_entry() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-torn-down"; + + bus.publish_external(make_log_event(sandbox_id, "before teardown")); + assert_eq!(bus.entry_counts(), (1, 1)); + + bus.remove(sandbox_id); + assert_eq!(bus.entry_counts(), (0, 0)); + + bus.publish_external(make_log_event(sandbox_id, "late line")); + assert_eq!(bus.entry_counts(), (0, 0)); + assert!(bus.tail(sandbox_id, 10).is_empty()); + } + + fn ocsf_ctx(sandbox_id: &str) -> openshell_ocsf::SandboxContext { + openshell_ocsf::SandboxContext { + sandbox_id: sandbox_id.to_string(), + sandbox_name: "gw".to_string(), + container_image: "openshell/gateway".to_string(), + hostname: "openshell-gateway".to_string(), + product_version: "0.0.0".to_string(), + proxy_ip: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), + proxy_port: 0, + } + } + + /// Run `f` with the bus layer installed as the active subscriber. + fn with_bus_layer(bus: &TracingLogBus, f: impl FnOnce()) { + use tracing_subscriber::layer::SubscriberExt; + let subscriber = tracing_subscriber::registry().with(bus.layer()); + tracing::subscriber::with_default(subscriber, f); + } + + fn log_message(event: &SandboxStreamEvent) -> &SandboxLogLine { + match event.payload { + Some(openshell_core::proto::sandbox_stream_event::Payload::Log(ref log)) => log, + _ => panic!("expected a log payload"), + } + } + + #[test] + fn ocsf_emit_events_reach_the_bus_with_shorthand_and_sandbox_id() { + use openshell_ocsf::{ + ActionId, ActivityId, DispositionId, Endpoint, NetworkActivityBuilder, SeverityId, + StatusId, ocsf_emit, + }; + + let bus = TracingLogBus::new(); + let event = NetworkActivityBuilder::new(&ocsf_ctx("sb-emit")) + .activity(ActivityId::Open) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443)) + .message("CONNECT denied blocked.example.com:443") + .build(); + let expected_message = event.format_shorthand(); + + with_bus_layer(&bus, || ocsf_emit!(event)); + + let tail = bus.tail("sb-emit", 10); + assert_eq!(tail.len(), 1, "ocsf_emit! event should reach the bus"); + let log = log_message(&tail[0]); + assert_eq!(log.sandbox_id, "sb-emit"); + assert_eq!(log.message, expected_message); + assert_eq!(log.level, "OCSF"); + assert_eq!(log.target, OCSF_TARGET); + assert_eq!(log.source, "gateway"); + } + + #[test] + fn ocsf_emit_events_without_a_sandbox_are_skipped() { + use openshell_ocsf::{ActivityId, AppLifecycleBuilder, SeverityId, ocsf_emit}; + + let bus = TracingLogBus::new(); + let event = AppLifecycleBuilder::new(&ocsf_ctx("")) + .activity(ActivityId::Open) + .severity(SeverityId::Informational) + .message("gateway TLS reloaded") + .build(); + + with_bus_layer(&bus, || ocsf_emit!(event)); + + assert_eq!(bus.entry_counts(), (0, 0)); + } + + #[test] + fn non_ocsf_events_still_use_the_sandbox_id_field() { + let bus = TracingLogBus::new(); + with_bus_layer(&bus, || { + tracing::info!(sandbox_id = "sb-plain", "plain gateway line"); + }); + + let tail = bus.tail("sb-plain", 10); + assert_eq!(tail.len(), 1); + let log = log_message(&tail[0]); + assert_eq!(log.message, "plain gateway line"); + assert_eq!(log.level, "INFO"); + } + + #[test] + fn removal_tombstones_are_bounded() { + let bus = TracingLogBus::new(); + let overflow = TracingLogBus::MAX_REMEMBERED_REMOVALS + 10; + for i in 0..overflow { + bus.remove(&format!("sb-{i}")); + } + + let inner = bus.inner.lock().unwrap(); + assert_eq!(inner.removed.len(), TracingLogBus::MAX_REMEMBERED_REMOVALS); + assert_eq!( + inner.removed_set.len(), + TracingLogBus::MAX_REMEMBERED_REMOVALS + ); + assert!(!inner.removed_set.contains("sb-0")); + assert!(inner.removed_set.contains(&format!("sb-{}", overflow - 1))); + } + + #[test] + fn publish_external_ignores_an_empty_sandbox_id() { + let bus = TracingLogBus::new(); + bus.publish_external(make_log_event("", "no sandbox association")); + + assert_eq!(bus.entry_counts(), (0, 0)); + assert!(bus.tail("", 10).is_empty()); + } + + #[test] + fn publish_external_still_accepts_a_real_sandbox_id() { + let bus = TracingLogBus::new(); + bus.publish_external(make_log_event("sb-real", "hello")); + assert_eq!(bus.tail("sb-real", 10).len(), 1); + } + #[test] fn display_level_maps_ocsf_target_to_ocsf() { assert_eq!(display_level(OCSF_TARGET, "INFO"), "OCSF"); From c493338b7d75598b04a548b8a36dbf6276b03498 Mon Sep 17 00:00:00 2001 From: Kris Hicks Date: Fri, 28 Aug 2026 12:39:40 -0700 Subject: [PATCH 2/3] fix(ocsf): make exported events valid and attributable Exported OCSF records are now schema-valid and can be reliably attributed. Gateway-origin records identify the configured gateway and emitting replica, while sandbox identity remains attached to the affected container and every event receives its own unique ID. API activity records retain their action and disposition when consumed and re-exported. Events also include the required device classification and omit empty container or image objects instead of reporting misleading identities. Refs #1055 Signed-off-by: Kris Hicks --- Cargo.lock | 1 + crates/openshell-ocsf/Cargo.toml | 1 + crates/openshell-ocsf/src/builders/mod.rs | 52 ++++++-- crates/openshell-ocsf/src/ctx.rs | 3 +- .../openshell-ocsf/src/enums/device_type.rs | 57 +++++++++ crates/openshell-ocsf/src/enums/mod.rs | 2 + .../openshell-ocsf/src/events/api_activity.rs | 30 ++++- crates/openshell-ocsf/src/lib.rs | 8 +- crates/openshell-ocsf/src/objects/device.rs | 59 +++++++++ crates/openshell-ocsf/src/objects/metadata.rs | 12 +- crates/openshell-ocsf/tests/event_identity.rs | 115 ++++++++++++++++++ .../openshell-ocsf/tests/gateway_context.rs | 113 +++++++++++++++++ crates/openshell-ocsf/tests/roundtrip.rs | 9 +- crates/openshell-sandbox/src/lib.rs | 1 + crates/openshell-server/src/cli.rs | 9 ++ crates/openshell-server/src/gateway_ocsf.rs | 90 ++++++++++++++ crates/openshell-server/src/grpc/policy.rs | 17 +-- crates/openshell-server/src/lib.rs | 1 + .../openshell-server/src/service_routing.rs | 19 ++- crates/openshell-server/src/tls.rs | 10 +- crates/openshell-server/src/tracing_bus.rs | 14 ++- .../src/l7/middleware.rs | 1 + .../src/l7/relay.rs | 1 + .../src/log_push.rs | 1 + .../src/supervisor_session.rs | 1 + 25 files changed, 566 insertions(+), 61 deletions(-) create mode 100644 crates/openshell-ocsf/src/enums/device_type.rs create mode 100644 crates/openshell-ocsf/tests/event_identity.rs create mode 100644 crates/openshell-ocsf/tests/gateway_context.rs create mode 100644 crates/openshell-server/src/gateway_ocsf.rs diff --git a/Cargo.lock b/Cargo.lock index edc940ea68..b5864fdaba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4173,6 +4173,7 @@ dependencies = [ "serde_repr", "tracing", "tracing-subscriber", + "uuid", ] [[package]] diff --git a/crates/openshell-ocsf/Cargo.toml b/crates/openshell-ocsf/Cargo.toml index be91b1547a..69c7b7e0aa 100644 --- a/crates/openshell-ocsf/Cargo.toml +++ b/crates/openshell-ocsf/Cargo.toml @@ -16,6 +16,7 @@ serde = { workspace = true } serde_json = { workspace = true } serde_repr = "0.1" tracing = { workspace = true } +uuid = { workspace = true } tracing-subscriber = { workspace = true } [dev-dependencies] diff --git a/crates/openshell-ocsf/src/builders/mod.rs b/crates/openshell-ocsf/src/builders/mod.rs index e63b2be88f..4a08652aa5 100644 --- a/crates/openshell-ocsf/src/builders/mod.rs +++ b/crates/openshell-ocsf/src/builders/mod.rs @@ -175,6 +175,19 @@ use crate::enums::StatusId; use crate::events::base_event::BaseEventData; use crate::objects::{Container, Device, Endpoint, Image, Metadata, Product}; +/// Which `OpenShell` component produced an event. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub enum EventOrigin { + /// The sandbox supervisor, running inside a sandbox container. + #[default] + Sandbox, + /// The gateway process, which has no sandbox or container of its own. + Gateway { + /// Operator-assigned gateway name (`[openshell.gateway] name`). + name: String, + }, +} + /// Immutable context created once at sandbox startup. /// /// Passed to every event builder to populate shared OCSF fields @@ -195,37 +208,49 @@ pub struct SandboxContext { pub proxy_ip: IpAddr, /// Proxy listen port. pub proxy_port: u16, + /// Which component is emitting. + pub origin: EventOrigin, } impl SandboxContext { /// Build the OCSF `Metadata` object for any event. #[must_use] pub fn metadata(&self, profiles: &[&str]) -> Metadata { + let product = match self.origin { + EventOrigin::Sandbox => Product::openshell_sandbox(&self.product_version), + EventOrigin::Gateway { .. } => Product::openshell_gateway(&self.product_version), + }; Metadata { version: OCSF_VERSION.to_string(), - product: Product::openshell_sandbox(&self.product_version), + product, profiles: profiles.iter().map(|s| (*s).to_string()).collect(), - uid: Some(self.sandbox_id.clone()), + uid: Some(uuid::Uuid::new_v4().to_string()), log_source: None, } } - /// Build the OCSF `Container` object. + /// Build the OCSF `Container` object when the event concerns a sandbox. #[must_use] - pub fn container(&self) -> Container { - Container { + pub fn container(&self) -> Option { + if self.sandbox_id.is_empty() { + return None; + } + Some(Container { name: self.sandbox_name.clone(), uid: Some(self.sandbox_id.clone()), - image: Some(Image { + image: (!self.container_image.is_empty()).then(|| Image { name: self.container_image.clone(), }), - } + }) } /// Build the OCSF `Device` object. #[must_use] pub fn device(&self) -> Device { - Device::linux(&self.hostname) + match &self.origin { + EventOrigin::Sandbox => Device::linux(&self.hostname), + EventOrigin::Gateway { name } => Device::gateway(&self.hostname, name), + } } /// Build the `proxy_endpoint` object for the Network Proxy profile. @@ -249,7 +274,9 @@ impl SandboxContext { base.set_message(m); } base.set_device(self.device()); - base.set_container(self.container()); + if let Some(container) = self.container() { + base.set_container(container); + } } } @@ -263,6 +290,7 @@ pub(crate) fn test_sandbox_context() -> SandboxContext { product_version: "0.1.0".to_string(), proxy_ip: "10.42.0.1".parse().unwrap(), proxy_port: 3128, + origin: EventOrigin::Sandbox, } } @@ -277,13 +305,15 @@ mod tests { assert_eq!(meta.version, "1.8.0"); assert_eq!(meta.product.name, "OpenShell Sandbox Supervisor"); assert_eq!(meta.profiles.len(), 2); - assert_eq!(meta.uid.as_deref(), Some("sandbox-abc123")); + let uid = meta.uid.as_deref().expect("uid is set"); + assert!(!uid.is_empty()); + assert_ne!(uid, "sandbox-abc123"); } #[test] fn test_sandbox_context_container() { let ctx = test_sandbox_context(); - let container = ctx.container(); + let container = ctx.container().expect("sandbox context has a container"); assert_eq!(container.name, "my-sandbox"); assert_eq!(container.uid.as_deref(), Some("sandbox-abc123")); } diff --git a/crates/openshell-ocsf/src/ctx.rs b/crates/openshell-ocsf/src/ctx.rs index 6916c55215..8aaf71ddbb 100644 --- a/crates/openshell-ocsf/src/ctx.rs +++ b/crates/openshell-ocsf/src/ctx.rs @@ -8,7 +8,7 @@ //! not been set (e.g. unit tests that exercise builders without booting the //! sandbox). -use crate::SandboxContext; +use crate::{EventOrigin, SandboxContext}; use std::sync::{LazyLock, OnceLock}; static OCSF_CTX: OnceLock = OnceLock::new(); @@ -21,6 +21,7 @@ static OCSF_CTX_FALLBACK: LazyLock = LazyLock::new(|| SandboxCon product_version: env!("CARGO_PKG_VERSION").to_string(), proxy_ip: std::net::IpAddr::from([127, 0, 0, 1]), proxy_port: 3128, + origin: EventOrigin::Sandbox, }); /// Initialise the process-wide OCSF sandbox context. diff --git a/crates/openshell-ocsf/src/enums/device_type.rs b/crates/openshell-ocsf/src/enums/device_type.rs new file mode 100644 index 0000000000..887950ec33 --- /dev/null +++ b/crates/openshell-ocsf/src/enums/device_type.rs @@ -0,0 +1,57 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OCSF `device.type_id` enum. + +use serde_repr::{Deserialize_repr, Serialize_repr}; + +/// OCSF Device Type ID. +/// +/// Only the values `OpenShell` can produce are modelled; the schema defines a +/// wider set (desktop, mobile, firewall, router, ...). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize_repr, Deserialize_repr)] +#[repr(u8)] +pub enum DeviceTypeId { + /// 0 — Unknown + Unknown = 0, + /// 1 — Server + Server = 1, + /// 99 — Other + Other = 99, +} + +impl DeviceTypeId { + #[must_use] + pub fn label(self) -> &'static str { + match self { + Self::Unknown => "Unknown", + Self::Server => "Server", + Self::Other => "Other", + } + } + + #[must_use] + pub fn as_u8(self) -> u8 { + self as u8 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn device_type_labels() { + assert_eq!(DeviceTypeId::Unknown.label(), "Unknown"); + assert_eq!(DeviceTypeId::Server.label(), "Server"); + assert_eq!(DeviceTypeId::Other.label(), "Other"); + } + + #[test] + fn device_type_json_roundtrip() { + let json = serde_json::to_value(DeviceTypeId::Server).unwrap(); + assert_eq!(json, serde_json::json!(1)); + let decoded: DeviceTypeId = serde_json::from_value(json).unwrap(); + assert_eq!(decoded, DeviceTypeId::Server); + } +} diff --git a/crates/openshell-ocsf/src/enums/mod.rs b/crates/openshell-ocsf/src/enums/mod.rs index 96567f86d5..b9ae43229d 100644 --- a/crates/openshell-ocsf/src/enums/mod.rs +++ b/crates/openshell-ocsf/src/enums/mod.rs @@ -6,6 +6,7 @@ mod action; mod activity; mod auth; +mod device_type; mod disposition; mod http_method; mod launch; @@ -16,6 +17,7 @@ mod status; pub use action::ActionId; pub use activity::ActivityId; pub use auth::AuthTypeId; +pub use device_type::DeviceTypeId; pub use disposition::DispositionId; pub use http_method::HttpMethod; pub use launch::LaunchTypeId; diff --git a/crates/openshell-ocsf/src/events/api_activity.rs b/crates/openshell-ocsf/src/events/api_activity.rs index 55c2f70d6d..cf7ed7580b 100644 --- a/crates/openshell-ocsf/src/events/api_activity.rs +++ b/crates/openshell-ocsf/src/events/api_activity.rs @@ -40,12 +40,16 @@ pub struct ApiActivityEvent { #[serde(default)] pub dst_endpoint: Option, - /// Action taken (allowed, denied, etc.). - #[serde(default)] + /// Action taken (typed enum serialized as `action_id` + `action` label). + #[serde(rename = "action_id", default, skip_serializing_if = "Option::is_none")] pub action: Option, - /// Disposition. - #[serde(default)] + /// Disposition (typed enum serialized as `disposition_id` + `disposition` label). + #[serde( + rename = "disposition_id", + default, + skip_serializing_if = "Option::is_none" + )] pub disposition: Option, } @@ -141,4 +145,22 @@ mod tests { assert_eq!(deserialized.api.operation, "POST /v1/messages"); assert!(deserialized.base.ai_model.is_some()); } + + #[test] + fn action_and_disposition_survive_a_roundtrip() { + // `insert_enum_pair!` writes `action_id` (u8) plus an `action` label, so + // the field must be renamed to read the id back rather than the label. + let mut event = test_api_activity(); + event.action = Some(ActionId::Allowed); + event.disposition = Some(DispositionId::Allowed); + + let json = serde_json::to_value(&event).unwrap(); + assert_eq!(json["action_id"], ActionId::Allowed.as_u8()); + assert_eq!(json["action"], ActionId::Allowed.label()); + assert_eq!(json["disposition_id"], DispositionId::Allowed.as_u8()); + + let deserialized: ApiActivityEvent = serde_json::from_value(json).unwrap(); + assert_eq!(deserialized.action, Some(ActionId::Allowed)); + assert_eq!(deserialized.disposition, Some(DispositionId::Allowed)); + } } diff --git a/crates/openshell-ocsf/src/lib.rs b/crates/openshell-ocsf/src/lib.rs index 345ea57175..fe1f60aa5a 100644 --- a/crates/openshell-ocsf/src/lib.rs +++ b/crates/openshell-ocsf/src/lib.rs @@ -44,8 +44,8 @@ pub use events::{ // --- Enum types --- pub use enums::{ - ActionId, ActivityId, AuthTypeId, ConfidenceId, DispositionId, HttpMethod, LaunchTypeId, - OcsfEnum, RiskLevelId, SecurityLevelId, SeverityId, StateId, StatusId, + ActionId, ActivityId, AuthTypeId, ConfidenceId, DeviceTypeId, DispositionId, HttpMethod, + LaunchTypeId, OcsfEnum, RiskLevelId, SecurityLevelId, SeverityId, StateId, StatusId, }; // --- Object types --- @@ -58,8 +58,8 @@ pub use objects::{ // --- Builders --- pub use builders::{ ApiActivityBuilder, AppLifecycleBuilder, BaseEventBuilder, ConfigStateChangeBuilder, - DetectionFindingBuilder, HttpActivityBuilder, NetworkActivityBuilder, ProcessActivityBuilder, - SandboxContext, SshActivityBuilder, + DetectionFindingBuilder, EventOrigin, HttpActivityBuilder, NetworkActivityBuilder, + ProcessActivityBuilder, SandboxContext, SshActivityBuilder, }; // --- Tracing layers --- diff --git a/crates/openshell-ocsf/src/objects/device.rs b/crates/openshell-ocsf/src/objects/device.rs index 4c42fb4a1f..8e6c7bff3e 100644 --- a/crates/openshell-ocsf/src/objects/device.rs +++ b/crates/openshell-ocsf/src/objects/device.rs @@ -5,12 +5,29 @@ use serde::{Deserialize, Serialize}; +use crate::enums::DeviceTypeId; + /// OCSF Device object. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Device { /// Device hostname. pub hostname: String, + /// Administrator-assigned device name, when one exists. + #[serde(skip_serializing_if = "Option::is_none")] + pub name: Option, + + /// Stable unique identifier for the device. + #[serde(skip_serializing_if = "Option::is_none")] + pub uid: Option, + + /// Device type id. Required by the OCSF schema. + pub type_id: DeviceTypeId, + + /// Sibling label for `type_id`. + #[serde(rename = "type")] + pub type_label: String, + /// Operating system info. #[serde(skip_serializing_if = "Option::is_none")] pub os: Option, @@ -29,11 +46,26 @@ impl Device { pub fn linux(hostname: &str) -> Self { Self { hostname: hostname.to_string(), + name: None, + uid: None, + type_id: DeviceTypeId::Server, + type_label: DeviceTypeId::Server.label().to_string(), os: Some(OsInfo { name: "Linux".to_string(), }), } } + + /// Create the device for a gateway replica. + #[must_use] + pub fn gateway(hostname: &str, name: &str) -> Self { + Self { + name: Some(name.to_string()), + // Keep the replica identity opaque rather than encoding multiple fields in the UID. + uid: Some(hostname.to_string()), + ..Self::linux(hostname) + } + } } #[cfg(test)] @@ -47,4 +79,31 @@ mod tests { assert_eq!(json["hostname"], "sandbox-abc123"); assert_eq!(json["os"]["name"], "Linux"); } + + #[test] + fn device_emits_the_schema_required_type_id() { + let json = serde_json::to_value(Device::linux("sandbox-abc123")).unwrap(); + assert_eq!(json["type_id"], DeviceTypeId::Server.as_u8()); + assert_eq!(json["type"], "Server"); + } + + #[test] + fn device_round_trips() { + let device = Device::linux("sandbox-abc123"); + let json = serde_json::to_value(&device).unwrap(); + let decoded: Device = serde_json::from_value(json.clone()).unwrap(); + assert_eq!(decoded, device); + assert_eq!(serde_json::to_value(&decoded).unwrap(), json); + } + + #[test] + fn gateway_replicas_have_distinct_device_uids() { + let first = Device::gateway("openshell-gateway-0", "production"); + let second = Device::gateway("openshell-gateway-1", "production"); + + assert_ne!( + first.uid, second.uid, + "gateway replicas must have distinct OCSF device UIDs" + ); + } } diff --git a/crates/openshell-ocsf/src/objects/metadata.rs b/crates/openshell-ocsf/src/objects/metadata.rs index 060f13ac6a..c580d0d840 100644 --- a/crates/openshell-ocsf/src/objects/metadata.rs +++ b/crates/openshell-ocsf/src/objects/metadata.rs @@ -18,7 +18,7 @@ pub struct Metadata { #[serde(skip_serializing_if = "Vec::is_empty")] pub profiles: Vec, - /// Unique event source identifier (sandbox ID). + /// Unique event identifier. #[serde(skip_serializing_if = "Option::is_none")] pub uid: Option, @@ -51,6 +51,16 @@ impl Product { version: Some(version.to_string()), } } + + /// Create the `OpenShell` Gateway product, for control-plane events. + #[must_use] + pub fn openshell_gateway(version: &str) -> Self { + Self { + name: "OpenShell Gateway".to_string(), + vendor_name: "OpenShell".to_string(), + version: Some(version.to_string()), + } + } } #[cfg(test)] diff --git a/crates/openshell-ocsf/tests/event_identity.rs b/crates/openshell-ocsf/tests/event_identity.rs new file mode 100644 index 0000000000..4e5ef2e7ef --- /dev/null +++ b/crates/openshell-ocsf/tests/event_identity.rs @@ -0,0 +1,115 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! `metadata.uid` identifies the event; `container.uid` identifies the sandbox. + +use std::net::{IpAddr, Ipv4Addr}; + +use openshell_ocsf::{ + ActivityId, EventOrigin, NetworkActivityBuilder, OcsfEvent, SandboxContext, SeverityId, +}; + +fn sandbox_ctx() -> SandboxContext { + SandboxContext { + sandbox_id: "sb-1".to_string(), + sandbox_name: "agent-01".to_string(), + container_image: "ghcr.io/nvidia/openshell/sandbox:0.42.1".to_string(), + hostname: "openshell-sb-1".to_string(), + product_version: "0.42.1".to_string(), + proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + proxy_port: 8888, + origin: EventOrigin::Sandbox, + } +} + +fn gateway_ctx(sandbox_id: &str, sandbox_name: &str) -> SandboxContext { + SandboxContext { + sandbox_id: sandbox_id.to_string(), + sandbox_name: sandbox_name.to_string(), + container_image: String::new(), + hostname: "openshell-gateway-0".to_string(), + product_version: "0.42.1".to_string(), + proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + proxy_port: 0, + origin: EventOrigin::Gateway { + name: "production-us-west".to_string(), + }, + } +} + +fn event(ctx: &SandboxContext) -> OcsfEvent { + NetworkActivityBuilder::new(ctx) + .activity(ActivityId::Open) + .severity(SeverityId::Medium) + .message("CONNECT api.example.com:443") + .build() +} + +#[test] +fn each_event_gets_its_own_metadata_uid() { + let ctx = sandbox_ctx(); + + let first = event(&ctx); + let second = event(&ctx); + + let first_uid = first.base().metadata.uid.clone().expect("uid is set"); + let second_uid = second.base().metadata.uid.clone().expect("uid is set"); + + assert!(!first_uid.is_empty()); + assert_ne!( + first_uid, second_uid, + "a shared uid invites a SIEM to dedup distinct events" + ); +} + +#[test] +fn metadata_uid_is_no_longer_the_sandbox_id() { + let event = event(&sandbox_ctx()); + + assert_ne!( + event.base().metadata.uid.as_deref(), + Some("sb-1"), + "metadata.uid is a per-event identifier, not a producer identifier" + ); +} + +#[test] +fn the_sandbox_id_is_carried_by_the_container() { + let json = event(&sandbox_ctx()).to_json().expect("serializes"); + + assert_eq!(json["container"]["uid"], "sb-1"); + assert_eq!(json["container"]["name"], "agent-01"); +} + +#[test] +fn a_gateway_event_about_a_sandbox_still_names_that_container() { + let json = event(&gateway_ctx("sb-7", "agent-07")) + .to_json() + .expect("serializes"); + + assert_eq!(json["container"]["uid"], "sb-7"); + assert_eq!(json["container"]["name"], "agent-07"); + assert_eq!(json["metadata"]["product"]["name"], "OpenShell Gateway"); +} + +#[test] +fn a_gateway_event_about_no_sandbox_omits_the_container() { + let json = event(&gateway_ctx("", "")).to_json().expect("serializes"); + + assert!( + json.get("container").is_none(), + "an event with no sandbox association has no container: {json}" + ); +} + +#[test] +fn a_container_without_an_image_omits_the_image() { + let json = event(&gateway_ctx("sb-7", "agent-07")) + .to_json() + .expect("serializes"); + + assert!( + json["container"].get("image").is_none(), + "an empty image reference is worse than none: {json}" + ); +} diff --git a/crates/openshell-ocsf/tests/gateway_context.rs b/crates/openshell-ocsf/tests/gateway_context.rs new file mode 100644 index 0000000000..a3fadb945b --- /dev/null +++ b/crates/openshell-ocsf/tests/gateway_context.rs @@ -0,0 +1,113 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Gateway-origin events carry gateway identity, not sandbox identity. + +use std::net::{IpAddr, Ipv4Addr}; + +use openshell_ocsf::{ + ActivityId, AppLifecycleBuilder, ConfigStateChangeBuilder, EventOrigin, SandboxContext, + SeverityId, StateId, StatusId, +}; + +fn gateway_ctx() -> SandboxContext { + SandboxContext { + sandbox_id: String::new(), + sandbox_name: String::new(), + container_image: String::new(), + hostname: "openshell-gateway-0".to_string(), + product_version: "0.42.1".to_string(), + proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + proxy_port: 0, + origin: EventOrigin::Gateway { + name: "production-us-west".to_string(), + }, + } +} + +fn sandbox_ctx() -> SandboxContext { + SandboxContext { + sandbox_id: "sb-1".to_string(), + sandbox_name: "agent-01".to_string(), + container_image: "ghcr.io/nvidia/openshell/sandbox:0.42.1".to_string(), + hostname: "openshell-sb-1".to_string(), + product_version: "0.42.1".to_string(), + proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + proxy_port: 8888, + origin: EventOrigin::Sandbox, + } +} + +#[test] +fn gateway_events_report_the_gateway_product() { + let event = AppLifecycleBuilder::new(&gateway_ctx()) + .activity(ActivityId::Open) + .severity(SeverityId::Informational) + .message("gateway started") + .build(); + let json = event.to_json().unwrap(); + + assert_eq!(json["metadata"]["product"]["name"], "OpenShell Gateway"); + assert_eq!(json["metadata"]["product"]["vendor_name"], "OpenShell"); +} + +#[test] +fn sandbox_events_still_report_the_supervisor_product() { + let event = AppLifecycleBuilder::new(&sandbox_ctx()) + .activity(ActivityId::Open) + .severity(SeverityId::Informational) + .message("supervisor started") + .build(); + let json = event.to_json().unwrap(); + + assert_eq!( + json["metadata"]["product"]["name"], + "OpenShell Sandbox Supervisor" + ); +} + +#[test] +fn gateway_events_identify_the_device_by_operator_assigned_name() { + let event = ConfigStateChangeBuilder::new(&gateway_ctx()) + .state(StateId::Enabled, "reloaded") + .severity(SeverityId::Informational) + .status(StatusId::Success) + .message("TLS certificate config reloaded") + .build(); + let json = event.to_json().unwrap(); + + assert_eq!(json["device"]["name"], "production-us-west"); + assert_eq!(json["device"]["uid"], "openshell-gateway-0"); + assert_eq!(json["device"]["hostname"], "openshell-gateway-0"); +} + +#[test] +fn gateway_events_omit_the_container_object() { + let event = ConfigStateChangeBuilder::new(&gateway_ctx()) + .state(StateId::Enabled, "reloaded") + .severity(SeverityId::Informational) + .status(StatusId::Success) + .message("TLS certificate config reloaded") + .build(); + let json = event.to_json().unwrap(); + + assert!( + json.get("container").is_none(), + "a gateway event without a sandbox association should omit container: {json}" + ); +} + +#[test] +fn sandbox_events_still_carry_their_container() { + let event = ConfigStateChangeBuilder::new(&sandbox_ctx()) + .state(StateId::Enabled, "loaded") + .severity(SeverityId::Informational) + .status(StatusId::Success) + .message("policy loaded") + .build(); + let json = event.to_json().unwrap(); + + assert_eq!(json["container"]["name"], "agent-01"); + assert_eq!(json["container"]["uid"], "sb-1"); + assert!(json["device"].get("name").is_none()); +} diff --git a/crates/openshell-ocsf/tests/roundtrip.rs b/crates/openshell-ocsf/tests/roundtrip.rs index 1c8a8bb307..e3f3e095e0 100644 --- a/crates/openshell-ocsf/tests/roundtrip.rs +++ b/crates/openshell-ocsf/tests/roundtrip.rs @@ -12,10 +12,10 @@ use std::net::{IpAddr, Ipv4Addr}; use openshell_ocsf::{ ActionId, ActivityId, AiModel, ApiActivityBuilder, AppLifecycleBuilder, Attack, AuthTypeId, BaseEventBuilder, ConfidenceId, ConfigStateChangeBuilder, ConnectionInfo, - DetectionFindingBuilder, DispositionId, Endpoint, FindingInfo, HttpActivityBuilder, HttpMethod, - HttpRequest, HttpResponse, LaunchTypeId, NetworkActivityBuilder, OcsfEvent, Process, - ProcessActivityBuilder, RiskLevelId, SandboxContext, SecurityLevelId, SeverityId, - SshActivityBuilder, StateId, StatusId, Url, + DetectionFindingBuilder, DispositionId, Endpoint, EventOrigin, FindingInfo, + HttpActivityBuilder, HttpMethod, HttpRequest, HttpResponse, LaunchTypeId, + NetworkActivityBuilder, OcsfEvent, Process, ProcessActivityBuilder, RiskLevelId, + SandboxContext, SecurityLevelId, SeverityId, SshActivityBuilder, StateId, StatusId, Url, }; fn ctx() -> SandboxContext { @@ -27,6 +27,7 @@ fn ctx() -> SandboxContext { product_version: "0.42.1".to_string(), proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), proxy_port: 8888, + origin: EventOrigin::Sandbox, } } diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index 41c6ca3c94..4c34cb0922 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -150,6 +150,7 @@ pub async fn run_sandbox( product_version: openshell_core::VERSION.to_string(), proxy_ip: std::net::IpAddr::from([127, 0, 0, 1]), proxy_port: 3128, + origin: openshell_ocsf::EventOrigin::Sandbox, }) { debug!("OCSF context already initialized, keeping existing"); } diff --git a/crates/openshell-server/src/cli.rs b/crates/openshell-server/src/cli.rs index 33008774b9..7f85fcacd8 100644 --- a/crates/openshell-server/src/cli.rs +++ b/crates/openshell-server/src/cli.rs @@ -504,6 +504,15 @@ async fn run_from_args( ) -> Result<()> { let prepared = prepare_server_config_with_drivers(&mut args, &matches, &compute_drivers)?; + // Initialize OCSF identity before tracing can emit gateway events. + let gateway_identity = crate::gateway_ocsf::GatewayIdentity { + name: prepared.config.name.clone(), + hostname: crate::compute::lease::replica_id(), + }; + if !crate::gateway_ocsf::set_identity(gateway_identity) { + tracing::debug!("gateway OCSF identity already initialized, keeping existing"); + } + let tracing_log_bus = TracingLogBus::new(); let otlp_config = prepared .config_file diff --git a/crates/openshell-server/src/gateway_ocsf.rs b/crates/openshell-server/src/gateway_ocsf.rs new file mode 100644 index 0000000000..1e0c7de2af --- /dev/null +++ b/crates/openshell-server/src/gateway_ocsf.rs @@ -0,0 +1,90 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Process-wide OCSF identity for gateway-origin events. +//! +//! Gateway events are emitted from places with no access to the server config +//! (the TLS reload watcher, the service router), so the identity is resolved +//! once at startup rather than threaded through all of them. + +use std::sync::OnceLock; + +use openshell_ocsf::{EventOrigin, SandboxContext}; + +/// Identity shared by every gateway-origin OCSF event. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GatewayIdentity { + /// Operator-assigned gateway name, shared across replicas of one install. + pub name: String, + /// Per-replica hostname (the pod name under Kubernetes). + pub hostname: String, +} + +static IDENTITY: OnceLock = OnceLock::new(); + +/// Initialise the process-wide gateway identity. +/// +/// Returns `false` if it was already set; the caller may log and continue. +pub fn set_identity(identity: GatewayIdentity) -> bool { + IDENTITY.set(identity).is_ok() +} + +/// Return the gateway identity, falling back to placeholders when unset (in +/// tests, and in any code path that runs before startup completes). +#[must_use] +pub fn identity() -> GatewayIdentity { + IDENTITY.get().cloned().unwrap_or_else(|| GatewayIdentity { + name: openshell_core::config::DEFAULT_GATEWAY_NAME.to_string(), + hostname: "openshell-gateway".to_string(), + }) +} + +/// Build the OCSF context for a gateway-origin event. +/// +/// `sandbox_id` and `sandbox_name` describe the sandbox the event is *about*, +/// and may be empty. The emitting device is always the gateway. +#[must_use] +pub fn context(sandbox_id: &str, sandbox_name: &str) -> SandboxContext { + let identity = identity(); + SandboxContext { + sandbox_id: sandbox_id.to_string(), + sandbox_name: sandbox_name.to_string(), + container_image: String::new(), + hostname: identity.hostname, + product_version: openshell_core::VERSION.to_string(), + proxy_ip: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), + proxy_port: 0, + origin: EventOrigin::Gateway { + name: identity.name, + }, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn context_marks_events_as_gateway_origin() { + let ctx = context("sb-1", "agent-01"); + + assert!(matches!(ctx.origin, EventOrigin::Gateway { .. })); + assert_eq!(ctx.sandbox_id, "sb-1"); + assert_eq!(ctx.sandbox_name, "agent-01"); + } + + #[test] + fn gateway_context_produces_gateway_product_and_no_container() { + let ctx = context("", ""); + + assert_eq!(ctx.metadata(&[]).product.name, "OpenShell Gateway"); + assert!(ctx.container().is_none()); + } + + #[test] + fn identity_falls_back_when_unset() { + let identity = identity(); + assert!(!identity.name.is_empty()); + assert!(!identity.hostname.is_empty()); + } +} diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index e46b2ae8ae..681a4fd128 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -50,14 +50,11 @@ use openshell_core::telemetry::{ LifecycleOperation, LifecycleResource, PolicyDecisionOperation, TelemetryOutcome, }; use openshell_core::{ - VERSION, endpoint_path::EndpointPathPattern, host_pattern::{host_matches, host_patterns_overlap}, settings::{self, SettingValueKind}, }; -use openshell_ocsf::{ - ConfigStateChangeBuilder, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, -}; +use openshell_ocsf::{ConfigStateChangeBuilder, OcsfEvent, SeverityId, StateId, StatusId}; use openshell_policy::{ PolicyMergeOp, ProviderPolicyLayer, canonicalize_advisor_add_rule, compose_effective_policy, merge_policy, policy_covers_rule, serialize_sandbox_policy, strip_provider_rule_names, @@ -74,7 +71,7 @@ use openshell_prover::{ use prost::Message; use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, HashMap, HashSet}; -use std::net::{IpAddr, Ipv4Addr}; +use std::net::IpAddr; use std::sync::Arc; use tonic::{Request, Response, Status}; use tracing::{debug, info, warn}; @@ -217,15 +214,7 @@ fn build_gateway_policy_audit_event( policy_hash: &str, extra_fields: &[(&str, String)], ) -> OcsfEvent { - let ctx = SandboxContext { - sandbox_id: sandbox_id.to_string(), - sandbox_name: sandbox_name.to_string(), - container_image: "openshell/gateway".to_string(), - hostname: "openshell-gateway".to_string(), - product_version: VERSION.to_string(), - proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), - proxy_port: 0, - }; + let ctx = crate::gateway_ocsf::context(sandbox_id, sandbox_name); let mut builder = ConfigStateChangeBuilder::new(&ctx) .state(StateId::Other, state_label) .severity(SeverityId::Informational) diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index a8c8afdf08..3eec00540c 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -21,6 +21,7 @@ pub mod config_file; mod credentials; mod defaults; mod gateway_listener; +mod gateway_ocsf; mod grpc; mod http; mod inference; diff --git a/crates/openshell-server/src/service_routing.rs b/crates/openshell-server/src/service_routing.rs index 1ca575c059..5ca97f9735 100644 --- a/crates/openshell-server/src/service_routing.rs +++ b/crates/openshell-server/src/service_routing.rs @@ -9,15 +9,14 @@ use axum::{ }; use http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode, header}; use hyper_util::rt::TokioIo; +use openshell_core::ObjectId; use openshell_core::config::ServiceRoutingConfig; use openshell_core::proto::{Sandbox, SandboxPhase, ServiceEndpoint, TcpRelayTarget, relay_open}; -use openshell_core::{ObjectId, VERSION}; use openshell_ocsf::{ ActionId, ActivityId, ConfigStateChangeBuilder, DispositionId, Endpoint, HttpActivityBuilder, HttpRequest, HttpResponse as OcsfHttpResponse, NetworkActivityBuilder, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, Url as OcsfUrl, }; -use std::net::{IpAddr, Ipv4Addr}; use std::sync::Arc; use std::time::Duration; use tokio::io::AsyncWriteExt; @@ -796,15 +795,7 @@ fn emit_gateway_ocsf_event(event: OcsfEvent) { } fn gateway_ocsf_ctx(sandbox_id: &str, sandbox_name: &str) -> SandboxContext { - SandboxContext { - sandbox_id: sandbox_id.to_string(), - sandbox_name: sandbox_name.to_string(), - container_image: "openshell/gateway".to_string(), - hostname: "openshell-gateway".to_string(), - product_version: VERSION.to_string(), - proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), - proxy_port: 0, - } + crate::gateway_ocsf::context(sandbox_id, sandbox_name) } fn endpoint_name(endpoint: &ServiceEndpoint) -> String { @@ -1337,7 +1328,11 @@ mod tests { ); assert_eq!( - event.base().metadata.uid.as_deref(), + event + .base() + .container + .as_ref() + .and_then(|container| container.uid.as_deref()), Some("sandbox-1"), "resolved sandbox id should reach the event" ); diff --git a/crates/openshell-server/src/tls.rs b/crates/openshell-server/src/tls.rs index e3b48b9fbb..7b8323cf5c 100644 --- a/crates/openshell-server/src/tls.rs +++ b/crates/openshell-server/src/tls.rs @@ -464,15 +464,7 @@ fn load_key(path: &Path) -> Result> { /// Build an OCSF context for gateway-level (non-sandbox) events. fn tls_ocsf_ctx() -> SandboxContext { - SandboxContext { - sandbox_id: String::new(), - sandbox_name: String::new(), - container_image: "openshell/gateway".to_string(), - hostname: "openshell-gateway".to_string(), - product_version: openshell_core::VERSION.to_string(), - proxy_ip: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), - proxy_port: 0, - } + crate::gateway_ocsf::context("", "") } #[cfg(test)] diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 15c7908e34..3cf55dfbba 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -180,7 +180,7 @@ where let (visitor_sandbox_id, visitor_message) = if meta.target() == OCSF_TARGET { openshell_ocsf::clone_current_event().map_or((None, None), |ocsf_event| { ( - ocsf_event.base().metadata.uid.clone(), + ocsf_sandbox_id(&ocsf_event), Some(ocsf_event.format_shorthand()), ) }) @@ -218,6 +218,17 @@ where } } +/// The sandbox an event concerns, if any. +/// +/// `container.uid` rather than `metadata.uid`: the latter identifies the event. +fn ocsf_sandbox_id(event: &openshell_ocsf::OcsfEvent) -> Option { + event + .base() + .container + .as_ref() + .and_then(|container| container.uid.clone()) +} + #[derive(Debug, Default)] struct LogVisitor { sandbox_id: Option, @@ -353,6 +364,7 @@ mod tests { product_version: "0.0.0".to_string(), proxy_ip: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), proxy_port: 0, + origin: openshell_ocsf::EventOrigin::Sandbox, } } diff --git a/crates/openshell-supervisor-network/src/l7/middleware.rs b/crates/openshell-supervisor-network/src/l7/middleware.rs index 6305653f6a..bd799a6ab9 100644 --- a/crates/openshell-supervisor-network/src/l7/middleware.rs +++ b/crates/openshell-supervisor-network/src/l7/middleware.rs @@ -1095,6 +1095,7 @@ mod tests { product_version: "0".into(), proxy_ip: [127, 0, 0, 1].into(), proxy_port: 3128, + origin: openshell_ocsf::EventOrigin::Sandbox, }; let eval = L7EvalContext { diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index 2697fedb3c..e422cb4a42 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -2925,6 +2925,7 @@ mod tests { product_version: "0".into(), proxy_ip: [127, 0, 0, 1].into(), proxy_port: 3128, + origin: openshell_ocsf::EventOrigin::Sandbox, }; let eval = L7EvalContext { diff --git a/crates/openshell-supervisor-process/src/log_push.rs b/crates/openshell-supervisor-process/src/log_push.rs index 256fbff4a4..a80e2dee4e 100644 --- a/crates/openshell-supervisor-process/src/log_push.rs +++ b/crates/openshell-supervisor-process/src/log_push.rs @@ -331,6 +331,7 @@ mod tests { product_version: "0.0.0".to_string(), proxy_ip: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), proxy_port: 8888, + origin: openshell_ocsf::EventOrigin::Sandbox, } } diff --git a/crates/openshell-supervisor-process/src/supervisor_session.rs b/crates/openshell-supervisor-process/src/supervisor_session.rs index 98a3c0497b..7108e5c8e8 100644 --- a/crates/openshell-supervisor-process/src/supervisor_session.rs +++ b/crates/openshell-supervisor-process/src/supervisor_session.rs @@ -904,6 +904,7 @@ mod ocsf_event_tests { product_version: "0.0.1".into(), proxy_ip: "127.0.0.1".parse().unwrap(), proxy_port: 3128, + origin: openshell_ocsf::EventOrigin::Sandbox, } } From 2e2e2c8a927077c8ad2f3cdef9bb586a5d66d185 Mon Sep 17 00:00:00 2001 From: Kris Hicks Date: Fri, 28 Aug 2026 14:07:29 -0700 Subject: [PATCH 3/3] feat(ocsf): preserve structured events in sandbox logs Sandbox log streams now retain complete OCSF event payloads, allowing exporters and API consumers to use the original structured security and audit records without reconstructing them from display text. `openshell logs` continues to show readable shorthand, including across supported mixed-version deployments. Ordinary log lines and malformed or older structured payloads still display their original message instead of being dropped. Refs #1055 Signed-off-by: Kris Hicks --- crates/openshell-cli/src/run.rs | 1 + .../sandbox_create_lifecycle_integration.rs | 1 + crates/openshell-server/src/tracing_bus.rs | 76 ++++++++++ .../tests/ocsf_wire_equivalence.rs | 135 ++++++++++++++++++ .../src/log_push.rs | 93 ++++++++++-- proto/openshell.proto | 3 + sdk/go/proto/openshellv1/openshell.pb.go | 17 ++- 7 files changed, 308 insertions(+), 18 deletions(-) create mode 100644 crates/openshell-server/tests/ocsf_wire_equivalence.rs diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index 4cd4e34a76..fd30cb119e 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -10027,6 +10027,7 @@ mod tests { .iter() .map(|(k, v)| ((*k).to_string(), (*v).to_string())) .collect(), + ocsf_json: Vec::new(), } } diff --git a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs index 7be771c442..60ffc54341 100644 --- a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs +++ b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs @@ -641,6 +641,7 @@ impl OpenShell for TestOpenShell { message: message.to_string(), source: "gateway".to_string(), fields: HashMap::new(), + ocsf_json: Vec::new(), })), })) .await; diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 3cf55dfbba..7c81bbe5bb 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -113,6 +113,10 @@ impl TracingLogBus { if log.sandbox_id.is_empty() { return; } + let mut log = log; + if let Some(event) = decode_ocsf_event(&log) { + log.message = event.format_shorthand(); + } let evt = SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( log.clone(), @@ -163,6 +167,26 @@ impl TracingLogBus { } } +/// Decode the structured OCSF event a sandbox line carries, if any. +/// +/// A malformed payload is logged and ignored rather than dropping the line. +fn decode_ocsf_event(log: &SandboxLogLine) -> Option { + if log.ocsf_json.is_empty() { + return None; + } + match serde_json::from_slice(&log.ocsf_json) { + Ok(event) => Some(event), + Err(error) => { + tracing::warn!( + sandbox_id = %log.sandbox_id, + %error, + "discarding undecodable OCSF payload from sandbox log line" + ); + None + } + } +} + #[derive(Debug, Clone)] struct SandboxLogLayer { bus: TracingLogBus, @@ -208,6 +232,7 @@ where message: msg, source: "gateway".to_string(), fields: HashMap::new(), + ocsf_json: Vec::new(), }; let evt = SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( @@ -274,6 +299,7 @@ mod tests { message: message.to_string(), source: "gateway".to_string(), fields: HashMap::new(), + ocsf_json: Vec::new(), } } @@ -339,6 +365,56 @@ mod tests { bus.remove("nonexistent"); } + #[test] + fn external_lines_render_shorthand_from_the_structured_event() { + use openshell_ocsf::{ + ActionId, ActivityId, DispositionId, Endpoint, NetworkActivityBuilder, SeverityId, + StatusId, + }; + + let event = NetworkActivityBuilder::new(&ocsf_ctx("sb-wire")) + .activity(ActivityId::Open) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443)) + .message("CONNECT denied blocked.example.com:443") + .build(); + let expected = event.format_shorthand(); + + let bus = TracingLogBus::new(); + let mut line = make_log_event("sb-wire", ""); + line.ocsf_json = event.to_json_line().unwrap().into_bytes(); + bus.publish_external(line); + + let tail = bus.tail("sb-wire", 10); + assert_eq!(tail.len(), 1); + assert_eq!(log_message(&tail[0]).message, expected); + } + + #[test] + fn external_lines_without_ocsf_json_keep_their_message() { + let bus = TracingLogBus::new(); + bus.publish_external(make_log_event("sb-plain", "already rendered")); + + let tail = bus.tail("sb-plain", 10); + assert_eq!(log_message(&tail[0]).message, "already rendered"); + } + + #[test] + fn undecodable_ocsf_json_does_not_drop_the_line() { + let bus = TracingLogBus::new(); + let mut line = make_log_event("sb-bad", "fallback text"); + line.ocsf_json = b"{not valid json".to_vec(); + bus.publish_external(line); + + // A malformed payload must not silently swallow the record. + let tail = bus.tail("sb-bad", 10); + assert_eq!(tail.len(), 1); + assert_eq!(log_message(&tail[0]).message, "fallback text"); + } + #[test] fn publish_after_remove_does_not_resurrect_the_bus_entry() { let bus = TracingLogBus::new(); diff --git a/crates/openshell-server/tests/ocsf_wire_equivalence.rs b/crates/openshell-server/tests/ocsf_wire_equivalence.rs new file mode 100644 index 0000000000..fa26594b07 --- /dev/null +++ b/crates/openshell-server/tests/ocsf_wire_equivalence.rs @@ -0,0 +1,135 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Gateway-rendered log text must match what the sandbox would have rendered. +//! +//! A decode gap corrupts `openshell logs` output rather than erroring. + +use std::net::{IpAddr, Ipv4Addr}; + +use openshell_ocsf::{ + ActionId, ActivityId, AppLifecycleBuilder, ConfigStateChangeBuilder, DetectionFindingBuilder, + DispositionId, Endpoint, EventOrigin, FindingInfo, HttpActivityBuilder, HttpMethod, + HttpRequest, HttpResponse, NetworkActivityBuilder, OcsfEvent, Process, ProcessActivityBuilder, + SandboxContext, SeverityId, SshActivityBuilder, StateId, StatusId, Url, +}; + +fn ctx() -> SandboxContext { + SandboxContext { + sandbox_id: "sb-1".to_string(), + sandbox_name: "agent-01".to_string(), + container_image: "ghcr.io/nvidia/openshell/sandbox:0.42.1".to_string(), + hostname: "openshell-sb-1".to_string(), + product_version: "0.42.1".to_string(), + proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + proxy_port: 8888, + origin: EventOrigin::Sandbox, + } +} + +/// Serialize as the supervisor does, then decode as the gateway does. +fn across_the_wire(event: &OcsfEvent) -> OcsfEvent { + let bytes = event.to_json_line().expect("serialize").into_bytes(); + serde_json::from_slice(&bytes).expect("gateway should decode the payload") +} + +fn assert_renders_identically(label: &str, event: &OcsfEvent) { + assert_eq!( + across_the_wire(event).format_shorthand(), + event.format_shorthand(), + "{label}: gateway-rendered text differs from sandbox-rendered text" + ); +} + +#[test] +fn network_activity_renders_identically_across_the_wire() { + let event = NetworkActivityBuilder::new(&ctx()) + .activity(ActivityId::Open) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .dst_endpoint(Endpoint::from_domain("api.example.com", 443)) + .src_endpoint_addr(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 5)), 51234) + .actor_process(Process::new("/usr/bin/curl", 4711).with_cmd_line("curl -sS https://x")) + .firewall_rule("default-deny-egress", "opa") + .message("CONNECT denied api.example.com:443") + .build(); + assert_renders_identically("network_activity", &event); +} + +#[test] +fn http_activity_renders_identically_across_the_wire() { + let event = HttpActivityBuilder::new(&ctx()) + .activity(ActivityId::Open) + .action(ActionId::Allowed) + .disposition(DispositionId::Allowed) + .severity(SeverityId::Informational) + .status(StatusId::Success) + .http_request(HttpRequest { + http_method: HttpMethod::Get, + url: Some(Url::new("https", "api.example.com", "/v1/items", 443)), + }) + .http_response(HttpResponse { code: 200 }) + .message("GET /v1/items 200") + .build(); + assert_renders_identically("http_activity", &event); +} + +#[test] +fn ssh_activity_renders_identically_across_the_wire() { + let event = SshActivityBuilder::new(&ctx()) + .activity(ActivityId::Open) + .severity(SeverityId::Informational) + .status(StatusId::Success) + .dst_endpoint(Endpoint::from_domain("sandbox.local", 22)) + .message("ssh session accepted") + .build(); + assert_renders_identically("ssh_activity", &event); +} + +#[test] +fn process_activity_renders_identically_across_the_wire() { + let event = ProcessActivityBuilder::new(&ctx()) + .activity(ActivityId::Open) + .severity(SeverityId::Informational) + .status(StatusId::Success) + .process(Process::new("/usr/bin/python3", 4713).with_cmd_line("python3 -m pytest")) + .message("process started") + .build(); + assert_renders_identically("process_activity", &event); +} + +#[test] +fn detection_finding_renders_identically_across_the_wire() { + let event = DetectionFindingBuilder::new(&ctx()) + .finding_info(FindingInfo::new("finding-1", "Sandbox bypass attempt")) + .severity(SeverityId::High) + .is_alert(true) + .evidence("dst_host", "169.254.169.254") + .message("bypass attempt detected") + .build(); + assert_renders_identically("detection_finding", &event); +} + +#[test] +fn config_state_change_renders_identically_across_the_wire() { + let event = ConfigStateChangeBuilder::new(&ctx()) + .state(StateId::Other, "policy-loaded") + .severity(SeverityId::Informational) + .status(StatusId::Success) + .message("policy reloaded") + .unmapped("policy_version", 7) + .build(); + assert_renders_identically("config_state_change", &event); +} + +#[test] +fn application_lifecycle_renders_identically_across_the_wire() { + let event = AppLifecycleBuilder::new(&ctx()) + .activity(ActivityId::Open) + .severity(SeverityId::Informational) + .message("supervisor started") + .build(); + assert_renders_identically("application_lifecycle", &event); +} diff --git a/crates/openshell-supervisor-process/src/log_push.rs b/crates/openshell-supervisor-process/src/log_push.rs index a80e2dee4e..2651ee147e 100644 --- a/crates/openshell-supervisor-process/src/log_push.rs +++ b/crates/openshell-supervisor-process/src/log_push.rs @@ -51,22 +51,27 @@ impl Layer for LogPushLayer { return; } - // OCSF events carry their payload in a thread-local; extract the - // shorthand representation for the push message. Non-OCSF events - // use the original visitor-based extraction. - let (msg, fields) = if meta.target() == openshell_ocsf::OCSF_TARGET { - if let Some(ocsf_event) = openshell_ocsf::clone_current_event() { - ( - ocsf_event.format_shorthand(), - std::collections::HashMap::new(), - ) - } else { + // OCSF events carry their payload in a thread-local. Send both the + // structured event and shorthand display text: older gateways ignore + // `ocsf_json`, so `message` remains the mixed-version fallback. + let (msg, fields, ocsf_json) = if meta.target() == openshell_ocsf::OCSF_TARGET { + let Some(ocsf_event) = openshell_ocsf::clone_current_event() else { return; - } + }; + let shorthand = ocsf_event.format_shorthand(); + let Ok(json) = ocsf_event.to_json_line() else { + return; + }; + ( + shorthand, + std::collections::HashMap::new(), + json.into_bytes(), + ) } else { let mut visitor = LogVisitor::default(); event.record(&mut visitor); - visitor.into_parts(meta.name()) + let (msg, fields) = visitor.into_parts(meta.name()); + (msg, fields, Vec::new()) }; let ts = openshell_core::time::now_ms(); @@ -85,6 +90,7 @@ impl Layer for LogPushLayer { message: msg, source: "sandbox".to_string(), fields, + ocsf_json, }; // Best-effort: drop if the channel is full (don't block tracing). @@ -102,7 +108,6 @@ pub fn spawn_log_push_task( sandbox_id: String, ) -> (mpsc::Sender, tokio::task::JoinHandle<()>) { let (tx, rx) = mpsc::channel::(1024); - let handle = tokio::spawn(run_push_loop(endpoint, sandbox_id, rx)); (tx, handle) @@ -354,7 +359,7 @@ mod tests { } #[test] - fn ocsf_events_push_shorthand_with_ocsf_level_and_no_fields() { + fn ocsf_events_push_the_structured_event_with_shorthand_fallback() { let event = NetworkActivityBuilder::new(&ocsf_ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) @@ -364,9 +369,67 @@ mod tests { .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443)) .message("CONNECT denied blocked.example.com:443".to_string()) .build(); + let expected_json = event.to_json().expect("serialize"); let expected_shorthand = event.format_shorthand(); let lines = capture(16, || ocsf_emit!(event)); + assert_eq!(lines.len(), 1); + let line = &lines[0]; + + assert!( + !line.ocsf_json.is_empty(), + "structured event should be sent" + ); + let decoded: serde_json::Value = + serde_json::from_slice(&line.ocsf_json).expect("payload should be valid JSON"); + assert_eq!(decoded, expected_json); + + // Older gateways ignore `ocsf_json`, so the shorthand remains in the + // legacy message field as a mixed-version fallback. + assert_eq!(line.message, expected_shorthand); + + // What the receiver will render must match what the sandbox would have. + let decoded_event: openshell_ocsf::OcsfEvent = + serde_json::from_slice(&line.ocsf_json).expect("payload should decode"); + assert_eq!(decoded_event.format_shorthand(), expected_shorthand); + } + + #[test] + fn remove_mixed_version_shorthand_fallback_after_2026_10_15() { + const REMOVE_FALLBACK_AFTER_UNIX_SECS: u64 = 1_792_022_400; + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock should be after the Unix epoch") + .as_secs(); + + assert!( + now < REMOVE_FALLBACK_AFTER_UNIX_SECS, + "remove the OCSF shorthand message fallback now that gateways predating ocsf_json are no longer supported" + ); + } + + #[test] + fn non_ocsf_lines_carry_no_ocsf_payload() { + let lines = capture(16, || { + tracing::info!(target: "test_target", "plain line"); + }); + assert!(lines[0].ocsf_json.is_empty()); + assert_eq!(lines[0].message, "plain line"); + } + + #[test] + fn ocsf_events_push_with_ocsf_level_and_no_fields() { + let event = NetworkActivityBuilder::new(&ocsf_ctx()) + .activity(ActivityId::Open) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443)) + .message("CONNECT denied blocked.example.com:443".to_string()) + .build(); + let lines = capture(16, || ocsf_emit!(event)); assert_eq!(lines.len(), 1); let line = &lines[0]; @@ -374,7 +437,6 @@ mod tests { assert_eq!(line.target, openshell_ocsf::OCSF_TARGET); assert_eq!(line.source, "sandbox"); assert_eq!(line.sandbox_id, "sb-test"); - assert_eq!(line.message, expected_shorthand); assert!(line.fields.is_empty()); assert!(line.timestamp_ms > 0); } @@ -454,6 +516,7 @@ mod tests { message: message.to_string(), source: "sandbox".to_string(), fields: std::collections::HashMap::new(), + ocsf_json: Vec::new(), } } diff --git a/proto/openshell.proto b/proto/openshell.proto index 138b973474..272ee85893 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1566,6 +1566,9 @@ message SandboxLogLine { string source = 6; // Structured key-value fields from the tracing event (e.g. dst_host, action). map fields = 7; + // Full OCSF event as JSON, for OCSF lines only. When empty, `message` is + // used as-is. + bytes ocsf_json = 8; } message SandboxStreamWarning { diff --git a/sdk/go/proto/openshellv1/openshell.pb.go b/sdk/go/proto/openshellv1/openshell.pb.go index f9e7f39030..3e16ec0e82 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -5367,7 +5367,10 @@ type SandboxLogLine struct { // Empty is treated as "gateway" for backward compatibility. Source string `protobuf:"bytes,6,opt,name=source,proto3" json:"source,omitempty"` // Structured key-value fields from the tracing event (e.g. dst_host, action). - Fields map[string]string `protobuf:"bytes,7,rep,name=fields,proto3" json:"fields,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + Fields map[string]string `protobuf:"bytes,7,rep,name=fields,proto3" json:"fields,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // Full OCSF event as JSON, for OCSF lines only. When empty, `message` is + // used as-is. + OcsfJson []byte `protobuf:"bytes,8,opt,name=ocsf_json,json=ocsfJson,proto3" json:"ocsf_json,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -5451,6 +5454,13 @@ func (x *SandboxLogLine) GetFields() map[string]string { return nil } +func (x *SandboxLogLine) GetOcsfJson() []byte { + if x != nil { + return x.OcsfJson + } + return nil +} + type SandboxStreamWarning struct { state protoimpl.MessageState `protogen:"open.v1"` Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"` @@ -15219,7 +15229,7 @@ const file_openshell_proto_rawDesc = "" + "\x05event\x18\x03 \x01(\v2\x1b.openshell.v1.PlatformEventH\x00R\x05event\x12>\n" + "\awarning\x18\x04 \x01(\v2\".openshell.v1.SandboxStreamWarningH\x00R\awarning\x12Q\n" + "\x13draft_policy_update\x18\x05 \x01(\v2\x1f.openshell.v1.DraftPolicyUpdateH\x00R\x11draftPolicyUpdateB\t\n" + - "\apayload\"\xaf\x02\n" + + "\apayload\"\xcc\x02\n" + "\x0eSandboxLogLine\x12\x1d\n" + "\n" + "sandbox_id\x18\x01 \x01(\tR\tsandboxId\x12!\n" + @@ -15228,7 +15238,8 @@ const file_openshell_proto_rawDesc = "" + "\x06target\x18\x04 \x01(\tR\x06target\x12\x18\n" + "\amessage\x18\x05 \x01(\tR\amessage\x12\x16\n" + "\x06source\x18\x06 \x01(\tR\x06source\x12@\n" + - "\x06fields\x18\a \x03(\v2(.openshell.v1.SandboxLogLine.FieldsEntryR\x06fields\x1a9\n" + + "\x06fields\x18\a \x03(\v2(.openshell.v1.SandboxLogLine.FieldsEntryR\x06fields\x12\x1b\n" + + "\tocsf_json\x18\b \x01(\fR\bocsfJson\x1a9\n" + "\vFieldsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"0\n" +