feat(mxc): warn when the ETW consumer receives zero provider events

crates/openshell-driver-mxc/src/etw_consumer.rs

@@ -111,6 +111,12 @@ const EVENT_QUEUE_BYTE_CAPACITY: usize = 16 * 1024 * 1024;
 /// while the consumer remains behind.
 const OVERLOAD_WARNING_INTERVAL: Duration = Duration::from_secs(30);
 
+/// Grace period, after the first sandbox activity is observed, before the
+/// zero-events watchdog warns that the session has matched no provider
+/// events at all. Generous on purpose: the goal is to catch a genuinely
+/// non-firing provider, not to flag normal per-sandbox event latency.
+const ZERO_EVENTS_GRACE: Duration = Duration::from_secs(30);
+
 /// `EVENT_CONTROL_CODE_ENABLE_PROVIDER`.
 const EVENT_CONTROL_CODE_ENABLE_PROVIDER: u32 = 1;
 
@@ -291,6 +297,14 @@ struct CaptureHealth {
     queued_bytes: AtomicUsize,
     /// Largest observed value of `queued_bytes`, retained for diagnostics.
     queue_high_water_bytes: AtomicUsize,
+    /// Count of raw events the callback matched to [`SANDBOXING_PROVIDER_GUID`]
+    /// and forwarded to the consumer thread, regardless of whether TDH decode
+    /// later succeeded. Zero here after real sandbox activity means the OS
+    /// session is not delivering *any* events under this GUID at all -- a
+    /// provider-identity mismatch or a provider that isn't firing on this
+    /// host/build, not a decode/attribution bug. See the zero-events watchdog
+    /// in `start_session`'s consumer loop.
+    events_matched: AtomicU64,
 }
 
 struct CallbackContext {
@@ -380,6 +394,13 @@ impl EtwSession {
     pub fn dropped_event_count(&self) -> u64 {
         self.health.dropped_events.load(Ordering::Relaxed)
     }
+
+    /// Count of raw provider-matched events received since the session
+    /// started. Exposed so the backend can surface "capture is running but
+    /// producing nothing" in status/diagnostics, alongside `is_capture_alive`.
+    pub fn events_received(&self) -> u64 {
+        self.health.events_matched.load(Ordering::Relaxed)
+    }
 }
 
 impl Drop for EtwSession {
@@ -420,10 +441,12 @@ pub(crate) fn start_session(index: Arc<Mutex<AttributionIndex>>) -> Result<EtwSe
     let handle = start_trace_session(&session_name)?;
     enable_provider(handle, &session_name)?;
 
+    // Created before the consumer thread spawns (unlike the pump thread's
+    // `health` clone below) so the consumer loop can track received-event
+    // count for the zero-events watchdog.
     let health = Arc::new(CaptureHealth::default());
     let consumer_health = health.clone();
     let (tx, rx) = mpsc::sync_channel::<RawEtwEvent>(EVENT_QUEUE_CAPACITY);
-
     let consumer_thread = std::thread::Builder::new()
         .name("etw-ocsf-consumer".into())
         .spawn(move || {
@@ -435,20 +458,32 @@ pub(crate) fn start_session(index: Arc<Mutex<AttributionIndex>>) -> Result<EtwSe
             // lull: an event that beat the driver's `register_launch` is replayed
             // within one tick once attribution lands, without having to wait for
             // the next ETW event (which may never arrive for a lone/last sandbox).
+            //
+            // Zero-events watchdog: `EnableTraceEx2` success is not proof the
+            // provider exists or will ever fire (see `SANDBOXING_PROVIDER_GUID`'s
+            // doc comment). If real sandbox activity has happened and this
+            // session still hasn't matched a single event after a grace period,
+            // that is a provider-identity mismatch or a non-firing provider on
+            // this host/build -- warn once instead of silently producing an
+            // empty audit trail.
+            let mut activity_since: Option<Instant> = None;
+            let mut warned_no_events = false;
             loop {
                 match rx.recv_timeout(Duration::from_millis(200)) {
                     Ok(mut raw) => {
                         release_queue_bytes(&consumer_health, raw.queued_bytes);
-                        if let Some(ev) = decode_raw(&mut raw) {
-                            process_event(&index, ev);
-                        } else {
-                            tracing::debug!(
+                        consumer_health
+                            .events_matched
+                            .fetch_add(1, Ordering::Relaxed);
+                        match decode_raw(&mut raw) {
+                            Some(ev) => process_event(&index, ev),
+                            None => tracing::debug!(
                                 target: "mxc_etw",
                                 id = raw.header.EventDescriptor.Id,
                                 opcode = raw.header.EventDescriptor.Opcode,
                                 pid = raw.header.ProcessId,
                                 "TDH decode failed for event"
-                            );
+                            ),
                         }
                         drain_and_emit(&index);
                         overload_reporter.report_if_due(&consumer_health, false);
@@ -462,6 +497,18 @@ pub(crate) fn start_session(index: Arc<Mutex<AttributionIndex>>) -> Result<EtwSe
                         break;
                     }
                 }
+
+                if !warned_no_events
+                    && should_warn_zero_events(
+                        consumer_health.events_matched.load(Ordering::Relaxed),
+                        index.lock().unwrap().total_launches(),
+                        &mut activity_since,
+                        ZERO_EVENTS_GRACE,
+                    )
+                {
+                    warned_no_events = true;
+                    warn_zero_events_received();
+                }
             }
             // Final drain on shutdown so anything still resolvable is emitted.
             drain_and_emit(&index);
@@ -1282,6 +1329,11 @@ pub(crate) struct AttributionIndex {
     /// them here and replay when a later registration/cross-link resolves them.
     /// Bounded by [`PENDING_MAX`] and [`PENDING_TTL`].
     pending: VecDeque<PendingEvent>,
+    /// Total sandboxes ever registered, never decremented by [`Self::forget`].
+    /// Used to gate the zero-events watchdog: a session that hasn't seen any
+    /// sandbox activity yet is expected to be quiet, so only warn once real
+    /// activity has happened and still produced nothing.
+    total_launches: u64,
 }
 
 impl AttributionIndex {
@@ -1300,6 +1352,7 @@ impl AttributionIndex {
         wxc_pid: u32,
         process_start_key: u64,
     ) {
+        self.total_launches += 1;
         let now = Instant::now();
         self.purge_expired_retirements(now);
         let previous = self.by_pid.remove(&wxc_pid);
@@ -1371,6 +1424,12 @@ impl AttributionIndex {
             .any(|registration| registration.sid == sandbox_id)
     }
 
+    /// Total sandboxes ever registered via [`Self::register_launch`], including
+    /// ones since [`Self::forget`]-ten. Used to gate the zero-events watchdog.
+    pub fn total_launches(&self) -> u64 {
+        self.total_launches
+    }
+
     /// Drop all keys for a finished sandbox to bound memory.
     pub fn forget(&mut self, sandbox_id: &str) {
         self.by_pid
@@ -1828,6 +1887,74 @@ fn map_finding(ctx: &EventContext, ev: &DecodedEtwEvent) -> OcsfEvent {
         .build()
 }
 
+/// Decide whether the zero-events watchdog should fire, and track when
+/// sandbox activity was first observed. Pure aside from `*activity_since`,
+/// which the caller retains across calls (and across a loop tick that
+/// doesn't warn) so the grace period is measured from first activity, not
+/// re-armed on every tick.
+///
+/// Gates on `total_launches` (not the index's current size) specifically so
+/// a quiet, idle gateway with `etw_audit=true` but zero sandboxes created
+/// never warns -- only genuine "activity happened, nothing arrived" does.
+fn should_warn_zero_events(
+    events_matched: u64,
+    total_launches: u64,
+    activity_since: &mut Option<Instant>,
+    grace: Duration,
+) -> bool {
+    if events_matched != 0 {
+        return false;
+    }
+    if activity_since.is_none() && total_launches > 0 {
+        *activity_since = Some(Instant::now());
+    }
+    activity_since.is_some_and(|since| since.elapsed() >= grace)
+}
+
+/// Fired once per session by the zero-events watchdog when real sandbox
+/// activity has happened but the session has never matched a single event to
+/// [`SANDBOXING_PROVIDER_GUID`]. `EnableTraceEx2` success only proves the
+/// *request* to enable the provider succeeded, not that the provider exists
+/// on this host/build or will ever actually fire -- this is the detection gap
+/// that made a real-world provider-identity mismatch silently produce an
+/// empty OCSF audit trail with no diagnostic at all.
+fn warn_zero_events_received() {
+    tracing::warn!(
+        target: "mxc_etw",
+        provider = ?SANDBOXING_PROVIDER_GUID,
+        session = SESSION_NAME,
+        "MXC ETW->OCSF consumer has received zero events from the Sandboxing \
+         provider despite sandbox activity; the OS-sourced audit trail is \
+         empty for this session. EnableTraceEx2 succeeding does not prove the \
+         provider exists on this host/build or will ever fire -- verify with \
+         `logman query providers` and confirm wxc-exec targets this GUID."
+    );
+    emit_ocsf(
+        "",
+        DetectionFindingBuilder::new(&etw_ctx("", "mxc-etw-consumer"))
+            .activity(ActivityId::Open) // finding label = "Create"
+            .severity(SeverityId::High)
+            .is_alert(true)
+            .finding_info(
+                FindingInfo::new(
+                    "mxc-etw-zero-events",
+                    "MXC ETW audit consumer received zero provider events",
+                )
+                .with_desc(
+                    "The Sandboxing ETW provider produced zero events despite \
+                     observed sandbox activity; the OS-sourced portion of the \
+                     audit trail is empty for this session.",
+                ),
+            )
+            .message(
+                "MXC ETW->OCSF consumer active but received zero events from \
+                 the Sandboxing provider after sandbox activity"
+                    .to_string(),
+            )
+            .build(),
+    );
+}
+
 /// Best-effort executable name from a command line: first whitespace-delimited
 /// token, stripped of any directory prefix and surrounding quotes.
 fn exe_name(cmd_line: &str) -> String {
@@ -2198,6 +2325,74 @@ mod tests {
         assert_eq!(idx.resolve(&ev_b).as_deref(), Some("sbx-B"));
     }
 
+    // Regression test for the zero-events watchdog: it must gate on whether
+    // sandbox activity has *ever* happened, not on the index's current size,
+    // since `forget` empties the index for every sandbox that completes
+    // normally -- `total_launches` must keep counting past that.
+    #[test]
+    fn total_launches_survives_forget() {
+        let mut idx = AttributionIndex::new();
+        assert_eq!(idx.total_launches(), 0);
+
+        idx.register_launch("sbx-1", "s1", 100, 1);
+        idx.forget("sbx-1");
+        assert_eq!(idx.total_launches(), 1);
+
+        idx.register_launch("sbx-2", "s2", 200, 2);
+        idx.register_launch("sbx-3", "s3", 300, 3);
+        assert_eq!(idx.total_launches(), 3);
+    }
+
+    #[test]
+    fn zero_events_watchdog_stays_quiet_without_sandbox_activity() {
+        // An idle gateway with etw_audit=true but no sandboxes created yet
+        // must never warn, however long it's been running.
+        let mut activity_since = None;
+        assert!(!should_warn_zero_events(
+            0,
+            0,
+            &mut activity_since,
+            Duration::ZERO
+        ));
+        assert!(activity_since.is_none());
+    }
+
+    #[test]
+    fn zero_events_watchdog_stays_quiet_once_any_event_matched() {
+        let mut activity_since = None;
+        assert!(!should_warn_zero_events(
+            1,
+            5,
+            &mut activity_since,
+            Duration::ZERO
+        ));
+    }
+
+    #[test]
+    fn zero_events_watchdog_waits_out_the_grace_period() {
+        let mut activity_since = None;
+        let grace = Duration::from_hours(1);
+        // First tick after activity starts: grace hasn't elapsed yet.
+        assert!(!should_warn_zero_events(0, 1, &mut activity_since, grace));
+        assert!(activity_since.is_some());
+        // A later tick still within the (long) grace window: still quiet.
+        assert!(!should_warn_zero_events(0, 1, &mut activity_since, grace));
+    }
+
+    #[test]
+    fn zero_events_watchdog_fires_once_grace_elapses() {
+        let mut activity_since = Some(Instant::now().checked_sub(Duration::from_mins(1)).unwrap());
+        assert!(should_warn_zero_events(
+            0,
+            1,
+            &mut activity_since,
+            Duration::from_secs(30)
+        ));
+    }
+
+    // Shailendra #1 (cmd ambiguity): two sandboxes running the identical command
+    // line must not let that command line resolve anything (it's ambiguous); a
+    // unique command line still works as a fallback.
     #[test]
     fn displaced_live_pid_does_not_claim_new_pre_registration_event() {
         let mut idx = AttributionIndex::new();