From 2e2e2c8a927077c8ad2f3cdef9bb586a5d66d185 Mon Sep 17 00:00:00 2001 From: Kris Hicks Date: Fri, 28 Aug 2026 14:07:29 -0700 Subject: [PATCH] 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" +