diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index b819be92f8..37fa33d40d 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -9020,6 +9020,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 118daff902..274d0deb9c 100644 --- a/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs +++ b/crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs @@ -523,6 +523,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..72f30cbaba 100644 --- a/crates/openshell-supervisor-process/src/log_push.rs +++ b/crates/openshell-supervisor-process/src/log_push.rs @@ -51,22 +51,25 @@ 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 the event + // itself; the gateway renders the display text from it. + let (msg, fields, ocsf_json) = if meta.target() == openshell_ocsf::OCSF_TARGET { + let Some(ocsf_event) = openshell_ocsf::clone_current_event() else { return; - } + }; + let Ok(json) = ocsf_event.to_json_line() else { + return; + }; + ( + String::new(), + 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 +88,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 +106,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 +357,7 @@ mod tests { } #[test] - fn ocsf_events_push_shorthand_with_ocsf_level_and_no_fields() { + fn ocsf_events_push_the_structured_event_not_shorthand() { let event = NetworkActivityBuilder::new(&ocsf_ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) @@ -364,9 +367,51 @@ 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); + + // The receiver renders the display text, so it is not sent. + assert!(line.message.is_empty()); + + // 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 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 +419,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 +498,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 42e635927f..963287f091 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1420,6 +1420,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 0d2b86fce0..9e687b2dfa 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -4574,7 +4574,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 } @@ -4658,6 +4661,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"` @@ -14377,7 +14387,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" + @@ -14386,7 +14396,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" +