diff --git a/crates/openshell-core/src/settings.rs b/crates/openshell-core/src/settings.rs index f819d8ab38..2871c98034 100644 --- a/crates/openshell-core/src/settings.rs +++ b/crates/openshell-core/src/settings.rs @@ -106,6 +106,13 @@ pub const PROPOSAL_APPROVAL_MODE_KEY: &str = "proposal_approval_mode"; /// fail-closes on unknown persisted values for defense in depth. pub const PROPOSAL_APPROVAL_MODE_VALUES: &[&str] = &["manual", "auto"]; +/// Interval in seconds for periodic OCSF Network Activity `Traffic` events. +/// +/// The events carry per-connection byte counts. 0 (the default) disables +/// them; values below 5 are raised to 5. Connection `Close` events with totals are always +/// emitted. Traffic events go to the OCSF JSONL file and OTLP export only. +pub const OCSF_TRAFFIC_INTERVAL_SECS_KEY: &str = "ocsf_traffic_interval_secs"; + pub const REGISTERED_SETTINGS: &[RegisteredSetting] = &[ // When true the supervisor emits OCSF-JSON records to stderr and writes // JSONL to `/var/log/openshell-ocsf*.jsonl` when file logging is available. @@ -115,6 +122,12 @@ pub const REGISTERED_SETTINGS: &[RegisteredSetting] = &[ kind: SettingValueKind::Bool, allowed_string_values: None, }, + // See OCSF_TRAFFIC_INTERVAL_SECS_KEY. Defaults to 0 (off). + RegisteredSetting { + key: OCSF_TRAFFIC_INTERVAL_SECS_KEY, + kind: SettingValueKind::Int, + allowed_string_values: None, + }, // Deprecated and ignored: OpenShell emits OCSF 1.8.0 only. Still // registered so existing values load; the supervisor reports a set value. RegisteredSetting { diff --git a/crates/openshell-ocsf/src/builders/network.rs b/crates/openshell-ocsf/src/builders/network.rs index 1e182af674..7ad98aad28 100644 --- a/crates/openshell-ocsf/src/builders/network.rs +++ b/crates/openshell-ocsf/src/builders/network.rs @@ -7,7 +7,7 @@ use crate::builders::ContextRef; use crate::enums::{ActionId, ActivityId, DispositionId, SeverityId, StatusId}; use crate::events::base_event::BaseEventData; use crate::events::{NetworkActivityEvent, OcsfEvent}; -use crate::objects::{Actor, ConnectionInfo, Endpoint, FirewallRule}; +use crate::objects::{Actor, ConnectionInfo, Endpoint, FirewallRule, TrafficReport}; /// Builder for Network Activity [4001] events. pub struct NetworkActivityBuilder<'a, EndpointState = MissingNetworkEndpoint> { @@ -28,6 +28,7 @@ pub struct NetworkActivityBuilder<'a, EndpointState = MissingNetworkEndpoint> { status_detail: Option, unmapped: Option>, log_source: Option, + traffic: TrafficReport, endpoint_state: std::marker::PhantomData, } @@ -68,6 +69,7 @@ impl<'a> NetworkActivityBuilder<'a, MissingNetworkEndpoint> { status_detail: None, unmapped: None, log_source: None, + traffic: TrafficReport::default(), endpoint_state: std::marker::PhantomData, } } @@ -127,6 +129,7 @@ impl<'a, EndpointState> NetworkActivityBuilder<'a, EndpointState> { status_detail: self.status_detail, unmapped: self.unmapped, log_source: self.log_source, + traffic: self.traffic, endpoint_state: std::marker::PhantomData, } } @@ -153,6 +156,7 @@ impl<'a, EndpointState> NetworkActivityBuilder<'a, EndpointState> { status_detail: self.status_detail, unmapped: self.unmapped, log_source: self.log_source, + traffic: self.traffic, endpoint_state: std::marker::PhantomData, } } @@ -198,6 +202,35 @@ impl<'a, EndpointState> NetworkActivityBuilder<'a, EndpointState> { self } + /// Bytes moved in this event's window. + #[must_use] + pub fn traffic(mut self, traffic: crate::objects::NetworkTraffic) -> Self { + self.traffic.traffic = Some(traffic); + self + } + + /// Bytes moved over the life of the connection. + #[must_use] + pub fn cumulative_traffic(mut self, traffic: crate::objects::NetworkTraffic) -> Self { + self.traffic.cumulative_traffic = Some(traffic); + self + } + + /// Window covered by this event, milliseconds since epoch. + #[must_use] + pub fn time_window(mut self, start_ms: i64, end_ms: i64) -> Self { + self.traffic.start_time = Some(start_ms); + self.traffic.end_time = Some(end_ms); + self + } + + /// Connection duration in milliseconds. + #[must_use] + pub fn duration_ms(mut self, duration_ms: i64) -> Self { + self.traffic.duration = Some(duration_ms); + self + } + #[must_use] pub fn firewall_rule(mut self, name: &str, rule_type: &str) -> Self { self.firewall_rule = Some(FirewallRule::new(name, rule_type)); @@ -255,6 +288,7 @@ impl NetworkActivityBuilder<'_, HasNetworkEndpoint> { disposition: self.disposition, observation_point_id: self.observation_point_id, is_src_dst_assignment_known: Some(true), + traffic: self.traffic, }) } } diff --git a/crates/openshell-ocsf/src/events/network_activity.rs b/crates/openshell-ocsf/src/events/network_activity.rs index 96cf3c5dac..dcda2af748 100644 --- a/crates/openshell-ocsf/src/events/network_activity.rs +++ b/crates/openshell-ocsf/src/events/network_activity.rs @@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize}; use crate::enums::{ActionId, DispositionId}; use crate::events::base_event::BaseEventData; -use crate::objects::{Actor, ConnectionInfo, Endpoint, FirewallRule}; +use crate::objects::{Actor, ConnectionInfo, Endpoint, FirewallRule, TrafficReport}; /// OCSF Network Activity Event [4001]. /// @@ -61,6 +61,10 @@ pub struct NetworkActivityEvent { /// Whether src/dst assignment is known (v1.6.0+). #[serde(skip_serializing_if = "Option::is_none")] pub is_src_dst_assignment_known: Option, + + /// Byte counts and window for Traffic and Close events. + #[serde(flatten, default)] + pub traffic: TrafficReport, } impl Serialize for NetworkActivityEvent { @@ -86,6 +90,11 @@ impl Serialize for NetworkActivityEvent { "is_src_dst_assignment_known", self.is_src_dst_assignment_known ); + insert_optional!(obj, "traffic", self.traffic.traffic); + insert_optional!(obj, "cumulative_traffic", self.traffic.cumulative_traffic); + insert_optional!(obj, "start_time", self.traffic.start_time); + insert_optional!(obj, "end_time", self.traffic.end_time); + insert_optional!(obj, "duration", self.traffic.duration); base_val.serialize(serializer) } @@ -127,6 +136,7 @@ mod tests { disposition: Some(DispositionId::Allowed), observation_point_id: Some(2), is_src_dst_assignment_known: Some(true), + traffic: TrafficReport::default(), }; let json = serde_json::to_value(&event).unwrap(); diff --git a/crates/openshell-ocsf/src/format/shorthand.rs b/crates/openshell-ocsf/src/format/shorthand.rs index 5b80f29205..38780faf62 100644 --- a/crates/openshell-ocsf/src/format/shorthand.rs +++ b/crates/openshell-ocsf/src/format/shorthand.rs @@ -732,6 +732,7 @@ mod tests { disposition: Some(DispositionId::Allowed), observation_point_id: None, is_src_dst_assignment_known: None, + traffic: TrafficReport::default(), }); let shorthand = event.format_shorthand(); @@ -759,6 +760,7 @@ mod tests { disposition: Some(DispositionId::Blocked), observation_point_id: Some(3), is_src_dst_assignment_known: Some(true), + traffic: TrafficReport::default(), }); let shorthand = event.format_shorthand(); @@ -898,6 +900,7 @@ mod tests { disposition: Some(DispositionId::Blocked), observation_point_id: None, is_src_dst_assignment_known: None, + traffic: TrafficReport::default(), }); let shorthand = event.format_shorthand(); @@ -925,6 +928,7 @@ mod tests { disposition: Some(DispositionId::Allowed), observation_point_id: None, is_src_dst_assignment_known: None, + traffic: TrafficReport::default(), }); let shorthand = event.format_shorthand(); @@ -952,6 +956,7 @@ mod tests { disposition: None, observation_point_id: None, is_src_dst_assignment_known: None, + traffic: TrafficReport::default(), }); let shorthand = event.format_shorthand(); @@ -1012,6 +1017,7 @@ mod tests { disposition: Some(DispositionId::Blocked), observation_point_id: None, is_src_dst_assignment_known: None, + traffic: TrafficReport::default(), }); event.format_shorthand() } diff --git a/crates/openshell-ocsf/src/lib.rs b/crates/openshell-ocsf/src/lib.rs index 661b4c6f5a..c95b4110a7 100644 --- a/crates/openshell-ocsf/src/lib.rs +++ b/crates/openshell-ocsf/src/lib.rs @@ -53,8 +53,9 @@ pub use enums::{ // --- Object types --- pub use objects::{ Actor, AiModel, Api, Attack, ConnectionInfo, Container, Device, Endpoint, Evidence, - FindingInfo, FirewallRule, HttpRequest, HttpResponse, Image, Metadata, OsInfo, Process, - Product, Remediation, ResourceDetails, Service, Tactic, Technique, Url, User, + FindingInfo, FirewallRule, HttpRequest, HttpResponse, Image, Metadata, NetworkTraffic, OsInfo, + Process, Product, Remediation, ResourceDetails, Service, Tactic, Technique, TrafficReport, Url, + User, }; // --- Sanitizers --- @@ -69,7 +70,7 @@ pub use builders::{ // --- Tracing layers --- pub use tracing_layers::{ - OCSF_TARGET, OcsfJsonSettings, OcsfJsonlLayer, OcsfShorthandLayer, bind_thread_json_settings, - clear_current_event, clone_current_event, emit_ocsf_event, emit_ocsf_event_routed, - set_current_event, + OCSF_TARGET, OCSF_TRAFFIC_TARGET, OcsfJsonSettings, OcsfJsonlLayer, OcsfShorthandLayer, + bind_thread_json_settings, clear_current_event, clone_current_event, emit_ocsf_event, + emit_ocsf_event_routed, emit_ocsf_traffic_event, is_structured_ocsf_target, set_current_event, }; diff --git a/crates/openshell-ocsf/src/objects/connection.rs b/crates/openshell-ocsf/src/objects/connection.rs index 4a25054bf8..43bf8c3ca8 100644 --- a/crates/openshell-ocsf/src/objects/connection.rs +++ b/crates/openshell-ocsf/src/objects/connection.rs @@ -21,6 +21,10 @@ pub struct ConnectionInfo { /// Sibling label for `direction_id`. pub direction: String, + + /// Stable per-connection identifier shared by its Traffic and Close events. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub uid: Option, } impl ConnectionInfo { @@ -32,8 +36,16 @@ impl ConnectionInfo { protocol_name: protocol.to_string(), direction_id, direction: direction_id.to_string(), + uid: None, } } + + /// Set the stable connection identifier. + #[must_use] + pub fn with_uid(mut self, uid: impl Into) -> Self { + self.uid = Some(uid.into()); + self + } } #[cfg(test)] diff --git a/crates/openshell-ocsf/src/objects/mod.rs b/crates/openshell-ocsf/src/objects/mod.rs index 9025834bfe..6e97dd28d2 100644 --- a/crates/openshell-ocsf/src/objects/mod.rs +++ b/crates/openshell-ocsf/src/objects/mod.rs @@ -17,6 +17,7 @@ mod metadata; mod process; mod resource; mod service; +mod traffic; mod user; pub use ai_model::AiModel; @@ -33,4 +34,5 @@ pub use metadata::{Metadata, Product}; pub use process::{Actor, Process}; pub use resource::ResourceDetails; pub use service::Service; +pub use traffic::{NetworkTraffic, TrafficReport}; pub use user::User; diff --git a/crates/openshell-ocsf/src/objects/traffic.rs b/crates/openshell-ocsf/src/objects/traffic.rs new file mode 100644 index 0000000000..c419ac380c --- /dev/null +++ b/crates/openshell-ocsf/src/objects/traffic.rs @@ -0,0 +1,52 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OCSF `network_traffic` object. + +use serde::{Deserialize, Serialize}; + +/// OCSF Network Traffic object. +/// +/// The supervisor is a userspace proxy, so it reports socket bytes only and +/// never packet counts. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct NetworkTraffic { + /// Bytes received from the remote endpoint. + pub bytes_in: u64, + /// Bytes sent to the remote endpoint. + pub bytes_out: u64, + /// `bytes_in + bytes_out`. + pub bytes: u64, +} + +impl NetworkTraffic { + #[must_use] + pub fn new(bytes_in: u64, bytes_out: u64) -> Self { + Self { + bytes_in, + bytes_out, + bytes: bytes_in.saturating_add(bytes_out), + } + } +} + +/// Traffic amounts and the time window they cover, carried by Network +/// Activity `Traffic` (interval deltas) and `Close` (totals) events. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)] +pub struct TrafficReport { + /// Amounts for this event's window (OCSF reserves `traffic` for deltas). + #[serde(default)] + pub traffic: Option, + /// Totals over the life of the connection. + #[serde(default)] + pub cumulative_traffic: Option, + /// Window start, milliseconds since epoch. + #[serde(default)] + pub start_time: Option, + /// Window end, milliseconds since epoch. + #[serde(default)] + pub end_time: Option, + /// Connection duration in milliseconds (close events). + #[serde(default)] + pub duration: Option, +} diff --git a/crates/openshell-ocsf/src/tracing_layers/event_bridge.rs b/crates/openshell-ocsf/src/tracing_layers/event_bridge.rs index c4a5ccb895..863ddff095 100644 --- a/crates/openshell-ocsf/src/tracing_layers/event_bridge.rs +++ b/crates/openshell-ocsf/src/tracing_layers/event_bridge.rs @@ -19,6 +19,19 @@ std::thread_local! { /// Target string used to identify OCSF tracing events. pub const OCSF_TARGET: &str = "ocsf"; +/// Target string for high-volume traffic accounting events. +/// +/// Only the structured sinks (JSONL file, OTLP export) accept this target. +/// Human-readable sinks (shorthand console and file, gateway log push) skip +/// it so periodic byte reports never flood `openshell logs`. +pub const OCSF_TRAFFIC_TARGET: &str = "ocsf_traffic"; + +/// Whether `target` carries an OCSF event that structured sinks export. +#[must_use] +pub fn is_structured_ocsf_target(target: &str) -> bool { + target == OCSF_TARGET || target == OCSF_TRAFFIC_TARGET +} + /// Clone the current thread-local OCSF event, if any. /// /// Multiple layers can call this for the same event — each receives @@ -46,6 +59,13 @@ pub fn emit_ocsf_event(event: OcsfEvent) { clear_current_event(); } +/// Emit a traffic accounting `OcsfEvent` to the structured sinks only. +pub fn emit_ocsf_traffic_event(event: OcsfEvent) { + set_current_event(event); + tracing::info!(target: "ocsf_traffic", "ocsf_event"); + clear_current_event(); +} + /// Store an `OcsfEvent` in the thread-local bridge so OCSF layers /// (`OcsfJsonlLayer` / `OcsfShorthandLayer`) can `clone_current_event()` it /// during tracing dispatch. diff --git a/crates/openshell-ocsf/src/tracing_layers/jsonl_layer.rs b/crates/openshell-ocsf/src/tracing_layers/jsonl_layer.rs index 245f92e596..2fdb22e4de 100644 --- a/crates/openshell-ocsf/src/tracing_layers/jsonl_layer.rs +++ b/crates/openshell-ocsf/src/tracing_layers/jsonl_layer.rs @@ -13,7 +13,7 @@ use tracing::Subscriber; use tracing_subscriber::Layer; use tracing_subscriber::layer::Context; -use crate::tracing_layers::event_bridge::{OCSF_TARGET, clone_current_event}; +use crate::tracing_layers::event_bridge::{OCSF_TARGET, OCSF_TRAFFIC_TARGET, clone_current_event}; /// OCSF JSON output settings owned by one hosted sandbox instance. /// @@ -88,7 +88,9 @@ where W: Write + Send + 'static, { fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) { - if event.metadata().target() != OCSF_TARGET { + let target = event.metadata().target(); + // Traffic accounting events go to files and collectors, never the console. + if target != OCSF_TARGET && (self.console_format || target != OCSF_TRAFFIC_TARGET) { return; } diff --git a/crates/openshell-ocsf/src/tracing_layers/mod.rs b/crates/openshell-ocsf/src/tracing_layers/mod.rs index 808051b061..67e0b6f984 100644 --- a/crates/openshell-ocsf/src/tracing_layers/mod.rs +++ b/crates/openshell-ocsf/src/tracing_layers/mod.rs @@ -12,8 +12,8 @@ mod jsonl_layer; mod shorthand_layer; pub use event_bridge::{ - OCSF_TARGET, clear_current_event, clone_current_event, emit_ocsf_event, emit_ocsf_event_routed, - set_current_event, + OCSF_TARGET, OCSF_TRAFFIC_TARGET, clear_current_event, clone_current_event, emit_ocsf_event, + emit_ocsf_event_routed, emit_ocsf_traffic_event, is_structured_ocsf_target, set_current_event, }; pub use jsonl_layer::{OcsfJsonSettings, OcsfJsonlLayer, bind_thread_json_settings}; pub use shorthand_layer::OcsfShorthandLayer; diff --git a/crates/openshell-ocsf/src/tracing_layers/shorthand_layer.rs b/crates/openshell-ocsf/src/tracing_layers/shorthand_layer.rs index 37109ad6bf..10630b5e01 100644 --- a/crates/openshell-ocsf/src/tracing_layers/shorthand_layer.rs +++ b/crates/openshell-ocsf/src/tracing_layers/shorthand_layer.rs @@ -11,7 +11,7 @@ use tracing::Subscriber; use tracing_subscriber::Layer; use tracing_subscriber::layer::Context; -use crate::tracing_layers::event_bridge::{OCSF_TARGET, clone_current_event}; +use crate::tracing_layers::event_bridge::{OCSF_TARGET, OCSF_TRAFFIC_TARGET, clone_current_event}; /// A tracing `Layer` that intercepts OCSF events and writes shorthand output. /// @@ -53,6 +53,9 @@ where fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) { let meta = event.metadata(); + if meta.target() == OCSF_TRAFFIC_TARGET { + return; + } let ts = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ"); if meta.target() == OCSF_TARGET { diff --git a/crates/openshell-otel/src/ocsf.rs b/crates/openshell-otel/src/ocsf.rs index 680b4c1e3e..be59237d0d 100644 --- a/crates/openshell-otel/src/ocsf.rs +++ b/crates/openshell-otel/src/ocsf.rs @@ -7,7 +7,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, SystemTime}; -use openshell_ocsf::{OCSF_TARGET, OcsfEvent, SeverityId, clone_current_event}; +use openshell_ocsf::{OcsfEvent, SeverityId, clone_current_event, is_structured_ocsf_target}; use opentelemetry::InstrumentationScope; use opentelemetry::logs::{AnyValue, LogRecord as _, Logger as _, LoggerProvider as _, Severity}; use opentelemetry_sdk::Resource; @@ -110,7 +110,7 @@ pub struct OcsfOtlpLayer { impl Layer for OcsfOtlpLayer { fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { - if event.metadata().target() != OCSF_TARGET { + if !is_structured_ocsf_target(event.metadata().target()) { return; } let Some(event) = clone_current_event() else { diff --git a/crates/openshell-supervisor-network/src/lib.rs b/crates/openshell-supervisor-network/src/lib.rs index d9fcd004b6..c2f8651813 100644 --- a/crates/openshell-supervisor-network/src/lib.rs +++ b/crates/openshell-supervisor-network/src/lib.rs @@ -25,6 +25,7 @@ pub mod run; pub mod sigv4; mod spiffe_endpoint; mod token_grant; +pub mod traffic; pub mod upstream_proxy; #[cfg(test)] diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 7560a3aeaf..78bf089348 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -1364,7 +1364,7 @@ async fn handle_transparent_tcp_connection( let approved_real_ip_candidates = connector.addrs().to_vec(); generation_guard.ensure_current()?; let dial_span = egress::dial_span(&connect_span); - let mut upstream = + let upstream = dial_transparent_upstream(&upstream_proxy, &host, port, &approved_real_ip_candidates) .instrument(dial_span.clone()) .await @@ -1433,6 +1433,7 @@ async fn handle_transparent_tcp_connection( } )); emit_activity(&activity_tx, false, "transparent_tcp"); + let mut upstream = track_upstream(upstream, &decision, &host, port); relay::relay_tcp(&mut client, &mut upstream, &generation_guard, &ctx).await } @@ -3094,7 +3095,7 @@ async fn handle_mediated_connection( return Ok(()); }; let mut upstream = match upstream_result { - Ok(upstream) => upstream, + Ok(upstream) => track_upstream(upstream, &decision, &host_lc, port), Err(error) => { egress::mark_error(&dial_span); if let Some(observer) = connect_endpoint_observer.as_ref() { @@ -4587,6 +4588,33 @@ fn validate_declared_endpoint_resolved_addrs( Ok(()) } +/// Count the bytes an authorized upstream connection moves and report them +/// as OCSF traffic events. +fn track_upstream( + stream: S, + decision: &EgressDecision, + host: &str, + port: u16, +) -> crate::traffic::CountingStream { + let policy = match &decision.action { + NetworkAction::Allow { matched_policy } => matched_policy.as_deref().unwrap_or("-"), + NetworkAction::Deny { .. } => "-", + }; + crate::traffic::track( + stream, + crate::traffic::ConnectionMeta { + host: host.to_string(), + port, + policy: policy.to_string(), + binary: decision + .binary + .as_ref() + .map_or_else(|| "-".to_string(), |path| path.display().to_string()), + pid: decision.binary_pid, + }, + ) +} + /// Dial a validated upstream destination for a TLS (CONNECT) tunnel. /// /// Connects directly to the SSRF-checked resolved addresses, or chains @@ -6716,7 +6744,7 @@ async fn handle_forward_proxy( let dial_span = egress::dial_span(&connect_span); let dial_result = connector.connect().instrument(dial_span.clone()).await; let mut upstream = match dial_result { - Ok(s) => s, + Ok(s) => track_upstream(s, &decision, &host_lc, port), Err(e) => { egress::mark_error(&dial_span); let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) diff --git a/crates/openshell-supervisor-network/src/traffic.rs b/crates/openshell-supervisor-network/src/traffic.rs new file mode 100644 index 0000000000..f27706f80b --- /dev/null +++ b/crates/openshell-supervisor-network/src/traffic.rs @@ -0,0 +1,322 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Per-connection egress byte accounting reported as OCSF Network Activity. +//! +//! [`track`] wraps the upstream side of a relay in a [`CountingStream`] that +//! adds every byte read from or written to the upstream socket to shared +//! atomics. A reporter thread, never the relay task, reads them: +//! +//! - every connection gets one `Close` event with its totals +//! (`cumulative_traffic`), lifetime, and the bytes not yet reported +//! (`traffic`); +//! - when `ocsf_traffic_interval_secs` is non-zero, connections that moved +//! bytes also get a `Traffic` event per interval with that interval's +//! amounts. +//! +//! Summing `traffic` over a connection's events gives its totals. Events use +//! [`openshell_ocsf::OCSF_TRAFFIC_TARGET`], so only the JSONL file and OTLP +//! export receive them. + +use std::pin::Pin; +use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; +use std::sync::{Arc, LazyLock, Mutex, Once, PoisonError}; +use std::task::{Context, Poll}; +use std::time::Duration; + +use openshell_ocsf::{ + ActionId, ActivityId, ConnectionInfo, ContextRef, DispositionId, Endpoint, + NetworkActivityBuilder, NetworkTraffic, Process, SeverityId, StatusId, emit_ocsf_traffic_event, +}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +/// Smallest accepted interval. Shorter values are raised to this. +pub const MIN_INTERVAL_SECS: u64 = 5; +/// How often the reporter checks for closed connections and due intervals. +const TICK: Duration = Duration::from_secs(1); + +/// Interim report interval in seconds; 0 disables `Traffic` events. +static INTERVAL_SECS: AtomicU64 = AtomicU64::new(0); +static REGISTRY: LazyLock>>> = LazyLock::new(Mutex::default); +static REPORTER: Once = Once::new(); + +/// Apply the `ocsf_traffic_interval_secs` setting. Returns the effective value. +pub fn set_interval_secs(requested: u64) -> u64 { + let effective = if requested == 0 { + 0 + } else { + requested.max(MIN_INTERVAL_SECS) + }; + INTERVAL_SECS.store(effective, Ordering::Relaxed); + effective +} + +/// What a tracked connection is attributed to. +#[derive(Debug, Clone)] +pub struct ConnectionMeta { + pub host: String, + pub port: u16, + /// Policy key that authorized the connection, or `-`. + pub policy: String, + pub binary: String, + pub pid: Option, +} + +struct Connection { + uid: String, + meta: ConnectionMeta, + ctx: ContextRef<'static>, + started_ms: i64, + bytes_in: AtomicU64, + bytes_out: AtomicU64, + /// 0 while open. + closed_ms: AtomicI64, + // Reporter-owned progress. + reported_in: AtomicU64, + reported_out: AtomicU64, + window_start_ms: AtomicI64, +} + +/// Count bytes moved through `inner` and report them for `meta`. +pub fn track(inner: S, meta: ConnectionMeta) -> CountingStream { + REPORTER.call_once(spawn_reporter); + let now = now_ms(); + let connection = Arc::new(Connection { + uid: format!("conn-{}", uuid::Uuid::new_v4()), + meta, + ctx: openshell_ocsf::ctx::ctx(), + started_ms: now, + bytes_in: AtomicU64::new(0), + bytes_out: AtomicU64::new(0), + closed_ms: AtomicI64::new(0), + reported_in: AtomicU64::new(0), + reported_out: AtomicU64::new(0), + window_start_ms: AtomicI64::new(now), + }); + REGISTRY + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(Arc::clone(&connection)); + CountingStream { inner, connection } +} + +/// Upstream stream wrapper that counts bytes in each direction. +pub struct CountingStream { + inner: S, + connection: Arc, +} + +impl Drop for CountingStream { + fn drop(&mut self) { + self.connection.closed_ms.store(now_ms(), Ordering::Release); + } +} + +impl AsyncRead for CountingStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + let before = buf.filled().len(); + let poll = Pin::new(&mut this.inner).poll_read(cx, buf); + if matches!(poll, Poll::Ready(Ok(()))) { + let read = (buf.filled().len() - before) as u64; + this.connection.bytes_in.fetch_add(read, Ordering::Relaxed); + } + poll + } +} + +impl AsyncWrite for CountingStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let this = self.get_mut(); + let poll = Pin::new(&mut this.inner).poll_write(cx, buf); + if let Poll::Ready(Ok(written)) = poll { + this.connection + .bytes_out + .fetch_add(written as u64, Ordering::Relaxed); + } + poll + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[std::io::IoSlice<'_>], + ) -> Poll> { + let this = self.get_mut(); + let poll = Pin::new(&mut this.inner).poll_write_vectored(cx, bufs); + if let Poll::Ready(Ok(written)) = poll { + this.connection + .bytes_out + .fetch_add(written as u64, Ordering::Relaxed); + } + poll + } + + fn is_write_vectored(&self) -> bool { + self.inner.is_write_vectored() + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_shutdown(cx) + } +} + +fn spawn_reporter() { + let spawned = std::thread::Builder::new() + .name("openshell-traffic".to_string()) + .spawn(|| { + let mut last_interval_ms = now_ms(); + loop { + std::thread::sleep(TICK); + let now = now_ms(); + let interval = INTERVAL_SECS.load(Ordering::Relaxed); + let interval_due = interval > 0 + && now - last_interval_ms >= i64::try_from(interval * 1000).unwrap_or(i64::MAX); + if interval_due { + last_interval_ms = now; + } + report(now, interval_due); + } + }); + if let Err(error) = spawned { + tracing::warn!(%error, "traffic reporter could not start; no traffic events"); + } +} + +/// Emit Close events for closed connections and, when `interval_due`, +/// Traffic events for open connections that moved bytes. +fn report(now: i64, interval_due: bool) { + let (closed, open): (Vec<_>, Vec<_>) = { + let mut registry = REGISTRY.lock().unwrap_or_else(PoisonError::into_inner); + let snapshot = std::mem::take(&mut *registry); + let (closed, open): (Vec<_>, Vec<_>) = snapshot + .into_iter() + .partition(|c| c.closed_ms.load(Ordering::Acquire) != 0); + registry.extend(open.iter().cloned()); + (closed, open) + }; + if interval_due { + for connection in &open { + if let Some(delta) = take_delta(connection) { + let start = connection.window_start_ms.swap(now, Ordering::Relaxed); + emit(connection, ActivityId::Traffic, delta, None, start, now); + } + } + } + for connection in &closed { + let delta = take_delta(connection).unwrap_or_default(); + let closed_ms = connection.closed_ms.load(Ordering::Acquire); + let total = NetworkTraffic::new( + connection.bytes_in.load(Ordering::Relaxed), + connection.bytes_out.load(Ordering::Relaxed), + ); + emit( + connection, + ActivityId::Close, + delta, + Some(total), + connection.started_ms, + closed_ms, + ); + } +} + +/// Bytes not yet reported, or `None` when nothing moved. +fn take_delta(connection: &Connection) -> Option { + let bytes_in = connection.bytes_in.load(Ordering::Relaxed); + let bytes_out = connection.bytes_out.load(Ordering::Relaxed); + let prev_in = connection.reported_in.swap(bytes_in, Ordering::Relaxed); + let prev_out = connection.reported_out.swap(bytes_out, Ordering::Relaxed); + let delta = NetworkTraffic::new(bytes_in - prev_in, bytes_out - prev_out); + (delta.bytes > 0).then_some(delta) +} + +fn emit( + connection: &Connection, + activity: ActivityId, + delta: NetworkTraffic, + total: Option, + start_ms: i64, + end_ms: i64, +) { + let meta = &connection.meta; + let verb = if total.is_some() { "CLOSE" } else { "TRAFFIC" }; + let mut builder = NetworkActivityBuilder::new(connection.ctx.clone()) + .activity(activity) + .action(ActionId::Allowed) + .disposition(DispositionId::Allowed) + .severity(SeverityId::Informational) + .status(StatusId::Success) + .dst_endpoint(Endpoint::from_domain(&meta.host, meta.port)) + .actor_process(Process::new(&meta.binary, meta.pid.map_or(0, i64::from))) + .firewall_rule(&meta.policy, "opa") + .connection_info(ConnectionInfo::new("tcp").with_uid(&connection.uid)) + .traffic(delta) + .time_window(start_ms, end_ms) + .message(format!( + "{verb} {}:{} in={} out={}", + meta.host, meta.port, delta.bytes_in, delta.bytes_out + )); + if let Some(total) = total { + builder = builder + .cumulative_traffic(total) + .duration_ms(end_ms - connection.started_ms); + } + emit_ocsf_traffic_event(builder.build()); +} + +fn now_ms() -> i64 { + openshell_core::time::now_ms() +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + #[tokio::test] + async fn counting_stream_counts_both_directions() { + let (client, mut server) = tokio::io::duplex(1024); + let mut stream = track( + client, + ConnectionMeta { + host: "artifactory.internal".to_string(), + port: 443, + policy: "pkg_proxy".to_string(), + binary: "/usr/bin/curl".to_string(), + pid: Some(42), + }, + ); + stream.write_all(b"hello").await.unwrap(); + server.write_all(b"world!!").await.unwrap(); + let mut buf = [0u8; 7]; + stream.read_exact(&mut buf).await.unwrap(); + let connection = Arc::clone(&stream.connection); + assert_eq!(connection.bytes_out.load(Ordering::Relaxed), 5); + assert_eq!(connection.bytes_in.load(Ordering::Relaxed), 7); + assert_eq!(take_delta(&connection), Some(NetworkTraffic::new(7, 5))); + assert_eq!(take_delta(&connection), None); + drop(stream); + assert_ne!(connection.closed_ms.load(Ordering::Relaxed), 0); + } + + #[test] + fn interval_is_clamped() { + assert_eq!(set_interval_secs(0), 0); + assert_eq!(set_interval_secs(1), MIN_INTERVAL_SECS); + assert_eq!(set_interval_secs(30), 30); + set_interval_secs(0); + } +} diff --git a/crates/openshell-supervisor-process/src/log_push.rs b/crates/openshell-supervisor-process/src/log_push.rs index 2d1c8dc701..c1eb293d8c 100644 --- a/crates/openshell-supervisor-process/src/log_push.rs +++ b/crates/openshell-supervisor-process/src/log_push.rs @@ -110,6 +110,10 @@ impl Layer for LogPushLayer { fn push_event(sink: &LogSink, event: &Event<'_>) { let meta = event.metadata(); + // Traffic accounting events go to structured sinks only. + if meta.target() == openshell_ocsf::OCSF_TRAFFIC_TARGET { + 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. diff --git a/crates/openshell-supervisor/src/config_runtime.rs b/crates/openshell-supervisor/src/config_runtime.rs index 8ced34ffec..2d19c227ac 100644 --- a/crates/openshell-supervisor/src/config_runtime.rs +++ b/crates/openshell-supervisor/src/config_runtime.rs @@ -689,6 +689,7 @@ impl ConfigRuntime { // Apply OCSF JSON toggle from the `ocsf_json_enabled` setting. apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings); + apply_ocsf_traffic_setting(&result.settings); report_deprecated_ocsf_schema_version(&ctx.ocsf_schema_version, &result.settings); // Apply the agent-proposals feature toggle. On a false→true transition @@ -1277,6 +1278,7 @@ async fn run_policy_poll_loop_with_client( runtime.applied_revision = Some(sandbox_config_revision(&result)); } apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings); + apply_ocsf_traffic_setting(&result.settings); report_deprecated_ocsf_schema_version( &ctx.ocsf_schema_version, &result.settings, @@ -1319,6 +1321,7 @@ async fn run_policy_poll_loop_with_client( } (InitialPollDisposition::TrackOnly, _) => { apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings); + apply_ocsf_traffic_setting(&result.settings); report_deprecated_ocsf_schema_version( &ctx.ocsf_schema_version, &result.settings, @@ -1455,6 +1458,32 @@ fn apply_ocsf_json_setting( } } +/// Apply the `ocsf_traffic_interval_secs` setting to the traffic reporter. +fn apply_ocsf_traffic_setting( + settings: &std::collections::HashMap, +) { + use openshell_core::proto::setting_value; + use std::sync::atomic::{AtomicU64, Ordering}; + + static APPLIED: AtomicU64 = AtomicU64::new(0); + let requested = settings + .get(openshell_core::settings::OCSF_TRAFFIC_INTERVAL_SECS_KEY) + .and_then(|es| es.value.as_ref()) + .and_then(|sv| sv.value.as_ref()) + .and_then(|v| match v { + setting_value::Value::IntValue(v) => u64::try_from(*v).ok(), + _ => None, + }) + .unwrap_or(0); + let effective = openshell_supervisor_network::traffic::set_interval_secs(requested); + if APPLIED.swap(effective, Ordering::Relaxed) != effective { + info!( + ocsf_traffic_interval_secs = effective, + "OCSF traffic interval changed" + ); + } +} + /// Extract a bool value from an effective setting, if present. fn extract_bool_setting( settings: &std::collections::HashMap, diff --git a/examples/egress-usage-detector/.gitignore b/examples/egress-usage-detector/.gitignore new file mode 100644 index 0000000000..76a95043ac --- /dev/null +++ b/examples/egress-usage-detector/.gitignore @@ -0,0 +1,3 @@ +# Spike: no tracked lockfile, so the PR diff stays comparable. +Cargo.lock +target/ diff --git a/examples/egress-usage-detector/Cargo.toml b/examples/egress-usage-detector/Cargo.toml new file mode 100644 index 0000000000..7cbe2f1f37 --- /dev/null +++ b/examples/egress-usage-detector/Cargo.toml @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[workspace] + +[package] +name = "openshell-egress-usage-detector" +description = "Example external detector that reacts to OpenShell OCSF egress events" +version = "0.0.0" +edition = "2024" +rust-version = "1.94" +license = "Apache-2.0" +publish = false + +[dependencies] +clap = { version = "4.5", features = ["derive"] } +openshell-ocsf = { path = "../../crates/openshell-ocsf" } +opentelemetry-proto = { version = "0.33", default-features = false, features = ["gen-tonic", "logs"] } +serde_json = "1" +tokio = { version = "1.43", features = ["macros", "process", "rt-multi-thread", "sync"] } +tonic = { version = "0.14", features = ["gzip", "transport"] } + +[[bin]] +name = "egress-usage-detector" +path = "src/main.rs" diff --git a/examples/egress-usage-detector/README.md b/examples/egress-usage-detector/README.md new file mode 100644 index 0000000000..b7f1fe499a --- /dev/null +++ b/examples/egress-usage-detector/README.md @@ -0,0 +1,204 @@ + + + +# Egress usage detector (spike 01b) + +An example external detector for sandbox egress. It receives OCSF events as +OTLP/gRPC log records, keeps sliding windows in memory, and reacts through the +`openshell` CLI when a rule trips. + +Supervisors report bytes per upstream connection as OCSF Network Activity +events: a `Close` event for every connection with `cumulative_traffic` totals, +and, when the `ocsf_traffic_interval_secs` setting is non-zero, a `Traffic` +event per interval for connections that moved bytes. Each event's `traffic` +holds the bytes since the connection's previous event, so sums never double +count. + +## Layout + +- `src/detect.rs`: the rules, with no I/O and no event format. Callers pick the + keys and pass the clock, so the same rules can sit behind OCSF events (here) + or middleware hooks (spike 02b, `examples/supervisor-middleware-egress-usage`). + - `VolumeRule`: bytes per key over a sliding window (D2). + - `DistinctRule`: distinct members per group over a window (D1). +- `src/main.rs`: OTLP logs receiver. It parses each record body into + `openshell_ocsf::OcsfEvent` and feeds the rules. +- `src/react.rs`: reactions through the CLI. +- `observability/`: collector fan-out to this detector and Grafana LGTM. + +## Rules and reactions + +| Rule | Input | Key | Trips when | `--reaction deny` | `--reaction stop` | +|---|---|---|---|---|---| +| Volume (D2) | Network Activity Traffic (6) and Close (2) | sandbox name, policy key | `traffic` bytes in `--window` exceed `--max-bytes` | `policy update --remove-rule ` | `sandbox stop ` | +| Writers (D1) | HTTP Activity (4002), method PUT | host | distinct sandboxes in `--writers-window` exceed `--max-writers` | per writer: `policy update --add-deny host:port:PUT:/** --rule-name ` plus the rule's binaries | `sandbox stop` per writer | + +- Removing the policy key also cuts in-flight connections: the policy + generation guard closes every relay once the supervisor loads the new + revision. +- Writers needs `protocol: rest` endpoints; only inspected endpoints emit HTTP + Activity. +- A tripped key or host is ignored for `--cooldown` (default 300s). +- `--direction in|out|both` picks which bytes count. `--policy` limits Volume + to some policy keys. `--writers-host` limits Writers to one host. +- `--dry-run` logs reactions without calling the CLI. `--verbose` logs every + traffic event. +- Each reaction logs `detect->react` (rule trip to CLI exit) and + `supervisor event->react` (the event's `end_time` to CLI exit). + +## Run + +Run a gateway with push config delivery (`config_delivery_mode = "push"`); +with the default polling, a deny takes effect only at the next settings poll, +5 to 14 seconds later. `observability/run-gateway.sh` writes such a gateway. + +```shell +cd examples/egress-usage-detector +cargo build +target/debug/egress-usage-detector --gateway spike-01b --reaction deny \ + --max-bytes 20MB --window 60s +openshell --gateway spike-01b settings set --global --yes --key ocsf_traffic_interval_secs --value 5 +``` + +Point the gateway's OCSF OTLP export (`[openshell.gateway.otlp] endpoint` plus +`[openshell.gateway.otlp.ocsf] enabled = true`) at the detector's `--listen` +address (default `127.0.0.1:14317`), or at the collector in `observability/`, +which forwards Network Activity and HTTP Activity to the detector. + +## Measured locally + +Docker driver, push delivery, interval 5 s, through the collector, 100 MB +download at about 4 MB/s, threshold 20 MB / 60 s: + +| Step | Result | +|---|---| +| Trip | second Traffic event, 33.0 MB in the window, event lag 596 ms | +| Reaction | `--remove-rule` returned 39 ms after the trip, 636 ms after the supervisor event | +| Cut | relay closed about 0.6 s later; 34.4 MB delivered | +| New connection | denied (`PermissionError 13`) | +| Writers | one PUT with `--max-writers 0`: tripped, deny appended with `--binary /**`, 525 ms after the event | + +## Shortcuts taken in this spike + +The OpenShell implementation is unchanged by these notes. They record where the +spike took a shortcut and what we would do instead. + +### Routing traffic events by tracing target + +Traffic and Close events use a separate tracing target, `ocsf_traffic`. The +JSONL file layer and the OTLP layer accept it; the shorthand console and file, +the stderr OCSF-JSON console, and gateway log push (`openshell logs`) skip it. +This keeps per-connection volume out of the places people read. + +The general version is a delivery class chosen at emit time, independent of +OCSF severity and of the tracing level: + +| Class | Meaning | Sinks by default | +|---|---|---| +| `audit` (default) | security decisions and lifecycle | all | +| `telemetry` | high-volume accounting | JSONL file, OTLP | + +Each sink declares the classes it accepts, which gives one place to configure +later. The class can stay a tracing target, since targets are static metadata +that per-layer filters match cheaply. + +Alternatives we rejected: + +- OCSF `severity_id = 99` ("Other") as the switch. In OCSF it means "a severity + outside the enum", not "low priority", and SIEM rules that filter on + `severity_id` would misread routine events. +- The tracing level (`debug!`). `--log-level debug` drives the console filter, + so debugging a supervisor would print every Traffic event. + +The delivery class could still set the OTLP severity: `telemetry` records +exported as OTLP `DEBUG` show up as debug in Loki and Grafana, while the OCSF +body keeps `severity_id = 1`. + +### OTLP `severity_text` carries OCSF labels + +The OTLP mapping sets `severity_text` to the OCSF label (`Informational`, +`Low`, `Medium`, ...). Loki copies it verbatim into `detected_level`, and +Grafana does not recognize `Low` or `Medium` as log levels. Setting +`severity_text` to the OTel name of the band already in `severity_number` +(`INFO`, `WARN`, `ERROR`, `FATAL`) fixes level coloring and filters; the exact +OCSF level stays in `ocsf.severity_id`. + +### Attributes on the OTLP record + +The OTLP record carries the OCSF JSON as its body plus a few routing +attributes for every OCSF class: `ocsf.class_uid`, `ocsf.activity_id`, +`ocsf.severity_id`, `ocsf.metadata.uid`, and `openshell.sandbox.id`. An earlier +revision of this spike also promoted `destination.*`, `openshell.policy`, +`ocsf.connection_info.uid`, and `ocsf.traffic.bytes_*`. Those served only the +Prometheus path below, so they were removed. Consumers that need other fields +can: + +- read the body, as this detector does; +- extract fields in the collector with OTTL, for example + `merge_maps(log.cache, ParseJSON(log.body), "upsert")` followed by `set` + statements, scoped with `conditions` to the classes they need; +- use LogQL `| json` in Loki. + +A supervisor-side "promote these fields" setting would duplicate collector +extraction on every supervisor and add configuration that must reach them. +Sending the body as an OTLP map instead of a JSON string would let OTTL and +receivers address fields without parsing, but it changes the wire contract for +consumers that read the string. + +### Prometheus versus Loki + +An earlier revision derived `openshell_egress_bytes_{in,out}` counters in the +collector (`signaltometrics` plus `deltatocumulative`). It was removed: + +- A new series has no earlier sample, so `increase()` lost the first delta of + every series and undercounted short bursts. +- Running totals lived in collector memory and reset with the collector. +- Labels were limited to sandbox id, policy, and destination, and + `destination_address` has unbounded cardinality. + +Loki sums the per-event deltas exactly (`sum_over_time(... | json ... | unwrap +...)`) and reaches every OCSF field, which the dashboard now uses. Prometheus +remains the better store for long budgets and baselines (hourly or daily totals, +"10x the weekly average"); the counters should then come from the supervisor as +cumulative values, not be derived from events in the collector. + +### Grafana alerts + +Grafana alert rules could run Volume and Writers on Loki, plus deny spikes and +long-window budgets. They would be a second, slower layer, not a replacement: +rules evaluate every 10 s to 1 min, so alerts land 15 to 75 s after a crossing, +and a contact point can only call a webhook, so blocking still needs code that +turns alert labels into CLI calls. + +### Measurement + +- Counts are upstream socket bytes: transport bytes, including TLS overhead and + HTTP framing, after credential injection, not split per HTTP request. +- The proxy reads ahead into its buffers, so a connection counts what left the + upstream, not what the workload consumed. A client that read 2 MB and closed + counted about 4.7 MB. +- `actor.process.pid` is 0 when the decision has no PID, and IP-literal hosts + land in `dst_endpoint.domain`. + +### Supervisor and transport + +- Dedicated supervisor only. A shared multi-sandbox supervisor would emit with + the right context but share one interval setting. +- No registry cap or event volume limit. Traffic events share the supervisor's + OTLP queue (10k records or 16 MiB) with security events, so a large fleet at + a short interval can push security events into gap drops. +- No OCSF schema-validation tests and no published docs for the new fields or + the setting. + +### Detector + +- Reactions shell out to the CLI instead of calling the gateway API through + `openshell-sdk`. +- State is in memory and lost on restart; deduplication remembers the last + 100,000 `metadata.uid` values. +- The receiver implements only the OTLP logs service, so trace export pointed + at it fails with `UNIMPLEMENTED`. It accepts gzip, which the collector's + exporter uses by default. +- A Writers deny on an IP-literal REST endpoint was not enforced in the earlier + evaluation; it works for hostname endpoints such as `host.openshell.internal`. + Not investigated. diff --git a/examples/egress-usage-detector/observability/README.md b/examples/egress-usage-detector/observability/README.md new file mode 100644 index 0000000000..befd656084 --- /dev/null +++ b/examples/egress-usage-detector/observability/README.md @@ -0,0 +1,57 @@ + + + +# Collector fan-out: detector + Grafana + +A local OpenTelemetry Collector receives OCSF records and spans from the +gateway and every sandbox supervisor, then fans them out: + +``` +supervisors ──OTLP──► collector :4317 ─┬─ logs (class 4001/4002 only) ──► detector :14317 ──► openshell CLI +gateway ──OTLP──► └─ logs, traces ─────────────────► Grafana LGTM (Loki, Tempo) +``` + +- `compose.yaml` runs the collector (host network, so Docker supervisors reach + `127.0.0.1:4317`) and `grafana/otel-lgtm` (Grafana on `127.0.0.1:3000`, its + own OTLP receiver on `127.0.0.1:24317`). +- `collector.yaml` filters on the `ocsf.class_uid` attribute for the detector + and sends everything to LGTM. It does not parse the OCSF body. +- `run-gateway.sh` starts a Docker gateway with OTLP export to the collector + and push config delivery. +- `dashboard.json` sums the `traffic` deltas from Loki per sandbox name and + policy key, and lists the raw events. + +## Run + +```shell +examples/egress-usage-detector/observability/up.sh +(cd examples/egress-usage-detector && cargo build) +examples/egress-usage-detector/target/debug/egress-usage-detector \ + --openshell "$PWD/target/debug/openshell" --gateway spike-01b --reaction deny +examples/egress-usage-detector/observability/run-gateway.sh +openshell --gateway spike-01b settings set --global --yes --key ocsf_traffic_interval_secs --value 5 +``` + +Open http://127.0.0.1:3000/d/openshell-egress. Anonymous access is enabled as +Admin, which is acceptable for local use only. Tear down with +`docker compose -f examples/egress-usage-detector/observability/compose.yaml down`. + +The collector queues and retries per exporter, so a stopped detector does not +block Grafana, and the reverse is also true. + +## Useful queries + +Bytes per sandbox and policy key over five minutes: + +```logql +sum by (sandbox, policy) (sum_over_time( + {service_name="openshell-supervisor"} | ocsf_class_uid="4001" | ocsf_activity_id=~"2|6" + | json bytes="traffic.bytes", sandbox="container.name", policy="firewall_rule.name" + | unwrap bytes [5m])) +``` + +Raw Traffic and Close events: +`{service_name="openshell-supervisor"} | ocsf_class_uid="4001" | ocsf_activity_id=~"2|6"`. + +See the detector README for why there is no Prometheus path and for the other +shortcuts in this setup. diff --git a/examples/egress-usage-detector/observability/collector.yaml b/examples/egress-usage-detector/observability/collector.yaml new file mode 100644 index 0000000000..d5efa698d8 --- /dev/null +++ b/examples/egress-usage-detector/observability/collector.yaml @@ -0,0 +1,45 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +receivers: + otlp: + protocols: + grpc: + endpoint: 127.0.0.1:4317 + +processors: + batch: + timeout: 200ms + # Only OCSF Network Activity and HTTP Activity records go to the detector. + filter/network: + error_mode: ignore + logs: + log_record: + - 'attributes["ocsf.class_uid"] != 4001 and attributes["ocsf.class_uid"] != 4002' + +exporters: + otlp/detector: + endpoint: 127.0.0.1:14317 + tls: + insecure: true + sending_queue: + queue_size: 10000 + otlp/lgtm: + endpoint: 127.0.0.1:24317 + tls: + insecure: true + +service: + pipelines: + logs/detector: + receivers: [otlp] + processors: [filter/network, batch] + exporters: [otlp/detector] + logs/lgtm: + receivers: [otlp] + processors: [batch] + exporters: [otlp/lgtm] + traces/lgtm: + receivers: [otlp] + processors: [batch] + exporters: [otlp/lgtm] diff --git a/examples/egress-usage-detector/observability/compose.yaml b/examples/egress-usage-detector/observability/compose.yaml new file mode 100644 index 0000000000..5620eb7539 --- /dev/null +++ b/examples/egress-usage-detector/observability/compose.yaml @@ -0,0 +1,32 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# OTel Collector that fans OpenShell OCSF records out to the egress detector +# and to a local Grafana LGTM stack (Grafana, Loki, Tempo, Prometheus). +# +# Docker sandbox supervisors run with host networking and export to the +# gateway's OCSF endpoint, so the collector uses host networking too and +# listens on 127.0.0.1:4317. The detector listens on 127.0.0.1:14317. + +name: openshell-egress-obs + +services: + collector: + image: otel/opentelemetry-collector-contrib:0.139.0 + network_mode: host + command: ["--config=/etc/otelcol/config.yaml"] + volumes: + - ./collector.yaml:/etc/otelcol/config.yaml:ro + depends_on: + - lgtm + restart: unless-stopped + + lgtm: + image: grafana/otel-lgtm:0.11.10 + ports: + - "127.0.0.1:3000:3000" # Grafana (admin / admin) + - "127.0.0.1:24317:4317" # LGTM's own OTLP/gRPC receiver + environment: + GF_AUTH_ANONYMOUS_ENABLED: "true" + GF_AUTH_ANONYMOUS_ORG_ROLE: Admin + restart: unless-stopped diff --git a/examples/egress-usage-detector/observability/dashboard.json b/examples/egress-usage-detector/observability/dashboard.json new file mode 100644 index 0000000000..f9c7c6de36 --- /dev/null +++ b/examples/egress-usage-detector/observability/dashboard.json @@ -0,0 +1,111 @@ +{ + "uid": "openshell-egress", + "title": "OpenShell egress (spike 01b)", + "refresh": "5s", + "time": { + "from": "now-15m", + "to": "now" + }, + "schemaVersion": 39, + "panels": [ + { + "type": "timeseries", + "title": "bytes_in per minute by sandbox and policy key", + "gridPos": { + "x": 0, + "y": 0, + "w": 12, + "h": 9 + }, + "datasource": { + "type": "loki", + "uid": "loki" + }, + "fieldConfig": { + "defaults": { + "unit": "decbytes" + } + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (sandbox, policy) (sum_over_time({service_name=\"openshell-supervisor\"} | ocsf_class_uid=\"4001\" | ocsf_activity_id=~\"2|6\" | json bytes=\"traffic.bytes_in\", sandbox=\"container.name\", policy=\"firewall_rule.name\" | unwrap bytes [1m]))", + "legendFormat": "{{sandbox}} / {{policy}}" + } + ] + }, + { + "type": "timeseries", + "title": "bytes_out per minute by sandbox and policy key", + "gridPos": { + "x": 12, + "y": 0, + "w": 12, + "h": 9 + }, + "datasource": { + "type": "loki", + "uid": "loki" + }, + "fieldConfig": { + "defaults": { + "unit": "decbytes" + } + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (sandbox, policy) (sum_over_time({service_name=\"openshell-supervisor\"} | ocsf_class_uid=\"4001\" | ocsf_activity_id=~\"2|6\" | json bytes=\"traffic.bytes_out\", sandbox=\"container.name\", policy=\"firewall_rule.name\" | unwrap bytes [1m]))", + "legendFormat": "{{sandbox}} / {{policy}}" + } + ] + }, + { + "type": "stat", + "title": "bytes (in + out), last 5 min, by sandbox and policy key", + "gridPos": { + "x": 0, + "y": 9, + "w": 24, + "h": 5 + }, + "datasource": { + "type": "loki", + "uid": "loki" + }, + "fieldConfig": { + "defaults": { + "unit": "decbytes" + } + }, + "targets": [ + { + "refId": "A", + "queryType": "instant", + "expr": "sum by (sandbox, policy) (sum_over_time({service_name=\"openshell-supervisor\"} | ocsf_class_uid=\"4001\" | ocsf_activity_id=~\"2|6\" | json bytes=\"traffic.bytes\", sandbox=\"container.name\", policy=\"firewall_rule.name\" | unwrap bytes [5m]))", + "legendFormat": "{{sandbox}} / {{policy}}" + } + ] + }, + { + "type": "logs", + "title": "OCSF Traffic and Close events", + "gridPos": { + "x": 0, + "y": 14, + "w": 24, + "h": 12 + }, + "datasource": { + "type": "loki", + "uid": "loki" + }, + "targets": [ + { + "refId": "A", + "expr": "{service_name=\"openshell-supervisor\"} | ocsf_class_uid=\"4001\" | ocsf_activity_id=~\"2|6\"" + } + ] + } + ] +} diff --git a/examples/egress-usage-detector/observability/dashboard.json.license b/examples/egress-usage-detector/observability/dashboard.json.license new file mode 100644 index 0000000000..817df24424 --- /dev/null +++ b/examples/egress-usage-detector/observability/dashboard.json.license @@ -0,0 +1,2 @@ +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 diff --git a/examples/egress-usage-detector/observability/run-gateway.sh b/examples/egress-usage-detector/observability/run-gateway.sh new file mode 100755 index 0000000000..d13aabca18 --- /dev/null +++ b/examples/egress-usage-detector/observability/run-gateway.sh @@ -0,0 +1,104 @@ +#!/usr/bin/env bash + +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Run a standalone Docker-backed gateway whose supervisors export OCSF records +# (and spans) to the collector in compose.yaml on 127.0.0.1:4317. +# +# Unlike `mise run gateway:docker`, the OTLP tables are always written: that +# task probes the port with lsof, which cannot see the root-owned collector. +# +# Requires the supervisor and sandbox images named below; build them with +# CONTAINER_ENGINE=docker IMAGE_TAG=spike-01b mise run build:docker:supervisor +# CONTAINER_ENGINE=docker IMAGE_TAG=spike-01b mise run build:docker:sandbox + +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +PORT="${OPENSHELL_SERVER_PORT:-18081}" +GATEWAY_NAME="${OPENSHELL_DOCKER_GATEWAY_NAME:-spike-01b}" +STATE_DIR="${OPENSHELL_DOCKER_GATEWAY_STATE_DIR:-${ROOT}/.cache/gateway-${GATEWAY_NAME}}" +SANDBOX_LABEL="${OPENSHELL_SANDBOX_NAMESPACE:-${GATEWAY_NAME}}" +SUPERVISOR_IMAGE="${OPENSHELL_SUPERVISOR_IMAGE:-openshell/supervisor:spike-01b}" +SANDBOX_RUNTIME_IMAGE="${OPENSHELL_SANDBOX_RUNTIME_IMAGE:-openshell/sandbox:spike-01b}" +COLLECTOR="${OPENSHELL_OTLP_COLLECTOR:-http://127.0.0.1:4317}" +GATEWAY_BIN="${ROOT}/target/debug/openshell-gateway" + +for image in "${SUPERVISOR_IMAGE}" "${SANDBOX_RUNTIME_IMAGE}"; do + if ! docker image inspect "${image}" >/dev/null 2>&1; then + echo "ERROR: image ${image} not found; see the header of this script" >&2 + exit 1 + fi +done + +cargo build -p openshell-gateway --bin openshell-gateway + +TLS_DIR="${STATE_DIR}/tls" +"${GATEWAY_BIN}" generate-certs \ + --output-dir "${TLS_DIR}" \ + --server-san "127.0.0.1" \ + --server-san "localhost" \ + --server-san "host.openshell.internal" + +mkdir -p "${STATE_DIR}" +CONFIG_PATH="${STATE_DIR}/gateway.toml" +cat >"${CONFIG_PATH}" <"${GATEWAY_DIR}/metadata.json" </dev/null && break + sleep 2 +done +python3 -c 'import json,sys; print(json.dumps({"dashboard": json.load(open("dashboard.json")), "overwrite": True}))' | + curl -sf -X POST -H 'Content-Type: application/json' --data-binary @- \ + http://127.0.0.1:3000/api/dashboards/db >/dev/null +echo "Grafana: http://127.0.0.1:3000/d/openshell-egress" diff --git a/examples/egress-usage-detector/src/detect.rs b/examples/egress-usage-detector/src/detect.rs new file mode 100644 index 0000000000..fcda116e03 --- /dev/null +++ b/examples/egress-usage-detector/src/detect.rs @@ -0,0 +1,177 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Detection rules with no I/O and no event format. +//! +//! Callers choose the keys and pass the clock, so the same rules can run +//! behind OCSF events (this detector) or middleware hooks (spike 02b). + +use std::collections::{HashMap, VecDeque}; +use std::hash::Hash; +use std::time::{Duration, Instant}; + +/// Byte samples summed over a sliding window. +#[derive(Debug, Default)] +pub struct SlidingWindow { + samples: VecDeque<(Instant, u64)>, + total: u64, +} + +impl SlidingWindow { + /// Add `bytes` at `now`, drop samples older than `window`, and return the + /// window total. + pub fn add(&mut self, now: Instant, bytes: u64, window: Duration) -> u64 { + while let Some(&(at, bytes)) = self.samples.front() { + if now.duration_since(at) <= window { + break; + } + self.samples.pop_front(); + self.total -= bytes; + } + if bytes > 0 { + self.samples.push_back((now, bytes)); + self.total += bytes; + } + self.total + } +} + +/// Trips a key once its window total exceeds `max_bytes`, then ignores the +/// key for `cooldown`. +#[derive(Debug)] +pub struct VolumeRule { + max_bytes: u64, + window: Duration, + cooldown: Duration, + windows: HashMap, + tripped_until: HashMap, +} + +impl VolumeRule { + pub fn new(max_bytes: u64, window: Duration, cooldown: Duration) -> Self { + Self { + max_bytes, + window, + cooldown, + windows: HashMap::new(), + tripped_until: HashMap::new(), + } + } + + /// Add `bytes` to the key's window. Returns the window total when the key + /// trips. + pub fn record(&mut self, key: &K, bytes: u64, now: Instant) -> Option { + let total = self + .windows + .entry(key.clone()) + .or_default() + .add(now, bytes, self.window); + if total <= self.max_bytes + || self + .tripped_until + .get(key) + .is_some_and(|until| *until > now) + { + return None; + } + self.tripped_until.insert(key.clone(), now + self.cooldown); + Some(total) + } +} + +/// Trips a group once more than `max_members` distinct members were seen in +/// it within `window`, then ignores the group for `cooldown`. +#[derive(Debug)] +pub struct DistinctRule { + max_members: usize, + window: Duration, + cooldown: Duration, + groups: HashMap>, + tripped_until: HashMap, +} + +impl DistinctRule { + pub fn new(max_members: usize, window: Duration, cooldown: Duration) -> Self { + Self { + max_members, + window, + cooldown, + groups: HashMap::new(), + tripped_until: HashMap::new(), + } + } + + /// Record `member` in `group`. Returns the sorted members when the group + /// trips. + pub fn record(&mut self, group: &G, member: &M, now: Instant) -> Option> { + let members = self.groups.entry(group.clone()).or_default(); + members.insert(member.clone(), now); + members.retain(|_, seen| now.duration_since(*seen) <= self.window); + if members.len() <= self.max_members + || self + .tripped_until + .get(group) + .is_some_and(|until| *until > now) + { + return None; + } + let mut tripped: Vec = members.keys().cloned().collect(); + tripped.sort(); + self.tripped_until + .insert(group.clone(), now + self.cooldown); + Some(tripped) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const MINUTE: Duration = Duration::from_secs(60); + + #[test] + fn window_expires_old_samples() { + let mut window = SlidingWindow::default(); + let start = Instant::now(); + assert_eq!(window.add(start, 10, MINUTE), 10); + assert_eq!(window.add(start + Duration::from_secs(30), 5, MINUTE), 15); + assert_eq!(window.add(start + Duration::from_secs(61), 1, MINUTE), 6); + } + + #[test] + fn volume_trips_once_per_cooldown() { + let mut rule = VolumeRule::new(100, MINUTE, Duration::from_secs(300)); + let start = Instant::now(); + let key = ("sb".to_string(), "pkg".to_string()); + assert_eq!(rule.record(&key, 60, start), None); + assert_eq!(rule.record(&key, 60, start), Some(120)); + assert_eq!(rule.record(&key, 60, start), None); + assert_eq!( + rule.record(&key, 200, start + Duration::from_secs(301)), + Some(200) + ); + } + + #[test] + fn distinct_counts_members_within_the_window() { + let mut rule = DistinctRule::new(2, MINUTE, Duration::from_secs(300)); + let start = Instant::now(); + let host = "files".to_string(); + assert_eq!(rule.record(&host, &"a", start), None); + assert_eq!(rule.record(&host, &"b", start), None); + assert_eq!(rule.record(&host, &"a", start), None); + // "a" and "b" expire before "c" arrives. + assert_eq!( + rule.record(&host, &"c", start + Duration::from_secs(61)), + None + ); + assert_eq!( + rule.record(&host, &"d", start + Duration::from_secs(62)), + None + ); + assert_eq!( + rule.record(&host, &"e", start + Duration::from_secs(63)), + Some(vec!["c", "d", "e"]) + ); + } +} diff --git a/examples/egress-usage-detector/src/main.rs b/examples/egress-usage-detector/src/main.rs new file mode 100644 index 0000000000..d9dbc23dd3 --- /dev/null +++ b/examples/egress-usage-detector/src/main.rs @@ -0,0 +1,431 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Example external detector for OpenShell egress (spike 01b). +//! +//! Receives OCSF events as OTLP/gRPC log records, from supervisors directly +//! or through a collector, and reacts through the `openshell` CLI: +//! +//! - Volume (D2): bytes per sandbox and policy key over a sliding window, +//! from Network Activity Traffic (6) and Close (2) events. Each event's +//! `traffic` is the delta since the connection's previous report, so sums +//! never double count. +//! - Writers (D1): distinct sandboxes sending HTTP PUT to one host over a +//! window, from HTTP Activity (4002) events. + +mod detect; +mod react; + +use std::collections::{HashSet, VecDeque}; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use clap::Parser; +use openshell_ocsf::{HttpActivityEvent, HttpMethod, NetworkActivityEvent, OcsfEvent}; +use opentelemetry_proto::tonic::collector::logs::v1::logs_service_server::{ + LogsService, LogsServiceServer, +}; +use opentelemetry_proto::tonic::collector::logs::v1::{ + ExportLogsServiceRequest, ExportLogsServiceResponse, +}; +use opentelemetry_proto::tonic::common::v1::any_value::Value; +use tonic::codec::CompressionEncoding; +use tonic::transport::Server; +use tonic::{Request, Response, Status}; + +use crate::detect::{DistinctRule, VolumeRule}; +use crate::react::{Reaction, Reactor, Trip}; + +const ACTIVITY_CLOSE: u8 = 2; +const ACTIVITY_TRAFFIC: u8 = 6; +/// OTLP export is at least once; remember this many event uids. +const SEEN_UIDS: usize = 100_000; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)] +enum Direction { + /// bytes_in plus bytes_out. + Both, + /// bytes_in only (downloads). + In, + /// bytes_out only (uploads). + Out, +} + +#[derive(Debug, Parser)] +#[command(about = "Run the example OpenShell egress-usage detector")] +struct Cli { + /// OTLP/gRPC listen address. + #[arg(long, default_value = "127.0.0.1:14317")] + listen: SocketAddr, + #[arg(long, value_enum, default_value_t = Reaction::Deny)] + reaction: Reaction, + /// `openshell` CLI binary. + #[arg(long, default_value = "openshell")] + openshell: String, + /// Gateway name passed to the CLI as `--gateway`. + #[arg(long)] + gateway: Option, + /// Workspace of the sandboxes (`OPENSHELL_WORKSPACE`). + #[arg(long)] + workspace: Option, + /// Log reactions without calling the CLI. + #[arg(long)] + dry_run: bool, + /// Log every traffic event. + #[arg(long)] + verbose: bool, + /// How long a tripped key or host is ignored, such as 300s. + #[arg(long, default_value = "300s", value_parser = parse_duration)] + cooldown: Duration, + + /// Volume: bytes allowed per window per sandbox and policy key, such as 20MB. + #[arg(long, default_value = "20MB", value_parser = parse_bytes)] + max_bytes: u64, + /// Volume: sliding window length, such as 60s. + #[arg(long, default_value = "60s", value_parser = parse_duration)] + window: Duration, + /// Volume: which bytes count toward the window. + #[arg(long, value_enum, default_value_t = Direction::Both)] + direction: Direction, + /// Volume: only watch these policy keys (repeatable). + #[arg(long)] + policy: Vec, + + /// Writers: trip when distinct PUT writers to one host exceed this. + #[arg(long, default_value_t = 5)] + max_writers: usize, + /// Writers: sliding window length, such as 300s. + #[arg(long, default_value = "300s", value_parser = parse_duration)] + writers_window: Duration, + /// Writers: only watch this host. + #[arg(long)] + writers_host: Option, +} + +fn parse_bytes(value: &str) -> Result { + let upper = value.trim().to_ascii_uppercase(); + let (digits, multiplier) = [ + ("GB", 1_000_000_000), + ("MB", 1_000_000), + ("KB", 1_000), + ("B", 1), + ] + .iter() + .find_map(|(suffix, multiplier)| { + upper + .strip_suffix(suffix) + .map(|digits| (digits.trim().to_string(), *multiplier)) + }) + .unwrap_or((upper.clone(), 1)); + digits + .parse::() + .map(|value| value * multiplier) + .map_err(|_| format!("invalid byte size '{value}'")) +} + +fn parse_duration(value: &str) -> Result { + let value = value.trim(); + let (digits, unit) = value.split_at( + value + .find(|c: char| !c.is_ascii_digit()) + .unwrap_or(value.len()), + ); + let number: u64 = digits + .parse() + .map_err(|_| format!("invalid duration '{value}'"))?; + match unit { + "" | "s" => Ok(Duration::from_secs(number)), + "ms" => Ok(Duration::from_millis(number)), + "m" => Ok(Duration::from_secs(number * 60)), + _ => Err(format!("invalid duration '{value}', use ms, s, or m")), + } +} + +fn now_ms() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |elapsed| { + i64::try_from(elapsed.as_millis()).unwrap_or(i64::MAX) + }) +} + +/// Print `message` with a UTC `HH:MM:SS.mmm` prefix. +fn log(message: &str) { + let ms = now_ms(); + let day_ms = ms.rem_euclid(86_400_000); + println!( + "{:02}:{:02}:{:02}.{:03} {message}", + day_ms / 3_600_000, + day_ms / 60_000 % 60, + day_ms / 1000 % 60, + day_ms % 1000 + ); +} + +/// Sandbox name, which the CLI takes, falling back to its id. +fn sandbox_of(event: &OcsfEvent) -> String { + event + .base() + .container + .as_ref() + .map(|container| { + if container.name.is_empty() { + container.uid.clone().unwrap_or_default() + } else { + container.name.clone() + } + }) + .unwrap_or_else(|| "-".into()) +} + +struct State { + seen: HashSet, + seen_order: VecDeque, + volume: VolumeRule<(String, String)>, + writers: DistinctRule, +} + +impl State { + /// Whether `uid` was seen before; remembers it otherwise. + fn duplicate(&mut self, uid: &str) -> bool { + if !self.seen.insert(uid.to_string()) { + return true; + } + self.seen_order.push_back(uid.to_string()); + if self.seen_order.len() > SEEN_UIDS + && let Some(oldest) = self.seen_order.pop_front() + { + self.seen.remove(&oldest); + } + false + } +} + +#[derive(Clone)] +struct Detector { + cli: Arc, + reactor: Reactor, + state: Arc>, +} + +impl Detector { + fn ingest(&self, event: &OcsfEvent) { + if let Some(uid) = &event.base().metadata.uid + && self.state.lock().unwrap().duplicate(uid) + { + return; + } + match event { + OcsfEvent::NetworkActivity(network) + if matches!(network.base.activity_id, ACTIVITY_TRAFFIC | ACTIVITY_CLOSE) => + { + self.on_traffic(event, network); + } + OcsfEvent::HttpActivity(http) => self.on_http(event, http), + _ => {} + } + } + + fn on_traffic(&self, event: &OcsfEvent, network: &NetworkActivityEvent) { + let Some(traffic) = network.traffic.traffic else { + return; + }; + let bytes = match self.cli.direction { + Direction::Both => traffic.bytes, + Direction::In => traffic.bytes_in, + Direction::Out => traffic.bytes_out, + }; + if bytes == 0 { + return; + } + let sandbox = sandbox_of(event); + let policy = network + .firewall_rule + .as_ref() + .map_or_else(|| "-".to_string(), |rule| rule.name.clone()); + if !self.cli.policy.is_empty() && !self.cli.policy.contains(&policy) { + return; + } + let now = Instant::now(); + let key = (sandbox, policy); + let Some(total) = self.state.lock().unwrap().volume.record(&key, bytes, now) else { + if self.cli.verbose { + log(&format!("traffic {}/{} +{bytes}", key.0, key.1)); + } + return; + }; + let event_ms = network.traffic.end_time.unwrap_or(network.base.time); + log(&format!( + "VOLUME TRIP sandbox={} policy={} window_bytes={total} > {} in {:?} (event lag {} ms)", + key.0, + key.1, + self.cli.max_bytes, + self.cli.window, + now_ms() - event_ms, + )); + let (sandbox, policy) = key; + self.reactor + .react(Trip::Volume { sandbox, policy }, now, event_ms); + } + + fn on_http(&self, event: &OcsfEvent, http: &HttpActivityEvent) { + let Some(request) = &http.http_request else { + return; + }; + if request.http_method != HttpMethod::Put { + return; + } + let url = request.url.as_ref(); + let host = url + .and_then(|url| url.hostname.clone()) + .or_else(|| { + http.dst_endpoint + .as_ref() + .and_then(|dst| dst.domain.clone()) + }) + .unwrap_or_else(|| "-".into()); + let port = url + .and_then(|url| url.port) + .or_else(|| http.dst_endpoint.as_ref().and_then(|dst| dst.port)); + if self + .cli + .writers_host + .as_ref() + .is_some_and(|only| *only != host) + { + return; + } + let sandbox = sandbox_of(event); + let policy = http + .firewall_rule + .as_ref() + .map_or_else(|| "-".to_string(), |rule| rule.name.clone()); + let now = Instant::now(); + let Some(writers) = self + .state + .lock() + .unwrap() + .writers + .record(&host, &sandbox, now) + else { + return; + }; + log(&format!( + "WRITERS TRIP host={host} distinct_put_writers={} > {} in {:?}: {}", + writers.len(), + self.cli.max_writers, + self.cli.writers_window, + writers.join(", "), + )); + for sandbox in writers { + self.reactor.react( + Trip::Writers { + sandbox, + policy: policy.clone(), + host: host.clone(), + port, + }, + now, + http.base.time, + ); + } + } +} + +#[tonic::async_trait] +impl LogsService for Detector { + async fn export( + &self, + request: Request, + ) -> Result, Status> { + let records = request + .into_inner() + .resource_logs + .into_iter() + .flat_map(|resource| resource.scope_logs) + .flat_map(|scope| scope.log_records); + for record in records { + let Some(Value::StringValue(body)) = record.body.and_then(|body| body.value) else { + continue; + }; + // The body is the OCSF JSON record. + if let Ok(event) = serde_json::from_str::(&body) { + self.ingest(&event); + } + } + Ok(Response::new(ExportLogsServiceResponse::default())) + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let cli = Cli::parse(); + let reactor = Reactor { + reaction: cli.reaction, + openshell: cli.openshell.clone(), + gateway: cli.gateway.clone(), + workspace: cli.workspace.clone(), + dry_run: cli.dry_run, + }; + let state = State { + seen: HashSet::new(), + seen_order: VecDeque::new(), + volume: VolumeRule::new(cli.max_bytes, cli.window, cli.cooldown), + writers: DistinctRule::new(cli.max_writers, cli.writers_window, cli.cooldown), + }; + log(&format!( + "listening on {}; reaction={:?} volume>{}B/{:?} ({:?}) writers>{}/{:?}", + cli.listen, + cli.reaction, + cli.max_bytes, + cli.window, + cli.direction, + cli.max_writers, + cli.writers_window, + )); + let listen = cli.listen; + let detector = Detector { + cli: Arc::new(cli), + reactor, + state: Arc::new(Mutex::new(state)), + }; + Server::builder() + // The collector's OTLP exporter compresses with gzip by default. + .add_service(LogsServiceServer::new(detector).accept_compressed(CompressionEncoding::Gzip)) + .serve(listen) + .await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A Close event as a supervisor exports it. + const CLOSE_EVENT: &str = r#"{"class_uid":4001,"class_name":"Network Activity","category_uid":4,"category_name":"Network Activity","activity_id":2,"activity_name":"Close","type_uid":400102,"type_name":"Network Activity: Close","time":1793658742379,"severity_id":1,"severity":"Informational","status_id":1,"status":"Success","message":"CLOSE 172.22.0.2:8080 in=1572864 out=0","metadata":{"version":"1.8.0","product":{"name":"OpenShell Sandbox Supervisor","vendor_name":"OpenShell","version":"0.1.7"},"profiles":["security_control","network_proxy","container","host"],"uid":"74983dd9-7dc3-47fd-bc6a-91d33f211418"},"device":{"hostname":"host","type_id":99,"type":"Sandbox","os":{"name":"Linux","type_id":200,"type":"Linux"}},"container":{"name":"sb-22","uid":"fa86f9e5-8fa1-41ca-9a63-d02de84d8821"},"dst_endpoint":{"domain":"172.22.0.2","port":8080},"proxy_endpoint":{"ip":"127.0.0.1","port":3128},"actor":{"process":{"name":"/usr/local/bin/python3.13","pid":0}},"firewall_rule":{"name":"pkg_proxy","type":"opa"},"connection_info":{"protocol_name":"tcp","direction_id":0,"direction":"Unknown","uid":"conn-f4ab6d8c-e159-4df9-8c43-79c9127a0bc2"},"action_id":1,"action":"Allowed","disposition_id":1,"disposition":"Allowed","is_src_dst_assignment_known":true,"traffic":{"bytes_in":1572864,"bytes_out":0,"bytes":1572864},"cumulative_traffic":{"bytes_in":28195424,"bytes_out":107,"bytes":28195531},"start_time":1793658736378,"end_time":1793658741992,"duration":5614}"#; + + #[test] + fn close_event_parses_into_typed_fields() { + let event: OcsfEvent = serde_json::from_str(CLOSE_EVENT).unwrap(); + let OcsfEvent::NetworkActivity(network) = &event else { + panic!("expected Network Activity, got {event:?}"); + }; + assert_eq!(network.base.activity_id, ACTIVITY_CLOSE); + assert_eq!(sandbox_of(&event), "sb-22"); + assert_eq!(network.firewall_rule.as_ref().unwrap().name, "pkg_proxy"); + assert_eq!(network.traffic.traffic.unwrap().bytes, 1_572_864); + assert_eq!( + network.traffic.cumulative_traffic.unwrap().bytes, + 28_195_531 + ); + assert_eq!(network.traffic.end_time, Some(1_793_658_741_992)); + } + + #[test] + fn sizes_and_durations_parse() { + assert_eq!(parse_bytes("20MB"), Ok(20_000_000)); + assert_eq!(parse_duration("300s"), Ok(Duration::from_secs(300))); + assert_eq!(parse_duration("5m"), Ok(Duration::from_secs(300))); + } +} diff --git a/examples/egress-usage-detector/src/react.rs b/examples/egress-usage-detector/src/react.rs new file mode 100644 index 0000000000..3c95893418 --- /dev/null +++ b/examples/egress-usage-detector/src/react.rs @@ -0,0 +1,172 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Reactions through the public `openshell` CLI. + +use std::process::Output; +use std::time::Instant; + +use tokio::process::Command; + +use crate::{log, now_ms}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)] +pub enum Reaction { + /// `openshell sandbox stop `. + Stop, + /// Remove the policy key (volume) or append a PUT deny rule (writers). + Deny, +} + +/// What tripped, and against which sandbox and policy key. +#[derive(Debug, Clone)] +pub enum Trip { + Volume { + sandbox: String, + policy: String, + }, + Writers { + sandbox: String, + policy: String, + host: String, + port: Option, + }, +} + +#[derive(Debug, Clone)] +pub struct Reactor { + pub reaction: Reaction, + pub openshell: String, + pub gateway: Option, + pub workspace: Option, + pub dry_run: bool, +} + +impl Reactor { + /// React to `trip` in the background. `event_ms` is the supervisor's event + /// time, used to report end-to-end latency. + pub fn react(&self, trip: Trip, tripped_at: Instant, event_ms: i64) { + let reactor = self.clone(); + tokio::spawn(async move { reactor.run(trip, tripped_at, event_ms).await }); + } + + async fn run(&self, trip: Trip, tripped_at: Instant, event_ms: i64) { + let (kind, sandbox, args) = match (&trip, self.reaction) { + (Trip::Volume { sandbox, .. } | Trip::Writers { sandbox, .. }, Reaction::Stop) => ( + "stop", + sandbox, + vec!["sandbox".into(), "stop".into(), sandbox.clone()], + ), + (Trip::Volume { sandbox, policy }, Reaction::Deny) => ( + "volume", + sandbox, + vec![ + "policy".into(), + "update".into(), + sandbox.clone(), + "--remove-rule".into(), + policy.clone(), + ], + ), + ( + Trip::Writers { + sandbox, + policy, + host, + port, + }, + Reaction::Deny, + ) => { + let target = port.map_or_else(|| host.clone(), |port| format!("{host}:{port}")); + let mut args = vec![ + "policy".into(), + "update".into(), + sandbox.clone(), + "--add-deny".into(), + format!("{target}:PUT:/**"), + "--rule-name".into(), + policy.clone(), + ]; + args.extend(self.binary_scope(sandbox, policy).await); + ("writers", sandbox, args) + } + }; + if self.dry_run { + log(&format!("DRY RUN openshell {}", args.join(" "))); + return; + } + let output = self.cli(&args).await; + let status = match &output { + Ok(output) if output.status.success() => "ok".to_string(), + Ok(output) => format!("{}", output.status), + Err(error) => error.to_string(), + }; + log(&format!( + "REACT {kind} {:?} sandbox={sandbox} [{status}] detect->react {} ms, \ + supervisor event->react {} ms: {}", + self.reaction, + tripped_at.elapsed().as_millis(), + now_ms() - event_ms, + args.join(" "), + )); + if let Ok(output) = output + && !output.status.success() + { + log(&format!( + " stderr: {}", + String::from_utf8_lossy(&output.stderr).trim() + )); + } + } + + /// CLI flags naming the rule's complete binary list. + /// + /// An L7 append must restate the target rule's binaries exactly, so read + /// them from the sandbox's base policy. A rule without binaries allows any + /// binary. + async fn binary_scope(&self, sandbox: &str, policy: &str) -> Vec { + let args: Vec = ["policy", "get", sandbox, "--base", "-o", "json"] + .map(String::from) + .into(); + let paths = match self.cli(&args).await { + Ok(output) => serde_json::from_slice::(&output.stdout) + .ok() + .and_then(|policy_json| { + let rule = &policy_json["policy"]["network_policies"][policy]; + (!rule.is_null()).then(|| { + rule["binaries"] + .as_array() + .into_iter() + .flatten() + .filter_map(|binary| binary["path"].as_str().map(String::from)) + .collect::>() + }) + }), + Err(_) => None, + }; + match paths { + Some(paths) if !paths.is_empty() => paths + .into_iter() + .flat_map(|path| ["--binary".to_string(), path]) + .collect(), + Some(_) => vec!["--any-binary".into()], + None => { + log(&format!( + " could not read binaries of {sandbox}/{policy}; using --any-binary" + )); + vec!["--any-binary".into()] + } + } + } + + async fn cli(&self, args: &[String]) -> std::io::Result { + let mut command = Command::new(&self.openshell); + if let Some(gateway) = &self.gateway { + command.args(["--gateway", gateway]); + } + if let Some(workspace) = &self.workspace { + command.env("OPENSHELL_WORKSPACE", workspace); + } + command.args(args).output().await + } +}