diff --git a/Cargo.lock b/Cargo.lock index 6a7436cb90..4704e6acd7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4458,6 +4458,7 @@ dependencies = [ "axum", "base64", "bytes", + "chrono", "clap", "futures", "futures-util", diff --git a/architecture/gateway.md b/architecture/gateway.md index 59b2bd52ed..2383251464 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -821,6 +821,35 @@ startup and disables export. Export is best-effort — the SDK logs runtime failures, and a failed batch is dropped rather than retried. Buffered spans flush after the server loop exits so `SIGTERM` does not drop in-flight traces. +### OCSF file collection + +The gateway owns one optional native OCSF JSONL output, configured through +`openshell.gateway.ocsf_log`. An in-process bounded queue captures +gateway-produced OCSF records independently of diagnostic filtering and sandbox +log subscriptions, including events without a sandbox association. It preserves +event identity and does not translate records into destination-specific schemas. + +Console formatting and sandbox log subscriptions also read the structured event: +the console renders shorthand, and the log bus routes by the affected container's +UID. Gateway-wide events have no sandbox subscription. These consumers remain +available without enabling JSONL output. Event origin identifies the producer +(`Gateway` or `Supervisor`), independently of the affected sandbox. + +Tracing callbacks queue serialized records with count and byte bounds; overflow +drops incoming records. A dedicated filesystem worker owns appends, recovery of +incomplete tails, daily rename-based rotation, and retention. Failed writes are +not replayed, and subsequent records are discarded during bounded reopen backoff. +Metrics expose writer failures and known or uncertain losses outside the output. +Tail recovery reports discarded bytes, not an invented missing-record count. + +Each replica requires its own file. An external shipper owns rotation discovery, +checkpoints, remote retries, credentials, and SIEM or OTLP conversion. Retention +must exceed expected shipper outages to avoid pruning unread segments. Concurrent +writers and external copy-truncate rotation are unsupported. Output failures do +not block sandbox execution. Shutdown allows five seconds to drain; an outstanding +OS write cannot be cancelled and remains uncertain. Flush is not fsync, and neither +the queue nor append results promise durable acceptance or audit completeness. + ### Package-managed gateway registry The CLI reads its active-gateway and per-gateway metadata from diff --git a/architecture/sandbox.md b/architecture/sandbox.md index 7be66e97e6..5cb4542c20 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -457,6 +457,9 @@ Security-relevant sandbox behavior uses OCSF structured events; internal diagnostics use ordinary tracing. The OCSF device describes the sandbox environment, with type ID Other and type label `Sandbox`; its operating system is a separate attribute. +Gateway-origin events describe the gateway process's OS, independently of any +associated sandbox. Linux supervisor events retain Linux as their device OS, +including when a native macOS or Windows gateway manages the sandbox. ## Policy Proposals diff --git a/crates/openshell-ocsf/src/builders/mod.rs b/crates/openshell-ocsf/src/builders/mod.rs index eacd608b07..c8cc301f2b 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 supervisor process enforcing sandbox policy. + #[default] + Supervisor, + /// 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,15 +208,21 @@ pub struct EventContext { pub proxy_ip: IpAddr, /// Proxy listen port. pub proxy_port: u16, + /// Which component is emitting. + pub origin: EventOrigin, } impl EventContext { /// Build the OCSF `Metadata` object for any event. #[must_use] pub fn metadata(&self, profiles: &[&str]) -> Metadata { + let product = match self.origin { + EventOrigin::Supervisor => 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(uuid::Uuid::new_v4().to_string()), log_source: None, @@ -228,7 +247,10 @@ impl EventContext { /// Build the OCSF `Device` object. #[must_use] pub fn device(&self) -> Device { - Device::linux(&self.hostname) + match &self.origin { + EventOrigin::Supervisor => Device::linux(&self.hostname), + EventOrigin::Gateway { name } => Device::gateway(&self.hostname, name), + } } /// Build the `proxy_endpoint` object for the Network Proxy profile. @@ -268,6 +290,7 @@ pub(crate) fn test_sandbox_context() -> EventContext { product_version: "0.1.0".to_string(), proxy_ip: "10.42.0.1".parse().unwrap(), proxy_port: 3128, + origin: EventOrigin::Supervisor, } } diff --git a/crates/openshell-ocsf/src/ctx.rs b/crates/openshell-ocsf/src/ctx.rs index f714b77ab2..7e1c4c7796 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::EventContext; +use crate::{EventContext, EventOrigin}; use std::sync::{LazyLock, OnceLock}; static OCSF_CTX: OnceLock = OnceLock::new(); @@ -21,6 +21,7 @@ static OCSF_CTX_FALLBACK: LazyLock = LazyLock::new(|| EventContext product_version: env!("CARGO_PKG_VERSION").to_string(), proxy_ip: std::net::IpAddr::from([127, 0, 0, 1]), proxy_port: 3128, + origin: EventOrigin::Supervisor, }); /// 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 index b47596e9d0..369189390f 100644 --- a/crates/openshell-ocsf/src/enums/device_type.rs +++ b/crates/openshell-ocsf/src/enums/device_type.rs @@ -15,6 +15,8 @@ use serde_repr::{Deserialize_repr, Serialize_repr}; pub enum DeviceTypeId { /// 0 — Unknown Unknown = 0, + /// 1 — Server + Server = 1, /// 99 — Other Other = 99, } @@ -30,6 +32,7 @@ impl std::fmt::Display for DeviceTypeId { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str(match self { Self::Unknown => "Unknown", + Self::Server => "Server", Self::Other => "Other", }) } @@ -43,6 +46,7 @@ mod tests { fn device_type_display_uses_schema_labels() { for (device_type, expected) in [ (DeviceTypeId::Unknown, "Unknown"), + (DeviceTypeId::Server, "Server"), (DeviceTypeId::Other, "Other"), ] { assert_eq!(device_type.to_string(), expected); @@ -52,7 +56,11 @@ mod tests { #[test] fn device_type_json_roundtrip() { - for (device_type, expected) in [(DeviceTypeId::Unknown, 0), (DeviceTypeId::Other, 99)] { + for (device_type, expected) in [ + (DeviceTypeId::Unknown, 0), + (DeviceTypeId::Server, 1), + (DeviceTypeId::Other, 99), + ] { let json = serde_json::to_value(device_type).unwrap(); assert_eq!(json, serde_json::json!(expected)); let decoded: DeviceTypeId = serde_json::from_value(json).unwrap(); diff --git a/crates/openshell-ocsf/src/lib.rs b/crates/openshell-ocsf/src/lib.rs index e6be8c2644..317a891f39 100644 --- a/crates/openshell-ocsf/src/lib.rs +++ b/crates/openshell-ocsf/src/lib.rs @@ -58,8 +58,8 @@ pub use objects::{ // --- Builders --- pub use builders::{ ApiActivityBuilder, AppLifecycleBuilder, BaseEventBuilder, ConfigStateChangeBuilder, - DetectionFindingBuilder, EventContext, HttpActivityBuilder, NetworkActivityBuilder, - ProcessActivityBuilder, SshActivityBuilder, + DetectionFindingBuilder, EventContext, EventOrigin, HttpActivityBuilder, + NetworkActivityBuilder, ProcessActivityBuilder, SshActivityBuilder, }; // --- Tracing layers --- diff --git a/crates/openshell-ocsf/src/objects/device.rs b/crates/openshell-ocsf/src/objects/device.rs index b373a958c9..04cc6386c6 100644 --- a/crates/openshell-ocsf/src/objects/device.rs +++ b/crates/openshell-ocsf/src/objects/device.rs @@ -55,6 +55,28 @@ impl Device { }), } } + + /// Create the device for a gateway replica. + #[must_use] + pub fn gateway(hostname: &str, name: &str) -> Self { + Self { + hostname: hostname.to_string(), + type_id: DeviceTypeId::Server, + type_label: DeviceTypeId::Server.to_string(), + name: Some(name.to_string()), + // Keep the replica identity opaque rather than encoding multiple fields in the UID. + uid: Some(hostname.to_string()), + os: Some(OsInfo { + name: match std::env::consts::OS { + "linux" => "Linux", + "windows" => "Windows", + "macos" => "macOS", + other => other, + } + .to_string(), + }), + } + } } #[cfg(test)] @@ -85,4 +107,23 @@ mod tests { assert_eq!(decoded, device); assert_eq!(serde_json::to_value(&decoded).unwrap(), json); } + + #[test] + fn gateway_device_does_not_inherit_the_sandbox_type() { + let json = serde_json::to_value(Device::gateway("gateway-0", "production")).unwrap(); + + assert_eq!(json["type_id"], 1); + assert_eq!(json["type"], "Server"); + } + + #[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 14f5f8af61..c580d0d840 100644 --- a/crates/openshell-ocsf/src/objects/metadata.rs +++ b/crates/openshell-ocsf/src/objects/metadata.rs @@ -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 index 9fa51cc8bd..00aca3a492 100644 --- a/crates/openshell-ocsf/tests/event_identity.rs +++ b/crates/openshell-ocsf/tests/event_identity.rs @@ -5,7 +5,9 @@ use std::net::{IpAddr, Ipv4Addr}; -use openshell_ocsf::{ActivityId, EventContext, NetworkActivityBuilder, OcsfEvent, SeverityId}; +use openshell_ocsf::{ + ActivityId, EventContext, EventOrigin, NetworkActivityBuilder, OcsfEvent, SeverityId, +}; fn sandbox_ctx(container_image: &str) -> EventContext { EventContext { @@ -16,6 +18,22 @@ fn sandbox_ctx(container_image: &str) -> EventContext { product_version: "0.42.1".to_string(), proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), proxy_port: 8888, + origin: EventOrigin::Supervisor, + } +} + +fn gateway_ctx(sandbox_id: &str, sandbox_name: &str) -> EventContext { + EventContext { + 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(), + }, } } @@ -56,6 +74,27 @@ fn the_sandbox_id_is_carried_by_the_container() { 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(&sandbox_ctx("")).to_json().expect("serializes"); diff --git a/crates/openshell-ocsf/tests/gateway_context.rs b/crates/openshell-ocsf/tests/gateway_context.rs new file mode 100644 index 0000000000..5117cfe5f8 --- /dev/null +++ b/crates/openshell-ocsf/tests/gateway_context.rs @@ -0,0 +1,146 @@ +// 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, EventContext, EventOrigin, + SeverityId, StateId, StatusId, +}; + +fn gateway_ctx() -> EventContext { + EventContext { + 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() -> EventContext { + EventContext { + 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::Supervisor, + } +} + +#[test] +fn gateway_events_report_the_gateway_os() { + let expected_os = match std::env::consts::OS { + "linux" => "Linux", + "windows" => "Windows", + "macos" => "macOS", + other => other, + }; + + for sandbox_id in ["", "sb-1"] { + let mut context = gateway_ctx(); + context.sandbox_id = sandbox_id.to_string(); + let json = AppLifecycleBuilder::new(&context) + .activity(ActivityId::Open) + .build() + .to_json() + .unwrap(); + + assert_eq!(json["device"]["os"]["name"], expected_os); + } +} + +#[test] +fn linux_supervisor_events_report_linux_os() { + let json = AppLifecycleBuilder::new(&sandbox_ctx()) + .activity(ActivityId::Open) + .build() + .to_json() + .unwrap(); + + assert_eq!(json["device"]["os"]["name"], "Linux"); +} + +#[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 42664400ab..b0ed5b3ccc 100644 --- a/crates/openshell-ocsf/tests/roundtrip.rs +++ b/crates/openshell-ocsf/tests/roundtrip.rs @@ -12,7 +12,7 @@ use std::net::{IpAddr, Ipv4Addr}; use openshell_ocsf::{ ActionId, ActivityId, AiModel, ApiActivityBuilder, AppLifecycleBuilder, Attack, AuthTypeId, BaseEventBuilder, ConfidenceId, ConfigStateChangeBuilder, ConnectionInfo, - DetectionFindingBuilder, DispositionId, Endpoint, EventContext, FindingInfo, + DetectionFindingBuilder, DispositionId, Endpoint, EventContext, EventOrigin, FindingInfo, HttpActivityBuilder, HttpMethod, HttpRequest, HttpResponse, LaunchTypeId, NetworkActivityBuilder, OcsfEvent, Process, ProcessActivityBuilder, RiskLevelId, SecurityLevelId, SeverityId, SshActivityBuilder, StateId, StatusId, Url, @@ -27,6 +27,7 @@ fn ctx() -> EventContext { product_version: "0.42.1".to_string(), proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), proxy_port: 8888, + origin: EventOrigin::Supervisor, } } diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index b5b9358ac0..a32bdd630d 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::Supervisor, }) { debug!("OCSF context already initialized, keeping existing"); } diff --git a/crates/openshell-server/Cargo.toml b/crates/openshell-server/Cargo.toml index efff4f36b0..4debd3ed3c 100644 --- a/crates/openshell-server/Cargo.toml +++ b/crates/openshell-server/Cargo.toml @@ -82,6 +82,7 @@ metrics-exporter-prometheus = { workspace = true } base64 = { workspace = true } futures = { workspace = true } bytes = { workspace = true } +chrono = { version = "0.4", default-features = false, features = ["clock"] } pin-project-lite = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/openshell-server/src/cli.rs b/crates/openshell-server/src/cli.rs index b8b7dfd071..d757cac6ff 100644 --- a/crates/openshell-server/src/cli.rs +++ b/crates/openshell-server/src/cli.rs @@ -500,7 +500,23 @@ 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 ocsf_log = prepared + .config_file + .as_ref() + .and_then(|file| file.openshell.gateway.ocsf_log.clone()) + .map(crate::ocsf_log::OcsfLog::start) + .transpose() + .into_diagnostic()?; let otlp_config = prepared .config_file .as_ref() @@ -514,9 +530,11 @@ async fn run_from_args( &prepared.config.compute_driver_endpoints, ); let (tracing_handle, setup_error) = crate::tracing_setup::install( - EnvFilter::try_from_default_env() - .unwrap_or_else(|_| EnvFilter::new(&prepared.config.log_level)), + &EnvFilter::try_from_default_env() + .unwrap_or_else(|_| EnvFilter::new(&prepared.config.log_level)) + .to_string(), &tracing_log_bus, + ocsf_log.as_ref(), otlp_config, compute_driver_tracing, gateway_resource, @@ -580,6 +598,10 @@ async fn run_from_args( tracing_handle.shutdown(); + if let Some(log) = ocsf_log { + log.shutdown().await; + } + result.into_diagnostic() } diff --git a/crates/openshell-server/src/config_file.rs b/crates/openshell-server/src/config_file.rs index 64ac953624..aaaea7371d 100644 --- a/crates/openshell-server/src/config_file.rs +++ b/crates/openshell-server/src/config_file.rs @@ -183,6 +183,8 @@ pub struct GatewayFileSection { pub gateway_jwt: Option, #[serde(default)] pub otlp: Option, + #[serde(default)] + pub ocsf_log: Option, // ── Disallowed-in-file fields ──────────────────────────────────────── // @@ -193,6 +195,63 @@ pub struct GatewayFileSection { pub database_url: Option, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OcsfLogRotation { + Never, + #[default] + Daily, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(try_from = "RawOcsfLogConfig")] +pub struct OcsfLogConfig { + pub path: PathBuf, + pub rotation: OcsfLogRotation, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_files: Option, + pub queue_capacity: std::num::NonZeroUsize, + pub queue_max_bytes: std::num::NonZeroUsize, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawOcsfLogConfig { + path: PathBuf, + #[serde(default)] + rotation: OcsfLogRotation, + max_files: Option, + queue_capacity: Option, + queue_max_bytes: Option, +} + +impl TryFrom for OcsfLogConfig { + type Error = &'static str; + + fn try_from(raw: RawOcsfLogConfig) -> Result { + if raw.path.as_os_str().is_empty() { + return Err("ocsf_log.path must not be empty"); + } + if raw.rotation == OcsfLogRotation::Never && raw.max_files.is_some() { + return Err("ocsf_log.max_files requires daily rotation"); + } + Ok(Self { + path: raw.path, + rotation: raw.rotation, + max_files: (raw.rotation == OcsfLogRotation::Daily).then(|| { + raw.max_files + .unwrap_or(std::num::NonZeroUsize::new(7).unwrap()) + }), + queue_capacity: raw + .queue_capacity + .unwrap_or(std::num::NonZeroUsize::new(10_000).unwrap()), + queue_max_bytes: raw + .queue_max_bytes + .unwrap_or(std::num::NonZeroUsize::new(16 * 1024 * 1024).unwrap()), + }) + } +} + /// `[openshell.gateway.otlp]` section. /// /// Presence of this table enables OTLP export; there is no `enabled` flag. @@ -604,6 +663,43 @@ service_name = "openshell-gateway-dev" assert_eq!(otlp.service_name.as_deref(), Some("openshell-gateway-dev")); } + #[test] + fn gateway_accepts_a_single_ocsf_log_destination() { + let tmp = write_tmp("[openshell.gateway.ocsf_log]\npath = 'events.jsonl'\n"); + let config = load(tmp.path()) + .unwrap() + .openshell + .gateway + .ocsf_log + .unwrap(); + assert_eq!(config.rotation, OcsfLogRotation::Daily); + assert_eq!(config.max_files.unwrap().get(), 7); + assert_eq!(config.queue_capacity.get(), 10_000); + assert_eq!(config.queue_max_bytes.get(), 16 * 1024 * 1024); + assert!(ConfigFile::default().openshell.gateway.ocsf_log.is_none()); + } + + #[test] + fn ocsf_log_rejects_invalid_and_unshipped_options() { + for settings in [ + "", + "path = ''", + "path = 'log'\nqueue_capacity = 0", + "path = 'log'\nqueue_max_bytes = 0", + "path = 'log'\nmax_files = 0", + "path = 'log'\nrotation = 'hourly'", + "path = 'log'\nkind = 'jsonl'", + "path = 'log'\nrotation = 'never'\nmax_files = 7", + ] { + let tmp = write_tmp(&format!("[openshell.gateway.ocsf_log]\n{settings}\n")); + assert!(load(tmp.path()).is_err(), "accepted {settings}"); + } + let tmp = write_tmp("[openshell.gateway.ocsf_log]\npath = 'log'\nrotation = 'never'\n"); + let config = load(tmp.path()).unwrap(); + let encoded = toml::to_string(&config).unwrap(); + assert!(toml::from_str::(&encoded).is_ok()); + } + #[test] fn otlp_config_requires_only_endpoint() { let toml = r#" diff --git a/crates/openshell-server/src/gateway_ocsf.rs b/crates/openshell-server/src/gateway_ocsf.rs new file mode 100644 index 0000000000..abc823f415 --- /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::{EventContext, EventOrigin}; + +/// 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) -> EventContext { + let identity = identity(); + EventContext { + 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 62bc0f8575..cf52300dbc 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -54,14 +54,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, EventContext, OCSF_TARGET, OcsfEvent, 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, @@ -78,7 +75,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}; @@ -171,7 +168,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, @@ -180,11 +177,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 @@ -209,7 +202,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", @@ -218,14 +211,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, @@ -233,16 +222,8 @@ fn build_gateway_policy_audit_message( version: i64, policy_hash: &str, extra_fields: &[(&str, String)], -) -> String { - let ctx = EventContext { - 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, - }; +) -> OcsfEvent { + let ctx = crate::gateway_ocsf::context(sandbox_id, sandbox_name); let mut builder = ConfigStateChangeBuilder::new(&ctx) .state(StateId::Other, state_label) .severity(SeverityId::Informational) @@ -257,8 +238,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 { @@ -17074,16 +17054,35 @@ mod tests { } #[test] - fn build_gateway_policy_audit_message_formats_ocsf_config_line() { - let message = build_gateway_policy_audit_message( + fn policy_audit_identifies_the_gateway_as_the_producer() { + let event = build_gateway_policy_audit_event( "sb-123", "demo-sandbox", "merged", - "gateway merged incremental policy op: add-allow api.github.com:443 [POST /repos/*/issues]", + "updated policy", 7, "sha256:testhash", &[], ); + assert_eq!(event.base().metadata.product.name, "OpenShell Gateway"); + assert_eq!( + event.base().container.as_ref().unwrap().uid.as_deref(), + Some("sb-123") + ); + } + + #[test] + fn build_gateway_policy_audit_event_formats_ocsf_config_line() { + let message = build_gateway_policy_audit_event( + "sb-123", + "demo-sandbox", + "merged", + "gateway merged incremental policy op: add-allow api.github.com:443 [POST /repos/*/issues]", + 7, + "sha256:testhash", + &[], + ) + .format_shorthand(); assert_eq!( message, @@ -17098,13 +17097,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", @@ -17112,7 +17111,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/lib.rs b/crates/openshell-server/src/lib.rs index 50c4f98696..cf3fd08946 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -21,10 +21,12 @@ pub mod config_file; mod credentials; mod defaults; mod gateway_listener; +mod gateway_ocsf; mod grpc; mod http; mod middleware; mod multiplex; +mod ocsf_log; mod otel_tracing; mod persistence; pub(crate) mod policy_store; diff --git a/crates/openshell-server/src/ocsf_log.rs b/crates/openshell-server/src/ocsf_log.rs new file mode 100644 index 0000000000..07869fcd96 --- /dev/null +++ b/crates/openshell-server/src/ocsf_log.rs @@ -0,0 +1,672 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::collections::{BTreeMap, VecDeque}; +use std::fs::{File, OpenOptions}; +use std::io::{self, Read, Seek, SeekFrom, Write}; +use std::path::Path; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::{Duration, Instant}; + +use chrono::{DateTime, NaiveDate, Utc}; +use openshell_ocsf::OcsfEvent; +use tracing::{Event, Subscriber}; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; + +use crate::config_file::{OcsfLogConfig, OcsfLogRotation}; + +const BATCH_SIZE: usize = 100; +const FLUSH_INTERVAL: Duration = Duration::from_millis(500); +const SHUTDOWN_BUDGET: Duration = Duration::from_secs(5); + +#[derive(Default)] +struct State { + queue: VecDeque>, + queued_bytes: usize, + in_flight: usize, + closed: bool, + abandoned: bool, + written: u64, + losses: BTreeMap<&'static str, u64>, +} + +struct Shared { + state: Mutex, + ready: Condvar, + config: OcsfLogConfig, +} + +#[derive(Clone)] +pub struct Collector(Arc); + +impl Collector { + fn new(config: OcsfLogConfig) -> Self { + Self(Arc::new(Shared { + state: Mutex::new(State::default()), + ready: Condvar::new(), + config, + })) + } + + pub(crate) fn collect(&self, event: &OcsfEvent) { + match event.to_json_line() { + Ok(line) => { + self.enqueue(line.into_bytes()); + } + Err(_) => self.record_loss("serialization", 1), + } + } + + fn enqueue(&self, line: Vec) -> bool { + let mut state = self.0.state.lock().unwrap(); + let reason = if state.closed { + Some("closed") + } else if line.len() > self.0.config.queue_max_bytes.get() { + Some("oversized") + } else if state.queue.len() >= self.0.config.queue_capacity.get() { + Some("queue_capacity") + } else if line.len() > self.0.config.queue_max_bytes.get() - state.queued_bytes { + Some("queue_bytes") + } else { + None + }; + if let Some(reason) = reason { + Self::loss(&mut state, reason, 1); + return false; + } + state.queued_bytes += line.len(); + state.queue.push_back(line); + Self::queue_metrics(&state); + metrics::counter!("openshell_ocsf_log_queued_total").increment(1); + self.0.ready.notify_one(); + true + } + + #[allow(clippy::cast_precision_loss)] + fn queue_metrics(state: &State) { + metrics::gauge!("openshell_ocsf_log_queue_records").set(state.queue.len() as f64); + metrics::gauge!("openshell_ocsf_log_queue_bytes").set(state.queued_bytes as f64); + } + + fn loss(state: &mut State, reason: &'static str, count: u64) { + if count > 0 { + *state.losses.entry(reason).or_default() += count; + metrics::counter!("openshell_ocsf_log_dropped_total", "reason" => reason) + .increment(count); + } + } + + pub(crate) fn record_loss(&self, reason: &'static str, count: u64) { + Self::loss(&mut self.0.state.lock().unwrap(), reason, count); + } + + fn close(&self) { + self.0.state.lock().unwrap().closed = true; + self.0.ready.notify_all(); + } + + fn abandon(&self) { + let mut state = self.0.state.lock().unwrap(); + state.abandoned = true; + let queued = state.queue.len() as u64; + let uncertain = state.in_flight as u64; + state.queue.clear(); + state.queued_bytes = 0; + state.in_flight = 0; + Self::loss(&mut state, "shutdown", queued); + Self::loss(&mut state, "shutdown_uncertain", uncertain); + Self::queue_metrics(&state); + self.0.ready.notify_all(); + } + + fn batch(&self) -> Option>> { + let mut state = self.0.state.lock().unwrap(); + while state.queue.is_empty() && !state.closed { + state = self.0.ready.wait(state).unwrap(); + } + let deadline = Instant::now() + FLUSH_INTERVAL; + while !state.closed && state.queue.len() < BATCH_SIZE && Instant::now() < deadline { + state = self + .0 + .ready + .wait_timeout(state, deadline.saturating_duration_since(Instant::now())) + .unwrap() + .0; + } + if state.queue.is_empty() || state.abandoned { + return None; + } + let count = BATCH_SIZE.min(state.queue.len()); + let batch: Vec<_> = state.queue.drain(..count).collect(); + state.queued_bytes -= batch.iter().map(Vec::len).sum::(); + state.in_flight = count; + Self::queue_metrics(&state); + Some(batch) + } + + fn complete(&self, failure: Option<&'static str>) { + let mut state = self.0.state.lock().unwrap(); + if state.abandoned { + return; + } + state.in_flight -= 1; + if let Some(reason) = failure { + Self::loss(&mut state, reason, 1); + } else { + state.written += 1; + metrics::counter!("openshell_ocsf_log_written_total").increment(1); + } + } +} + +pub struct OcsfLog { + collector: Collector, + done: tokio::sync::oneshot::Receiver<()>, +} + +impl OcsfLog { + pub(crate) fn start(config: OcsfLogConfig) -> io::Result { + let collector = Collector::new(config); + let worker = collector.clone(); + let (done_tx, done) = tokio::sync::oneshot::channel(); + std::thread::Builder::new() + .name("ocsf-jsonl".into()) + .spawn(move || { + run_writer(&worker); + let _ = done_tx.send(()); + })?; + Ok(Self { collector, done }) + } + + pub(crate) fn collector(&self) -> Collector { + self.collector.clone() + } + + pub(crate) fn layer(&self) -> impl Layer + use { + CaptureLayer(self.collector()) + } + + pub(crate) async fn shutdown(mut self) { + self.shutdown_with_budget(SHUTDOWN_BUDGET).await; + } + + async fn shutdown_with_budget(&mut self, budget: Duration) { + self.collector.close(); + if !matches!( + tokio::time::timeout(budget, &mut self.done).await, + Ok(Ok(())) + ) { + self.collector.abandon(); + tracing::warn!( + "OCSF JSONL shutdown did not finish; queued records lost and outstanding writes uncertain" + ); + } + } +} + +impl Drop for OcsfLog { + fn drop(&mut self) { + self.collector.close(); + } +} + +struct CaptureLayer(Collector); + +impl Layer for CaptureLayer { + fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) { + if event.metadata().target() == openshell_ocsf::OCSF_TARGET + && let Some(event) = openshell_ocsf::clone_current_event() + { + self.0.collect(&event); + } + } +} + +fn run_writer(collector: &Collector) { + let mut writer = FileWriter::new(collector.0.config.clone()); + let mut retry_at = Instant::now(); + let mut backoff = Duration::from_millis(500); + while let Some(batch) = collector.batch() { + for line in batch { + if collector.0.state.lock().unwrap().abandoned { + return; + } + if Instant::now() < retry_at { + collector.complete(Some("unavailable")); + continue; + } + match writer.append(&line, Utc::now().date_naive()) { + Ok(()) => { + backoff = Duration::from_millis(500); + collector.complete(None); + } + Err(error) => { + writer.file = None; + metrics::counter!("openshell_ocsf_log_writer_errors_total").increment(1); + tracing::warn!(%error, "OCSF JSONL write failed; record may be incomplete, later records discarded until reopen"); + collector.complete(Some("write_uncertain")); + retry_at = Instant::now() + backoff; + backoff = (backoff * 2).min(Duration::from_secs(30)); + } + } + } + } +} + +struct FileWriter { + config: OcsfLogConfig, + file: Option, + day: Option, +} + +impl FileWriter { + fn new(config: OcsfLogConfig) -> Self { + Self { + config, + file: None, + day: None, + } + } + + fn open(&mut self) -> io::Result<()> { + if let Some(parent) = self + .config + .path + .parent() + .filter(|path| !path.as_os_str().is_empty()) + { + std::fs::create_dir_all(parent)?; + } + if let Ok(metadata) = std::fs::metadata(&self.config.path) + && !metadata.is_file() + { + return Err(io::Error::other( + "OCSF log destination must be a regular file", + )); + } + let mut options = OpenOptions::new(); + options.create(true).read(true).write(true).truncate(false); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options.open(&self.config.path)?; + self.day = Some(DateTime::::from(file.metadata()?.modified()?).date_naive()); + let discarded = recover_tail(&mut file)?; + if discarded > 0 { + metrics::counter!("openshell_ocsf_log_recovery_discarded_bytes_total") + .increment(discarded); + tracing::warn!( + discarded_bytes = discarded, + "OCSF JSONL incomplete tail removed; number of lost records unknown" + ); + } + self.file = Some(file); + Ok(()) + } + + fn append(&mut self, line: &[u8], today: NaiveDate) -> io::Result<()> { + if self.file.is_none() { + self.open()?; + } + if self.config.rotation == OcsfLogRotation::Daily && self.day != Some(today) { + self.file = None; + rotate(&self.config.path, self.day.unwrap())?; + prune_rotated(&self.config.path, self.config.max_files.unwrap().get())?; + self.open()?; + } + self.day = Some(today); + append_record(self.file.as_mut().unwrap(), line) + } +} + +trait RecordFile: Write + Seek { + fn truncate(&mut self, length: u64) -> io::Result<()>; +} + +impl RecordFile for File { + fn truncate(&mut self, length: u64) -> io::Result<()> { + self.set_len(length) + } +} + +fn append_record(file: &mut impl RecordFile, line: &[u8]) -> io::Result<()> { + let boundary = file.stream_position()?; + if let Err(error) = file.write_all(line).and_then(|()| file.flush()) { + file.truncate(boundary)?; + file.seek(SeekFrom::Start(boundary))?; + return Err(error); + } + Ok(()) +} + +fn recover_tail(file: &mut File) -> io::Result { + let length = file.metadata()?.len(); + let mut end = length; + let mut buffer = [0u8; 8192]; + while end > 0 { + let start = end.saturating_sub(buffer.len() as u64); + let count = usize::try_from(end - start).unwrap(); + file.seek(SeekFrom::Start(start))?; + file.read_exact(&mut buffer[..count])?; + if let Some(index) = buffer[..count].iter().rposition(|byte| *byte == b'\n') { + let boundary = start + index as u64 + 1; + file.set_len(boundary)?; + file.seek(SeekFrom::Start(boundary))?; + return Ok(length - boundary); + } + end = start; + } + file.set_len(0)?; + file.seek(SeekFrom::Start(0))?; + Ok(length) +} + +fn rotate(path: &Path, day: NaiveDate) -> io::Result<()> { + loop { + let mut name = path.as_os_str().to_os_string(); + name.push(format!(".{day}.{}", uuid::Uuid::new_v4())); + let archive = std::path::PathBuf::from(name); + match OpenOptions::new() + .write(true) + .create_new(true) + .open(&archive) + { + Ok(reservation) => { + drop(reservation); + if let Err(error) = std::fs::rename(path, &archive) { + let _ = std::fs::remove_file(&archive); + return Err(error); + } + return Ok(()); + } + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {} + Err(error) => return Err(error), + } + } +} + +fn prune_rotated(path: &Path, max_files: usize) -> io::Result<()> { + let parent = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let Some(stem) = path.file_name().and_then(|name| name.to_str()) else { + return Ok(()); + }; + let prefix = format!("{stem}."); + let mut rotated = Vec::new(); + for entry in std::fs::read_dir(parent)? { + let entry = entry?; + let name = entry.file_name(); + let Some(suffix) = name.to_str().and_then(|name| name.strip_prefix(&prefix)) else { + continue; + }; + let Some((day, unique)) = suffix.split_once('.') else { + continue; + }; + if let Ok(day) = NaiveDate::parse_from_str(day, "%Y-%m-%d") + && uuid::Uuid::parse_str(unique).is_ok() + && entry.file_type()?.is_file() + { + rotated.push((day, entry.metadata()?.modified()?, entry.path())); + } + } + rotated.sort(); + let excess = rotated.len().saturating_sub(max_files); + for (_, _, stale) in rotated.into_iter().take(excess) { + std::fs::remove_file(stale)?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use openshell_ocsf::{ConfigStateChangeBuilder, ocsf_emit}; + use tracing_subscriber::prelude::*; + + fn config(path: &Path) -> OcsfLogConfig { + toml::from_str(&format!( + "path = {:?}\nrotation = 'never'\n", + path.display().to_string() + )) + .unwrap() + } + + #[tokio::test] + async fn gateway_native_events_are_written_with_console_off() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("events.jsonl"); + let log = OcsfLog::start(config(&path)).unwrap(); + let event = ConfigStateChangeBuilder::new(&crate::gateway_ocsf::context("", "")) + .message("Gateway TLS configuration changed") + .build(); + let expected: serde_json::Value = + serde_json::from_str(&event.to_json_line().unwrap()).unwrap(); + let subscriber = tracing_subscriber::registry() + .with( + tracing_subscriber::fmt::layer() + .with_filter(tracing_subscriber::EnvFilter::new("off")), + ) + .with(log.layer()); + tracing::subscriber::with_default(subscriber, || { + tracing::info!("ordinary diagnostics must not enter the file"); + ocsf_emit!(event); + }); + log.shutdown().await; + let contents = std::fs::read_to_string(path).unwrap(); + assert_eq!(contents.lines().count(), 1); + assert_eq!( + serde_json::from_str::(&contents).unwrap(), + expected + ); + } + + #[tokio::test] + async fn gateway_certificate_reload_reaches_jsonl_without_a_sandbox() { + let directory = tempfile::tempdir().unwrap(); + crate::tls_test_utils::generate_test_certs_with_ca(directory.path()); + let acceptor = crate::tls::TlsAcceptor::from_files( + &directory.path().join("server-cert.pem"), + &directory.path().join("server-key.pem"), + None, + false, + None, + None, + Vec::new(), + ) + .unwrap(); + let path = directory.path().join("events.jsonl"); + let log = OcsfLog::start(config(&path)).unwrap(); + let subscriber = tracing_subscriber::registry().with(log.layer()); + tracing::subscriber::with_default(subscriber, || acceptor.reload().unwrap()); + log.shutdown().await; + let contents = std::fs::read_to_string(path).unwrap(); + let event: serde_json::Value = serde_json::from_str(&contents).unwrap(); + assert_eq!( + event["message"], + "TLS certificate config reloaded successfully" + ); + assert_eq!(event["metadata"]["product"]["name"], "OpenShell Gateway"); + assert!(event.get("container").is_none()); + } + + #[test] + fn queue_limits_drop_incoming_records_without_evicting_history() { + let mut config = config(Path::new("unused")); + config.queue_capacity = std::num::NonZeroUsize::new(2).unwrap(); + config.queue_max_bytes = std::num::NonZeroUsize::new(8).unwrap(); + let collector = Collector::new(config); + assert!(collector.enqueue(b"1234\n".to_vec())); + assert!(!collector.enqueue(b"5678\n".to_vec())); + assert!(!collector.enqueue(b"oversized\n".to_vec())); + assert!(collector.enqueue(b"0\n".to_vec())); + assert!(!collector.enqueue(b"\n".to_vec())); + collector.close(); + assert!(!collector.enqueue(b"\n".to_vec())); + let state = collector.0.state.lock().unwrap(); + assert_eq!(state.queued_bytes, 7); + assert_eq!(state.queue.front().unwrap(), b"1234\n"); + for reason in ["queue_bytes", "oversized", "queue_capacity", "closed"] { + assert_eq!(state.losses[reason], 1); + } + } + + #[test] + fn reopening_removes_only_the_incomplete_tail() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("events.jsonl"); + for prefix in [Vec::new(), b"{}\n".to_vec()] { + let mut damaged = prefix.clone(); + damaged.extend(vec![b'x'; 20_000]); + std::fs::write(&path, &damaged).unwrap(); + let mut file = OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .unwrap(); + assert_eq!(recover_tail(&mut file).unwrap(), 20_000); + append_record(&mut file, b"{\"new\":true}\n").unwrap(); + let mut expected = prefix; + expected.extend(b"{\"new\":true}\n"); + assert_eq!(std::fs::read(&path).unwrap(), expected); + } + } + + struct FailingFile { + bytes: io::Cursor>, + remaining: usize, + fail_truncate: bool, + fail_flush: bool, + } + + impl Write for FailingFile { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if self.remaining == 0 { + return Err(io::Error::other("disk full")); + } + let count = bytes.len().min(self.remaining); + self.remaining -= count; + self.bytes.write(&bytes[..count]) + } + fn flush(&mut self) -> io::Result<()> { + if self.fail_flush { + Err(io::Error::other("flush failed")) + } else { + Ok(()) + } + } + } + + impl Seek for FailingFile { + fn seek(&mut self, position: SeekFrom) -> io::Result { + self.bytes.seek(position) + } + } + + impl RecordFile for FailingFile { + fn truncate(&mut self, length: u64) -> io::Result<()> { + if self.fail_truncate { + return Err(io::Error::other("truncate failed")); + } + self.bytes + .get_mut() + .truncate(usize::try_from(length).unwrap()); + Ok(()) + } + } + + #[test] + fn partial_writes_do_not_replay_completed_records() { + let mut file = FailingFile { + bytes: io::Cursor::new(Vec::new()), + remaining: 6, + fail_truncate: false, + fail_flush: false, + }; + append_record(&mut file, b"{}\n").unwrap(); + assert!(append_record(&mut file, b"{\"partial\":true}\n").is_err()); + assert_eq!(file.bytes.get_ref(), b"{}\n"); + file.remaining = 100; + append_record(&mut file, b"{\"later\":true}\n").unwrap(); + assert_eq!(file.bytes.get_ref(), b"{}\n{\"later\":true}\n"); + file.fail_flush = true; + assert!(append_record(&mut file, b"{}\n").is_err()); + assert_eq!(file.bytes.get_ref(), b"{}\n{\"later\":true}\n"); + file.fail_truncate = true; + assert!(append_record(&mut file, b"{}\n").is_err()); + } + + #[test] + fn same_day_archives_never_overwrite_and_pruning_ignores_unrelated_files() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("events.jsonl"); + let today = Utc::now().date_naive(); + for contents in ["first\n", "second\n", "third\n"] { + std::fs::write(&path, contents).unwrap(); + rotate(&path, today).unwrap(); + } + let mut contents: Vec<_> = std::fs::read_dir(directory.path()) + .unwrap() + .map(|entry| std::fs::read_to_string(entry.unwrap().path()).unwrap()) + .collect(); + contents.sort(); + assert_eq!(contents, ["first\n", "second\n", "third\n"]); + let unrelated = directory.path().join("events.jsonl.notes"); + std::fs::write(&unrelated, "keep").unwrap(); + prune_rotated(&path, 1).unwrap(); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 2); + assert_eq!(std::fs::read_to_string(unrelated).unwrap(), "keep"); + } + + #[test] + fn daily_rotation_keeps_previous_day_records_out_of_the_active_file() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("events.jsonl"); + let mut settings = config(&path); + settings.rotation = OcsfLogRotation::Daily; + settings.max_files = std::num::NonZeroUsize::new(1); + let mut writer = FileWriter::new(settings); + let today = Utc::now().date_naive(); + writer.append(b"{}\n", today).unwrap(); + writer + .append(b"{\"next\":true}\n", today.succ_opt().unwrap()) + .unwrap(); + assert_eq!(std::fs::read_to_string(&path).unwrap(), "{\"next\":true}\n"); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 2); + } + + #[tokio::test] + async fn invalid_destination_is_visible_outside_the_output_file() { + let directory = tempfile::tempdir().unwrap(); + let log = OcsfLog::start(config(directory.path())).unwrap(); + let collector = log.collector(); + collector.enqueue(b"{}\n".to_vec()); + log.shutdown().await; + let state = collector.0.state.lock().unwrap(); + assert_eq!(state.written, 0); + assert_eq!(state.losses["write_uncertain"], 1); + } + + #[tokio::test] + async fn shutdown_budget_accounts_queued_and_uncertain_records() { + let collector = Collector::new(config(Path::new("unused"))); + collector.enqueue(b"{}\n".to_vec()); + collector.0.state.lock().unwrap().in_flight = 2; + let (_sender, done) = tokio::sync::oneshot::channel(); + let mut log = OcsfLog { + collector: collector.clone(), + done, + }; + log.shutdown_with_budget(Duration::from_millis(1)).await; + collector.complete(None); + let state = collector.0.state.lock().unwrap(); + assert!(state.queue.is_empty()); + assert_eq!(state.losses["shutdown"], 1); + assert_eq!(state.losses["shutdown_uncertain"], 2); + assert_eq!(state.written, 0); + } +} diff --git a/crates/openshell-server/src/service_routing.rs b/crates/openshell-server/src/service_routing.rs index 926f78094c..bc60b47939 100644 --- a/crates/openshell-server/src/service_routing.rs +++ b/crates/openshell-server/src/service_routing.rs @@ -9,19 +9,18 @@ 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, EventContext, HttpActivityBuilder, HttpRequest, HttpResponse as OcsfHttpResponse, NetworkActivityBuilder, - OCSF_TARGET, OcsfEvent, SeverityId, StateId, StatusId, Url as OcsfUrl, + OcsfEvent, 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}; @@ -582,12 +581,12 @@ 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); + openshell_ocsf::ocsf_emit!(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); + openshell_ocsf::ocsf_emit!(event); } pub fn emit_cross_origin_service_http_rejection(state: &ServerState, req: &Request) { @@ -623,13 +622,12 @@ fn emit_service_http_failure( endpoint, err, ); - let sandbox_id = endpoint.map_or("", |endpoint| endpoint.sandbox_id.as_str()); - emit_gateway_ocsf_event(sandbox_id, event); + openshell_ocsf::ocsf_emit!(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); + openshell_ocsf::ocsf_emit!(event); } fn build_service_endpoint_config_event( @@ -751,25 +749,8 @@ 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 - ); -} - fn gateway_ocsf_ctx(sandbox_id: &str, sandbox_name: &str) -> EventContext { - EventContext { - 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 { @@ -1234,4 +1215,48 @@ mod tests { "should not find endpoint in wrong workspace" ); } + + /// 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 9154034dfd..6613223f4a 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, EventContext, OCSF_TARGET, SeverityId, StateId, StatusId, -}; +use openshell_ocsf::{ConfigStateChangeBuilder, EventContext, SeverityId, StateId, StatusId}; use rustls::ServerConfig; use rustls::crypto::aws_lc_rs::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; @@ -474,15 +464,7 @@ fn load_key(path: &Path) -> Result> { /// Build an OCSF context for gateway-level (non-sandbox) events. fn tls_ocsf_ctx() -> EventContext { - EventContext { - 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 a91a5fd877..3e61c4f4ac 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -134,6 +134,20 @@ where let mut visitor = LogVisitor::default(); event.record(&mut visitor); + if meta.target() == OCSF_TARGET + && let Some(ocsf) = openshell_ocsf::clone_current_event() + { + // The structured bridge carries no tracing fields. Route by the + // affected sandbox, not by the gateway that produced the event. + visitor.sandbox_id = ocsf + .base() + .container + .as_ref() + .and_then(|container| container.uid.clone()) + .filter(|id| !id.is_empty()); + visitor.message = Some(ocsf.format_shorthand()); + } + let Some(sandbox_id) = visitor.sandbox_id else { return; }; @@ -196,6 +210,44 @@ fn display_level(target: &str, level: &str) -> String { mod tests { use super::*; + #[test] + fn gateway_ocsf_reaches_sandbox_tail_and_live_stream() { + use tracing_subscriber::prelude::*; + let bus = TracingLogBus::new(); + let mut receiver = bus.subscribe("sb-audit"); + let event = openshell_ocsf::ConfigStateChangeBuilder::new(&crate::gateway_ocsf::context( + "sb-audit", "audit", + )) + .message("policy approved") + .build(); + let expected = event.format_shorthand(); + let subscriber = tracing_subscriber::registry().with(bus.layer()); + tracing::subscriber::with_default(subscriber, || { + openshell_ocsf::ocsf_emit!(event); + openshell_ocsf::ocsf_emit!( + openshell_ocsf::ConfigStateChangeBuilder::new(&crate::gateway_ocsf::context( + "", "" + ),) + .message("gateway-wide event") + .build() + ); + }); + let live = receiver + .try_recv() + .expect("sandbox audit event must reach live stream"); + let Some(openshell_core::proto::sandbox_stream_event::Payload::Log(log)) = live.payload + else { + panic!("expected log payload"); + }; + assert_eq!(log.sandbox_id, "sb-audit"); + assert_eq!(log.source, "gateway"); + assert_eq!(log.level, "OCSF"); + assert_eq!(log.message, expected); + assert_eq!(bus.tail("sb-audit", 10).len(), 1); + assert!(bus.tail("", 10).is_empty()); + assert!(receiver.try_recv().is_err()); + } + fn make_log_event(sandbox_id: &str, message: &str) -> SandboxLogLine { SandboxLogLine { sandbox_id: sandbox_id.to_string(), diff --git a/crates/openshell-server/src/tracing_setup.rs b/crates/openshell-server/src/tracing_setup.rs index ed3f56c6e2..4086238e76 100644 --- a/crates/openshell-server/src/tracing_setup.rs +++ b/crates/openshell-server/src/tracing_setup.rs @@ -9,6 +9,7 @@ use opentelemetry_sdk::trace::SdkTracerProvider; use tracing_subscriber::EnvFilter; +use tracing_subscriber::filter::{FilterExt, filter_fn}; use tracing_subscriber::prelude::*; use crate::config_file::OtlpConfig; @@ -35,9 +36,41 @@ impl TracingHandle { } } +fn filter_from(directives: &str) -> EnvFilter { + EnvFilter::try_new(directives).unwrap_or_else(|_| EnvFilter::new("info")) +} + +struct GatewayEventFormat; + +impl tracing_subscriber::fmt::FormatEvent for GatewayEventFormat +where + S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, + N: for<'a> tracing_subscriber::fmt::FormatFields<'a> + 'static, +{ + fn format_event( + &self, + context: &tracing_subscriber::fmt::FmtContext<'_, S, N>, + mut writer: tracing_subscriber::fmt::format::Writer<'_>, + event: &tracing::Event<'_>, + ) -> std::fmt::Result { + if event.metadata().target() == openshell_ocsf::OCSF_TARGET + && let Some(ocsf) = openshell_ocsf::clone_current_event() + { + return writeln!( + writer, + "{} OCSF {}", + chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ"), + ocsf.format_shorthand() + ); + } + tracing_subscriber::fmt::format().format_event(context, writer, event) + } +} + pub fn install( - env_filter: EnvFilter, + filter_directives: &str, tracing_log_bus: &TracingLogBus, + ocsf_log: Option<&crate::ocsf_log::OcsfLog>, otlp_config: Option<&OtlpConfig>, driver: Option, gateway: GatewayResourceAttributes<'_>, @@ -61,19 +94,41 @@ pub fn install( ); tracing_subscriber::registry() - .with(env_filter) - .with(tracing_subscriber::fmt::layer()) - .with(tracing_log_bus.layer()) + .with( + ocsf_log + .map(crate::ocsf_log::OcsfLog::layer) + .with_filter(filter_fn(|metadata| { + metadata.target() == openshell_ocsf::OCSF_TARGET + })), + ) + .with( + tracing_subscriber::fmt::layer() + .event_format(GatewayEventFormat) + .with_filter(filter_from(filter_directives)), + ) + .with( + tracing_log_bus + .layer() + .with_filter(filter_from(filter_directives).or(filter_fn(|metadata| { + metadata.target() == openshell_ocsf::OCSF_TARGET + }))), + ) .with( tracer_provider .as_ref() - .map(|provider| crate::otel_tracing::layer(provider, driver)), + .map(|provider| crate::otel_tracing::layer(provider, driver)) + .with_filter(filter_from(filter_directives)), + ) + .with( + driver_tracer_provider + .as_ref() + .map(|provider| { + driver + .expect("a driver provider requires a selected driver") + .in_process_layer(provider) + }) + .with_filter(filter_from(filter_directives)), ) - .with(driver_tracer_provider.as_ref().map(|provider| { - driver - .expect("a driver provider requires a selected driver") - .in_process_layer(provider) - })) .init(); ( @@ -84,3 +139,41 @@ pub fn install( setup_error.or(driver_setup_error), ) } + +#[cfg(test)] +mod tests { + use std::io::{Read, Seek}; + + use super::*; + + #[test] + fn gateway_ocsf_console_preserves_details_without_jsonl() { + let file = tempfile::tempfile().unwrap(); + let reader = file.try_clone().unwrap(); + let subscriber = tracing_subscriber::registry().with( + tracing_subscriber::fmt::layer() + .event_format(GatewayEventFormat) + .with_ansi(false) + .with_writer(std::sync::Arc::new(file)) + .with_filter(filter_from("info")), + ); + let event = + openshell_ocsf::ConfigStateChangeBuilder::new(&crate::gateway_ocsf::context("", "")) + .message("TLS certificate config reloaded successfully") + .build(); + let expected = event.format_shorthand(); + tracing::subscriber::with_default(subscriber, || { + openshell_ocsf::ocsf_emit!(event); + tracing::info!(answer = 42, "ordinary diagnostic"); + }); + let mut reader = reader; + reader.rewind().unwrap(); + let mut output = String::new(); + reader.read_to_string(&mut output).unwrap(); + assert!(output.contains(&expected), "missing OCSF details: {output}"); + assert!(!output.contains("ocsf_event")); + assert!(output.contains("ordinary diagnostic")); + assert!(output.contains("answer=42")); + assert_eq!(output.lines().count(), 2); + } +} diff --git a/crates/openshell-supervisor-network/src/l7/middleware.rs b/crates/openshell-supervisor-network/src/l7/middleware.rs index f2df288019..13eebe333b 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::Supervisor, }; let eval = L7EvalContext { diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index cd2df94118..5ed065bdf3 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -3030,6 +3030,7 @@ mod tests { product_version: "0".into(), proxy_ip: [127, 0, 0, 1].into(), proxy_port: 3128, + origin: openshell_ocsf::EventOrigin::Supervisor, }; let eval = L7EvalContext { diff --git a/crates/openshell-supervisor-process/src/log_push.rs b/crates/openshell-supervisor-process/src/log_push.rs index b24382787c..260459e645 100644 --- a/crates/openshell-supervisor-process/src/log_push.rs +++ b/crates/openshell-supervisor-process/src/log_push.rs @@ -317,8 +317,8 @@ impl tracing::field::Visit for LogVisitor { mod tests { use super::*; use openshell_ocsf::{ - ActionId, ActivityId, DispositionId, Endpoint, EventContext, NetworkActivityBuilder, - SeverityId, StatusId, ocsf_emit, + ActionId, ActivityId, DispositionId, Endpoint, EventContext, EventOrigin, + NetworkActivityBuilder, SeverityId, StatusId, ocsf_emit, }; use tracing_subscriber::layer::SubscriberExt; @@ -326,6 +326,7 @@ mod tests { EventContext { sandbox_id: "sb-test".to_string(), sandbox_name: "test-sandbox".to_string(), + origin: EventOrigin::Supervisor, container_image: "openshell/sandbox:test".to_string(), hostname: "test-host".to_string(), product_version: "0.0.0".to_string(), diff --git a/crates/openshell-supervisor-process/src/supervisor_session.rs b/crates/openshell-supervisor-process/src/supervisor_session.rs index a8f60a5681..1f0b3c0fb2 100644 --- a/crates/openshell-supervisor-process/src/supervisor_session.rs +++ b/crates/openshell-supervisor-process/src/supervisor_session.rs @@ -900,6 +900,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::Supervisor, } } diff --git a/docs/observability/ocsf-json-export.mdx b/docs/observability/ocsf-json-export.mdx index 82c4e2b8cb..4ada901fad 100644 --- a/docs/observability/ocsf-json-export.mdx +++ b/docs/observability/ocsf-json-export.mdx @@ -7,9 +7,19 @@ description: "How to enable full OCSF JSON logging for SIEM integration, complia keywords: "Generative AI, Cybersecurity, OCSF, JSON, SIEM, Compliance, Observability" --- -The [shorthand log format](/observability/logging) is optimized for humans and agents reading logs in real time. For machine consumption, compliance archival, or SIEM integration, you can enable full OCSF JSON export. This writes every OCSF event as a complete JSON record in JSONL format, one JSON object per line. +The [shorthand log format](/observability/logging) is optimized for humans and agents reading logs in real time. For machine consumption or SIEM integration, enable OCSF JSON output: one native JSON object per line. Collection is best-effort, not a guarantee of complete audit history. -## Enable JSON Export +## Gateway Output + +Configure `[openshell.gateway.ocsf_log]` in `gateway.toml` to collect gateway-origin OCSF events into one JSONL file. This includes gateway-wide events such as TLS certificate reloads, even when no sandbox log stream applies. Collection preserves native OCSF fields and event IDs and does not depend on `RUST_LOG`. + +The gateway writes OCSF events to a local JSONL file. Give each replica its own path and configure the shipper to follow renamed rotation segments. See [OCSF JSONL configuration](/reference/gateway-config#ocsf-jsonl-output) for queue bounds, retention, loss metrics, and best-effort delivery limits. + +This output is separate from the sandbox-local file described below. Enabling one does not enable the other. + +Without JSONL enabled, gateway OCSF events still appear as shorthand in console output, subject to the diagnostic log filter. Events associated with a sandbox also appear in its gateway log stream; gateway-wide events have no sandbox association. + +## Enable Sandbox JSON Export Use the `ocsf_json_enabled` setting to toggle JSON export. The setting can be applied globally, for all sandboxes, or per-sandbox. @@ -39,6 +49,10 @@ When enabled, OCSF JSON records are written to `/var/log/openshell-ocsf.YYYY-MM- ## JSON Record Structure +Gateway-produced records use the `OpenShell Gateway` product identity; supervisor records use `OpenShell Sandbox Supervisor`. The producing device and the affected sandbox are separate identities, so a gateway event can still identify an affected sandbox through its container fields. + +For gateway-produced records, `device.os.name` identifies the OS running the gateway process: Linux, Windows, or macOS. Linux supervisor records continue to report Linux, regardless of the gateway's OS. A native macOS gateway therefore reports macOS while its Linux sandboxes report Linux; a gateway running inside a Linux container reports Linux, even on a Mac host. + `metadata.uid` uniquely identifies an event and stays unchanged when that record is serialized again. Use `container.uid` to associate the event with its sandbox, not `metadata.uid`. Records without a sandbox association omit the container; unknown images are omitted rather than represented by an empty image name. Each line is a complete OCSF v1.8.0 JSON object. Here is an example of a network connection event: diff --git a/docs/reference/gateway-config.mdx b/docs/reference/gateway-config.mdx index 14cdb05ece..eb6d065ffd 100644 --- a/docs/reference/gateway-config.mdx +++ b/docs/reference/gateway-config.mdx @@ -215,6 +215,37 @@ Local Docker, Podman, and VM gateways can also set `[openshell.gateway.mtls_auth `[openshell.gateway.auth] allow_unauthenticated_users = true` is an unsafe local-development and trusted-proxy escape hatch. It accepts user-facing CLI/API calls without OIDC or mTLS credentials while sandbox supervisors still authenticate with gateway-minted sandbox JWTs. Leave it false for shared and production gateways. +## OCSF JSONL Output + +`[openshell.gateway.ocsf_log]` enables a gateway-local file containing native OCSF events emitted by the gateway, including events without a sandbox association. Omit the table to disable it. Restart the gateway after changing this configuration. + +```toml +[openshell.gateway.ocsf_log] +path = "/var/log/openshell/gateway-ocsf.jsonl" +rotation = "daily" +max_files = 7 +queue_capacity = 10000 +queue_max_bytes = 16777216 +``` + +| Field | Default | Behavior | +|---|---|---| +| `path` | Required | Nonempty file path; relative paths use the gateway's working directory. | +| `rotation` | `daily` | Rotate on the first write of a new UTC day. `never` allows unbounded file growth. | +| `max_files` | `7` | Positive number of rotated segments to retain, in addition to the active file. Rejects explicit use with `rotation = "never"`. | +| `queue_capacity` | `10000` | Positive queued-record limit. Overflow discards incoming records. | +| `queue_max_bytes` | `16777216` | Positive limit on queued encoded bytes. Records exceeding this limit are discarded. | + +Unknown fields are rejected. This is one file destination, not an exporter registry. The existing `ocsf_json_enabled` sandbox setting remains independent. + +Give each gateway replica its own writable file and an external shipper read access to the containing directory. New files use owner-only permissions on Unix; arrange shipper access deliberately. Rotation renames the active file to a date- and UUID-suffixed sibling. The shipper must follow rotated files and checkpoint its progress. Do not use multiple writers or external copy-truncate rotation on this path. + +The gateway queues OCSF independently of `RUST_LOG`, excludes ordinary diagnostics, and preserves native event IDs. It writes up to 100 records per batch with a 500 ms batching interval. One additional bounded batch can be in flight. File errors do not stop sandbox execution: failed writes are not replayed, and subsequent records are discarded during reopen backoff. A restarted writer removes an incomplete trailing line before appending. + +Metrics include `openshell_ocsf_log_queued_total`, `openshell_ocsf_log_written_total`, `openshell_ocsf_log_queue_records`, `openshell_ocsf_log_queue_bytes`, `openshell_ocsf_log_dropped_total` (with a bounded `reason` label), and `openshell_ocsf_log_writer_errors_total`. Recovery reports discarded bytes through `openshell_ocsf_log_recovery_discarded_bytes_total`, not an invented lost-record count. Warnings report writer failures independently of this file. + +Allow enough retention and disk space for shipper outages. Graceful shutdown allows five seconds for draining; outstanding OS writes remain uncertain if they outlast that budget. Flush does not mean fsync. The file is a best-effort collection boundary, not durable acceptance, a replay protocol, or proof of a complete audit history. See [OCSF JSON export](/observability/ocsf-json-export) for the separate sandbox-local output. + ## OTLP Export `[openshell.gateway.otlp]` enables OpenTelemetry export over OTLP/gRPC. Omit the table to disable export; there is no separate `enabled` flag. diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 7d47b18065..7d15b37d0b 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -678,6 +678,7 @@ configuration — check that the gateway spawned the driver binary you expect | Symptom | Likely cause | Check | |---|---|---| | `openshell status` fails | Gateway endpoint unreachable or auth mismatch | `openshell gateway info`, gateway logs | +| Gateway OCSF JSONL stops growing or has gaps | File errors, queue pressure, or shipper falling behind retention | Inspect gateway warnings and `openshell_ocsf_log_*` metrics; check `[openshell.gateway.ocsf_log]`, directory permissions, free space, per-replica paths, and shipper rotation checkpoints. See the published [gateway configuration reference](https://docs.nvidia.com/openshell/latest/reference/gateway-config.md). | | `BatchSpanProcessor.ExportError` repeatedly reports connection refused on `127.0.0.1:4317` | The local gateway started with OTLP configured but the collector forwarding task later stopped, or the config was created manually | Restart `gateway:docker`, `gateway:podman`, or `gateway:vm` so it re-detects the listener; inspect the generated `gateway.toml` for `[openshell.gateway.otlp]` | | Gateway starts but sandbox create fails | Compute driver cannot reach runtime | Docker/Podman/Kubernetes/VM driver logs | | Gateway exits while resolving compute-driver listener requirements | Callback alias topology is unsupported, the Podman network cannot be inspected, or the selected address is not private/authorized | Gateway startup error, `podman info --debug`, Podman network inspection, host IPv4 default route |