diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 0e4c74af95..e46b2ae8ae 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -56,7 +56,7 @@ use openshell_core::{ settings::{self, SettingValueKind}, }; use openshell_ocsf::{ - ConfigStateChangeBuilder, OCSF_TARGET, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, + ConfigStateChangeBuilder, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, }; use openshell_policy::{ PolicyMergeOp, ProviderPolicyLayer, canonicalize_advisor_add_rule, compose_effective_policy, @@ -162,7 +162,7 @@ fn emit_gateway_policy_audit_log( version: i64, policy_hash: &str, ) { - let message = build_gateway_policy_audit_message( + let event = build_gateway_policy_audit_event( sandbox_id, sandbox_name, state_label, @@ -171,11 +171,7 @@ fn emit_gateway_policy_audit_log( policy_hash, &[], ); - info!( - target: OCSF_TARGET, - sandbox_id = %sandbox_id, - message = %message - ); + openshell_ocsf::ocsf_emit!(event); } /// Emit a `CONFIG:APPROVED` audit event for an auto-approval — same event @@ -200,7 +196,7 @@ fn emit_gateway_policy_auto_approve_audit_log( ("prover_delta", "empty".to_string()), ("resolved_from", resolved_from.to_string()), ]; - let message = build_gateway_policy_audit_message( + let event = build_gateway_policy_audit_event( sandbox_id, sandbox_name, "approved", @@ -209,14 +205,10 @@ fn emit_gateway_policy_auto_approve_audit_log( policy_hash, &extra, ); - info!( - target: OCSF_TARGET, - sandbox_id = %sandbox_id, - message = %message - ); + openshell_ocsf::ocsf_emit!(event); } -fn build_gateway_policy_audit_message( +fn build_gateway_policy_audit_event( sandbox_id: &str, sandbox_name: &str, state_label: &str, @@ -224,7 +216,7 @@ fn build_gateway_policy_audit_message( version: i64, policy_hash: &str, extra_fields: &[(&str, String)], -) -> String { +) -> OcsfEvent { let ctx = SandboxContext { sandbox_id: sandbox_id.to_string(), sandbox_name: sandbox_name.to_string(), @@ -248,8 +240,7 @@ fn build_gateway_policy_audit_message( for (key, value) in extra_fields { builder = builder.unmapped(key, value.clone()); } - let event: OcsfEvent = builder.build(); - event.format_shorthand() + builder.build() } fn summarize_cli_policy_merge_op(operation: &PolicyMergeOp) -> String { @@ -15486,8 +15477,8 @@ mod tests { } #[test] - fn build_gateway_policy_audit_message_formats_ocsf_config_line() { - let message = build_gateway_policy_audit_message( + fn build_gateway_policy_audit_event_formats_ocsf_config_line() { + let message = build_gateway_policy_audit_event( "sb-123", "demo-sandbox", "merged", @@ -15495,7 +15486,8 @@ mod tests { 7, "sha256:testhash", &[], - ); + ) + .format_shorthand(); assert_eq!( message, @@ -15510,13 +15502,13 @@ mod tests { /// findings" — never "safe" — because the claim is about the prover's /// reasoning, not the world. #[test] - fn build_gateway_policy_audit_message_carries_auto_approve_provenance() { + fn build_gateway_policy_audit_event_carries_auto_approve_provenance() { let extra = [ ("auto", "true".to_string()), ("source", "agent_authored".to_string()), ("prover_delta", "empty".to_string()), ]; - let message = build_gateway_policy_audit_message( + let message = build_gateway_policy_audit_event( "sb-123", "demo-sandbox", "approved", @@ -15524,7 +15516,8 @@ mod tests { 12, "sha256:autohash", &extra, - ); + ) + .format_shorthand(); assert!( message.contains("CONFIG:APPROVED"), "auto-approval reuses CONFIG:APPROVED; got: {message}" diff --git a/crates/openshell-server/src/service_routing.rs b/crates/openshell-server/src/service_routing.rs index 3e80bc26f5..1ca575c059 100644 --- a/crates/openshell-server/src/service_routing.rs +++ b/crates/openshell-server/src/service_routing.rs @@ -14,17 +14,18 @@ use openshell_core::proto::{Sandbox, SandboxPhase, ServiceEndpoint, TcpRelayTarg use openshell_core::{ObjectId, VERSION}; use openshell_ocsf::{ ActionId, ActivityId, ConfigStateChangeBuilder, DispositionId, Endpoint, HttpActivityBuilder, - HttpRequest, HttpResponse as OcsfHttpResponse, NetworkActivityBuilder, OCSF_TARGET, OcsfEvent, + HttpRequest, HttpResponse as OcsfHttpResponse, NetworkActivityBuilder, OcsfEvent, SandboxContext, SeverityId, StateId, StatusId, Url as OcsfUrl, }; use std::net::{IpAddr, Ipv4Addr}; use std::sync::Arc; use std::time::Duration; use tokio::io::AsyncWriteExt; -use tracing::{info, warn}; +use tracing::warn; use crate::ServerState; use crate::persistence::{ObjectType, Store}; +use crate::sandbox_index::SandboxIndex; const ENDPOINT_OBJECT_TYPE: &str = "service_endpoint"; const ROUTING_RULE_NAME: &str = "sandbox_service_routing"; @@ -236,7 +237,15 @@ async fn proxy_to_endpoint( { Ok(endpoint) => endpoint, Err(err) => { - emit_service_http_failure(&state, &req, &sandbox_name, &service_name, None, &err); + emit_service_http_failure( + &state, + &req, + workspace, + &sandbox_name, + &service_name, + None, + &err, + ); return Err(err); } }; @@ -245,6 +254,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -264,6 +274,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -277,6 +288,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -290,6 +302,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -302,6 +315,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -314,6 +328,7 @@ async fn proxy_to_endpoint( emit_service_http_failure( &state, &req, + workspace, &sandbox_name, &service_name, Some(&endpoint), @@ -582,19 +597,19 @@ fn is_gateway_auth_cookie(name: &str) -> bool { pub fn emit_service_endpoint_config_event(endpoint: &ServiceEndpoint, url: &str, created: bool) { let event = build_service_endpoint_config_event(endpoint, url, created); - emit_gateway_ocsf_event(&endpoint.sandbox_id, event); + emit_gateway_ocsf_event(event); } pub fn emit_service_endpoint_delete_event(endpoint: &ServiceEndpoint) { let event = build_service_endpoint_delete_event(endpoint); - emit_gateway_ocsf_event(&endpoint.sandbox_id, event); + emit_gateway_ocsf_event(event); } pub fn emit_cross_origin_service_http_rejection(state: &ServerState, req: &Request) { let Some(host) = request_host(req) else { return; }; - let Some((_workspace, sandbox_name, service_name)) = + let Some((workspace, sandbox_name, service_name)) = parse_host(host, &state.config.service_routing) else { return; @@ -604,32 +619,42 @@ pub fn emit_cross_origin_service_http_rejection(state: &ServerState, req: &Reque "Cross-origin service request rejected", "cross-origin service request rejected", ); - emit_service_http_failure(state, req, &sandbox_name, &service_name, None, &err); + emit_service_http_failure( + state, + req, + &workspace, + &sandbox_name, + &service_name, + None, + &err, + ); } fn emit_service_http_failure( state: &ServerState, req: &Request, + workspace: &str, sandbox_name: &str, service_name: &str, endpoint: Option<&ServiceEndpoint>, err: &ServiceRouteError, ) { + let sandbox_id = + http_failure_sandbox_id(&state.sandbox_index, workspace, sandbox_name, endpoint); let event = build_service_http_failure_event( state.config.bind_address.port(), req, + &sandbox_id, sandbox_name, service_name, - endpoint, err, ); - let sandbox_id = endpoint.map_or("", |endpoint| endpoint.sandbox_id.as_str()); - emit_gateway_ocsf_event(sandbox_id, event); + emit_gateway_ocsf_event(event); } fn emit_service_relay_failure(endpoint: &ServiceEndpoint, target_port: u16, reason: &str) { let event = build_service_relay_failure_event(endpoint, target_port, reason); - emit_gateway_ocsf_event(&endpoint.sandbox_id, event); + emit_gateway_ocsf_event(event); } fn build_service_endpoint_config_event( @@ -679,20 +704,34 @@ fn build_service_endpoint_delete_event(endpoint: &ServiceEndpoint) -> OcsfEvent .build() } +/// Resolve an endpoint's sandbox id, falling back to the workspace/name index. +fn http_failure_sandbox_id( + index: &SandboxIndex, + workspace: &str, + sandbox_name: &str, + endpoint: Option<&ServiceEndpoint>, +) -> String { + endpoint.map_or_else( + || { + index + .sandbox_id_for_sandbox_name(workspace, sandbox_name) + .unwrap_or_default() + }, + |endpoint| endpoint.sandbox_id.clone(), + ) +} + fn build_service_http_failure_event( bind_port: u16, req: &Request, + sandbox_id: &str, sandbox_name: &str, service_name: &str, - endpoint: Option<&ServiceEndpoint>, err: &ServiceRouteError, ) -> OcsfEvent { let host = request_host(req).unwrap_or("unknown"); let (hostname, port) = split_authority_for_event(host, bind_port); - let ctx = gateway_ocsf_ctx( - endpoint.map_or("", |endpoint| endpoint.sandbox_id.as_str()), - sandbox_name, - ); + let ctx = gateway_ocsf_ctx(sandbox_id, sandbox_name); HttpActivityBuilder::new(&ctx) .activity(http_activity_for_method(req.method())) .action(ActionId::Denied) @@ -751,13 +790,9 @@ fn build_service_relay_failure_event( .build() } -fn emit_gateway_ocsf_event(sandbox_id: &str, event: OcsfEvent) { - let message = event.format_shorthand(); - info!( - target: OCSF_TARGET, - sandbox_id = %sandbox_id, - message = %message - ); +/// Emit through the structured OCSF tracing bridge. +fn emit_gateway_ocsf_event(event: OcsfEvent) { + openshell_ocsf::ocsf_emit!(event); } fn gateway_ocsf_ctx(sandbox_id: &str, sandbox_name: &str) -> SandboxContext { @@ -1120,8 +1155,14 @@ mod tests { "Cross-origin service request rejected", "cross-origin service request rejected", ); - let event = - build_service_http_failure_event(18080, &request, "my-sandbox", "web", None, &err); + let event = build_service_http_failure_event( + 18080, + &request, + "sandbox-1", + "my-sandbox", + "web", + &err, + ); let json = event.to_json().unwrap(); assert_eq!(json["class_uid"], 4002); @@ -1234,4 +1275,115 @@ mod tests { "should not find endpoint in wrong workspace" ); } + + fn indexed_sandbox() -> SandboxIndex { + let index = SandboxIndex::new(); + index.update_from_sandbox(&Sandbox { + metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { + id: "sandbox-1".to_string(), + name: "my-sandbox".to_string(), + workspace: "default".to_string(), + ..Default::default() + }), + ..Default::default() + }); + index + } + + #[test] + fn http_failure_resolves_the_sandbox_id_from_the_index_without_an_endpoint() { + let index = indexed_sandbox(); + + assert_eq!( + http_failure_sandbox_id(&index, "default", "my-sandbox", None), + "sandbox-1" + ); + } + + #[test] + fn http_failure_sandbox_id_prefers_the_endpoint_and_tolerates_unknown_names() { + let index = indexed_sandbox(); + let endpoint = endpoint(); + + assert_eq!( + http_failure_sandbox_id(&index, "default", "my-sandbox", Some(&endpoint)), + endpoint.sandbox_id + ); + assert_eq!( + http_failure_sandbox_id(&index, "default", "does-not-exist", None), + "" + ); + assert_eq!( + http_failure_sandbox_id(&index, "other-workspace", "my-sandbox", None), + "" + ); + } + + #[test] + fn http_failure_event_carries_the_resolved_sandbox_id() { + let req = Request::builder() + .method(Method::GET) + .uri("/") + .header(header::HOST, "default--my-sandbox--web.example.test") + .body(Body::empty()) + .unwrap(); + let event = build_service_http_failure_event( + 8443, + &req, + "sandbox-1", + "my-sandbox", + "web", + &ServiceRouteError::endpoint_not_found(), + ); + + assert_eq!( + event.base().metadata.uid.as_deref(), + Some("sandbox-1"), + "resolved sandbox id should reach the event" + ); + } + + /// Captures structured OCSF events during tracing dispatch. + #[derive(Clone, Default)] + struct ProbeLayer { + seen: Arc>>>, + } + + impl tracing_subscriber::Layer for ProbeLayer { + fn on_event( + &self, + event: &tracing::Event<'_>, + _ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + if event.metadata().target() == openshell_ocsf::OCSF_TARGET { + self.seen + .lock() + .unwrap() + .push(openshell_ocsf::clone_current_event()); + } + } + } + + #[test] + fn gateway_ocsf_events_expose_the_structured_event_to_layers() { + use tracing_subscriber::layer::SubscriberExt; + + let probe = ProbeLayer::default(); + let subscriber = tracing_subscriber::registry().with(probe.clone()); + + let endpoint = endpoint(); + let expected = build_service_endpoint_config_event(&endpoint, "https://example.test", true); + let expected_shorthand = expected.format_shorthand(); + + tracing::subscriber::with_default(subscriber, || { + emit_service_endpoint_config_event(&endpoint, "https://example.test", true); + }); + + let seen = probe.seen.lock().unwrap(); + assert_eq!(seen.len(), 1, "expected exactly one OCSF tracing event"); + let event = seen[0] + .as_ref() + .expect("structured OCSF event should be reachable from the layer"); + assert_eq!(event.format_shorthand(), expected_shorthand); + } } diff --git a/crates/openshell-server/src/tls.rs b/crates/openshell-server/src/tls.rs index aa627746ce..e3b48b9fbb 100644 --- a/crates/openshell-server/src/tls.rs +++ b/crates/openshell-server/src/tls.rs @@ -14,9 +14,7 @@ use arc_swap::ArcSwap; use notify::event::EventKind; use notify::{Event, RecursiveMode, Watcher}; use openshell_core::{Error, Result}; -use openshell_ocsf::{ - ConfigStateChangeBuilder, OCSF_TARGET, SandboxContext, SeverityId, StateId, StatusId, -}; +use openshell_ocsf::{ConfigStateChangeBuilder, SandboxContext, SeverityId, StateId, StatusId}; use rustls::ServerConfig; use rustls::crypto::ring::sign; use rustls::pki_types::{CertificateDer, PrivateKeyDer}; @@ -117,11 +115,7 @@ impl TlsAcceptor { .state(StateId::Enabled, "reloaded") .message("TLS certificate config reloaded successfully") .build(); - info!( - target: OCSF_TARGET, - sandbox_id = "", - message = %event.format_shorthand() - ); + openshell_ocsf::ocsf_emit!(event); Ok(()) } @@ -247,11 +241,7 @@ impl TlsAcceptor { "TLS certificate reload failed: {e}" )) .build(); - info!( - target: OCSF_TARGET, - sandbox_id = "", - message = %event.format_shorthand() - ); + openshell_ocsf::ocsf_emit!(event); warn!(error = %e, "TLS certificate reload failed, keeping existing config"); } break; diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index a91a5fd877..15c7908e34 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -3,7 +3,7 @@ //! Capture openshell-server tracing logs for streaming over gRPC. -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{Arc, Mutex}; use openshell_core::proto::{SandboxLogLine, SandboxStreamEvent}; @@ -24,6 +24,9 @@ pub struct TracingLogBus { struct Inner { per_id: HashMap>, tails: HashMap>, + /// Recently removed sandbox ids, in eviction order. + removed: VecDeque, + removed_set: HashSet, } impl Default for TracingLogBus { @@ -39,6 +42,8 @@ impl TracingLogBus { inner: Arc::new(Mutex::new(Inner { per_id: HashMap::new(), tails: HashMap::new(), + removed: VecDeque::new(), + removed_set: HashSet::new(), })), platform_event_bus: PlatformEventBus::new(), } @@ -51,8 +56,13 @@ impl TracingLogBus { } } - fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { + pub fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + if inner.removed_set.contains(sandbox_id) { + let (tx, rx) = broadcast::channel(1); + drop(tx); + return rx; + } inner .per_id .entry(sandbox_id.to_string()) @@ -60,11 +70,7 @@ impl TracingLogBus { let (tx, _rx) = broadcast::channel(1024); tx }) - .clone() - } - - pub fn subscribe(&self, sandbox_id: &str) -> broadcast::Receiver { - self.sender_for(sandbox_id).subscribe() + .subscribe() } /// Remove all bus entries for the given sandbox id. @@ -75,6 +81,15 @@ impl TracingLogBus { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); inner.per_id.remove(sandbox_id); inner.tails.remove(sandbox_id); + + if inner.removed_set.insert(sandbox_id.to_string()) { + inner.removed.push_back(sandbox_id.to_string()); + while inner.removed.len() > Self::MAX_REMEMBERED_REMOVALS { + if let Some(evicted) = inner.removed.pop_front() { + inner.removed_set.remove(&evicted); + } + } + } } pub fn tail(&self, sandbox_id: &str, max: usize) -> Vec { @@ -95,6 +110,9 @@ impl TracingLogBus { /// used by the tracing layer, so it appears in `WatchSandbox` and /// `GetSandboxLogs` transparently. pub fn publish_external(&self, log: SandboxLogLine) { + if log.sandbox_id.is_empty() { + return; + } let evt = SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log( log.clone(), @@ -106,16 +124,42 @@ impl TracingLogBus { /// Default tail buffer capacity (lines per sandbox). const DEFAULT_TAIL: usize = 2000; - fn publish(&self, sandbox_id: &str, event: SandboxStreamEvent, tail_cap: usize) { - let tx = self.sender_for(sandbox_id); - let _ = tx.send(event.clone()); + /// Number of `(sender, tail)` entries currently held, for leak assertions. + #[cfg(test)] + fn entry_counts(&self) -> (usize, usize) { + let inner = self.inner.lock().expect("tracing bus lock poisoned"); + (inner.per_id.len(), inner.tails.len()) + } + + /// Maximum number of removed sandbox ids to retain. + /// + /// This bounds memory; after eviction, a very late publisher may create a + /// fresh entry for that id. + const MAX_REMEMBERED_REMOVALS: usize = 1024; + fn publish(&self, sandbox_id: &str, event: SandboxStreamEvent, tail_cap: usize) { let mut inner = self.inner.lock().expect("tracing bus lock poisoned"); + if inner.removed_set.contains(sandbox_id) { + return; + } + + let tx = inner + .per_id + .entry(sandbox_id.to_string()) + .or_insert_with(|| { + let (tx, _rx) = broadcast::channel(1024); + tx + }) + .clone(); + let deque = inner.tails.entry(sandbox_id.to_string()).or_default(); - deque.push_back(event); + deque.push_back(event.clone()); while deque.len() > tail_cap { deque.pop_front(); } + drop(inner); + + let _ = tx.send(event); } } @@ -131,14 +175,28 @@ where { fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { let meta = event.metadata(); - let mut visitor = LogVisitor::default(); - event.record(&mut visitor); + // OCSF tracing events carry no fields; the payload arrives out of band + // through a thread-local. + let (visitor_sandbox_id, visitor_message) = if meta.target() == OCSF_TARGET { + openshell_ocsf::clone_current_event().map_or((None, None), |ocsf_event| { + ( + ocsf_event.base().metadata.uid.clone(), + Some(ocsf_event.format_shorthand()), + ) + }) + } else { + let mut visitor = LogVisitor::default(); + event.record(&mut visitor); + (visitor.sandbox_id, visitor.message) + }; - let Some(sandbox_id) = visitor.sandbox_id else { + // An empty id means no sandbox association; publishing would allocate a + // bucket nothing can subscribe to. + let Some(sandbox_id) = visitor_sandbox_id.filter(|id| !id.is_empty()) else { return; }; - let msg = visitor.message.unwrap_or_else(|| meta.name().to_string()); + let msg = visitor_message.unwrap_or_else(|| meta.name().to_string()); let level = display_level(meta.target(), &meta.level().to_string()); let ts = openshell_core::time::now_ms(); @@ -228,22 +286,22 @@ mod tests { } #[test] - fn tracing_log_bus_subscribe_after_remove_creates_fresh_channel() { + fn subscribe_after_remove_does_not_reactivate_the_bus() { let bus = TracingLogBus::new(); let sandbox_id = "sb-2"; - // Create and remove bus.publish_external(make_log_event(sandbox_id, "old message")); bus.remove(sandbox_id); - // Subscribe again — should get a fresh channel with no history let mut rx = bus.subscribe(sandbox_id); - assert!(bus.tail(sandbox_id, 10).is_empty()); + bus.publish_external(make_log_event(sandbox_id, "late message")); - // New publish should reach the new subscriber - bus.publish_external(make_log_event(sandbox_id, "new message")); - let evt = rx.try_recv().expect("should receive new event"); - assert!(evt.payload.is_some()); + assert_eq!(bus.entry_counts(), (0, 0)); + assert!(bus.tail(sandbox_id, 10).is_empty()); + assert!(matches!( + rx.try_recv(), + Err(broadcast::error::TryRecvError::Closed) + )); } #[test] @@ -270,6 +328,143 @@ mod tests { bus.remove("nonexistent"); } + #[test] + fn publish_after_remove_does_not_resurrect_the_bus_entry() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-torn-down"; + + bus.publish_external(make_log_event(sandbox_id, "before teardown")); + assert_eq!(bus.entry_counts(), (1, 1)); + + bus.remove(sandbox_id); + assert_eq!(bus.entry_counts(), (0, 0)); + + bus.publish_external(make_log_event(sandbox_id, "late line")); + assert_eq!(bus.entry_counts(), (0, 0)); + assert!(bus.tail(sandbox_id, 10).is_empty()); + } + + fn ocsf_ctx(sandbox_id: &str) -> openshell_ocsf::SandboxContext { + openshell_ocsf::SandboxContext { + sandbox_id: sandbox_id.to_string(), + sandbox_name: "gw".to_string(), + container_image: "openshell/gateway".to_string(), + hostname: "openshell-gateway".to_string(), + product_version: "0.0.0".to_string(), + proxy_ip: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), + proxy_port: 0, + } + } + + /// Run `f` with the bus layer installed as the active subscriber. + fn with_bus_layer(bus: &TracingLogBus, f: impl FnOnce()) { + use tracing_subscriber::layer::SubscriberExt; + let subscriber = tracing_subscriber::registry().with(bus.layer()); + tracing::subscriber::with_default(subscriber, f); + } + + fn log_message(event: &SandboxStreamEvent) -> &SandboxLogLine { + match event.payload { + Some(openshell_core::proto::sandbox_stream_event::Payload::Log(ref log)) => log, + _ => panic!("expected a log payload"), + } + } + + #[test] + fn ocsf_emit_events_reach_the_bus_with_shorthand_and_sandbox_id() { + use openshell_ocsf::{ + ActionId, ActivityId, DispositionId, Endpoint, NetworkActivityBuilder, SeverityId, + StatusId, ocsf_emit, + }; + + let bus = TracingLogBus::new(); + let event = NetworkActivityBuilder::new(&ocsf_ctx("sb-emit")) + .activity(ActivityId::Open) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443)) + .message("CONNECT denied blocked.example.com:443") + .build(); + let expected_message = event.format_shorthand(); + + with_bus_layer(&bus, || ocsf_emit!(event)); + + let tail = bus.tail("sb-emit", 10); + assert_eq!(tail.len(), 1, "ocsf_emit! event should reach the bus"); + let log = log_message(&tail[0]); + assert_eq!(log.sandbox_id, "sb-emit"); + assert_eq!(log.message, expected_message); + assert_eq!(log.level, "OCSF"); + assert_eq!(log.target, OCSF_TARGET); + assert_eq!(log.source, "gateway"); + } + + #[test] + fn ocsf_emit_events_without_a_sandbox_are_skipped() { + use openshell_ocsf::{ActivityId, AppLifecycleBuilder, SeverityId, ocsf_emit}; + + let bus = TracingLogBus::new(); + let event = AppLifecycleBuilder::new(&ocsf_ctx("")) + .activity(ActivityId::Open) + .severity(SeverityId::Informational) + .message("gateway TLS reloaded") + .build(); + + with_bus_layer(&bus, || ocsf_emit!(event)); + + assert_eq!(bus.entry_counts(), (0, 0)); + } + + #[test] + fn non_ocsf_events_still_use_the_sandbox_id_field() { + let bus = TracingLogBus::new(); + with_bus_layer(&bus, || { + tracing::info!(sandbox_id = "sb-plain", "plain gateway line"); + }); + + let tail = bus.tail("sb-plain", 10); + assert_eq!(tail.len(), 1); + let log = log_message(&tail[0]); + assert_eq!(log.message, "plain gateway line"); + assert_eq!(log.level, "INFO"); + } + + #[test] + fn removal_tombstones_are_bounded() { + let bus = TracingLogBus::new(); + let overflow = TracingLogBus::MAX_REMEMBERED_REMOVALS + 10; + for i in 0..overflow { + bus.remove(&format!("sb-{i}")); + } + + let inner = bus.inner.lock().unwrap(); + assert_eq!(inner.removed.len(), TracingLogBus::MAX_REMEMBERED_REMOVALS); + assert_eq!( + inner.removed_set.len(), + TracingLogBus::MAX_REMEMBERED_REMOVALS + ); + assert!(!inner.removed_set.contains("sb-0")); + assert!(inner.removed_set.contains(&format!("sb-{}", overflow - 1))); + } + + #[test] + fn publish_external_ignores_an_empty_sandbox_id() { + let bus = TracingLogBus::new(); + bus.publish_external(make_log_event("", "no sandbox association")); + + assert_eq!(bus.entry_counts(), (0, 0)); + assert!(bus.tail("", 10).is_empty()); + } + + #[test] + fn publish_external_still_accepts_a_real_sandbox_id() { + let bus = TracingLogBus::new(); + bus.publish_external(make_log_event("sb-real", "hello")); + assert_eq!(bus.tail("sb-real", 10).len(), 1); + } + #[test] fn display_level_maps_ocsf_target_to_ocsf() { assert_eq!(display_level(OCSF_TARGET, "INFO"), "OCSF");