From f708494bde34c544b61b761d9db09ba7b2e79f2e Mon Sep 17 00:00:00 2001 From: Kris Hicks Date: Fri, 28 Aug 2026 13:02:49 -0700 Subject: [PATCH] fix(server): route gateway events to sandbox logs Previously, service-routing failures where no endpoint could be resolved had no sandbox ID and disappeared from that sandbox's logs. Now, the gateway resolves the ID from the workspace and sandbox name. This makes failures such as requests for a missing service visible to operators in `openshell logs`. Additionally, gateway OCSF events used to be flattened into tracing fields to reach `openshell logs`. Now, the log bus reads structured OCSF events directly and routes sandbox-scoped policy and service-routing activity without changing its shorthand display. This gives gateway event producers one OCSF emission path and keeps the structured event available to tracing layers. Finally, gateway-wide events could allocate a phantom empty-sandbox log, and late activity could recreate a deleted sandbox's log state. Now, empty IDs are ignored and bounded removal tombstones keep deleted log streams closed. This avoids unreachable log buckets and prevents stale activity from reviving a removed sandbox's in-memory log stream. Refs #1055 Signed-off-by: Kris Hicks --- crates/openshell-server/src/grpc/policy.rs | 39 ++- .../openshell-server/src/service_routing.rs | 202 +++++++++++++-- crates/openshell-server/src/tls.rs | 16 +- crates/openshell-server/src/tracing_bus.rs | 241 ++++++++++++++++-- 4 files changed, 414 insertions(+), 84 deletions(-) diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 054a3c34c6..8711f2fe33 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, @@ -167,7 +167,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, @@ -176,11 +176,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 @@ -205,7 +201,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", @@ -214,14 +210,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, @@ -229,7 +221,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(), @@ -253,8 +245,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 { @@ -16839,8 +16830,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", @@ -16848,7 +16839,8 @@ mod tests { 7, "sha256:testhash", &[], - ); + ) + .format_shorthand(); assert_eq!( message, @@ -16863,13 +16855,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", @@ -16877,7 +16869,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 531daf7873..bba499c6d1 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::aws_lc_rs::sign; use rustls::pki_types::{CertificateDer, PrivateKeyDer}; @@ -117,11 +115,7 @@ impl TlsAcceptor { .state(StateId::Enabled, "reloaded") .message("TLS certificate config reloaded successfully") .build(); - info!( - target: OCSF_TARGET, - sandbox_id = "", - message = %event.format_shorthand() - ); + openshell_ocsf::ocsf_emit!(event); Ok(()) } @@ -247,11 +241,7 @@ impl TlsAcceptor { "TLS certificate reload failed: {e}" )) .build(); - info!( - target: OCSF_TARGET, - sandbox_id = "", - message = %event.format_shorthand() - ); + openshell_ocsf::ocsf_emit!(event); warn!(error = %e, "TLS certificate reload failed, keeping existing config"); } break; 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");