feat(supervisor): OCSF egress traffic events with external detector (spike)

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 {

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<String>,
     unmapped: Option<serde_json::Map<String, serde_json::Value>>,
     log_source: Option<String>,
+    traffic: TrafficReport,
     endpoint_state: std::marker::PhantomData<EndpointState>,
 }
 
@@ -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,
         })
     }
 }

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<bool>,
+
+    /// 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();

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()
     }

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,
 };

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<String>,
 }
 
 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<String>) -> Self {
+        self.uid = Some(uid.into());
+        self
+    }
 }
 
 #[cfg(test)]

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;

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<NetworkTraffic>,
+    /// Totals over the life of the connection.
+    #[serde(default)]
+    pub cumulative_traffic: Option<NetworkTraffic>,
+    /// Window start, milliseconds since epoch.
+    #[serde(default)]
+    pub start_time: Option<i64>,
+    /// Window end, milliseconds since epoch.
+    #[serde(default)]
+    pub end_time: Option<i64>,
+    /// Connection duration in milliseconds (close events).
+    #[serde(default)]
+    pub duration: Option<i64>,
+}

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.

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;
         }
 

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;