policy_fn: join the callback thread before the sandbox returns

crates/sandlock-core/src/policy_fn.rs

@@ -351,24 +351,52 @@ pub struct PolicyEvent {
 // Policy callback runner
 // ============================================================
 
+/// What the supervisor pushes to the policy thread.
+pub enum PolicyMsg {
+    Event(PolicyEvent),
+    Shutdown,
+}
+
+/// Owns the policy-callback thread. Dropping it is the guarantee that no
+/// callback runs afterwards: the supervisor's sender clones outlive an
+/// `abort()`, so shutdown is an in-band message rather than channel closure,
+/// and the drop joins the thread.
+pub(crate) struct PolicyFnWorker {
+    tx: tokio::sync::mpsc::UnboundedSender<PolicyMsg>,
+    thread: Option<std::thread::JoinHandle<()>>,
+}
+
+impl PolicyFnWorker {
+    pub(crate) fn sender(&self) -> tokio::sync::mpsc::UnboundedSender<PolicyMsg> {
+        self.tx.clone()
+    }
+}
+
+impl Drop for PolicyFnWorker {
+    fn drop(&mut self) {
+        let _ = self.tx.send(PolicyMsg::Shutdown);
+        if let Some(t) = self.thread.take() {
+            let _ = t.join();
+        }
+    }
+}
+
 /// Spawn a thread that receives syscall events and calls the policy callback.
-///
-/// Returns a sender for the supervisor to push events into.
 pub(crate) fn spawn_policy_fn(
     callback: PolicyCallback,
     live: Arc<RwLock<LivePolicy>>,
     ceiling: LivePolicy,
     pid_overrides: Arc<RwLock<HashMap<u32, HashSet<IpAddr>>>>,
     denied: Arc<crate::seccomp::state::DeniedSet>,
-) -> tokio::sync::mpsc::UnboundedSender<PolicyEvent> {
-    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<PolicyEvent>();
+) -> PolicyFnWorker {
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<PolicyMsg>();
 
-    std::thread::Builder::new()
+    let thread = std::thread::Builder::new()
         .name("sandlock-policy-fn".to_string())
         .spawn(move || {
             let mut ctx = PolicyContext::new(live, ceiling, pid_overrides, denied);
 
-            while let Some(pe) = rx.blocking_recv() {
+            while let Some(PolicyMsg::Event(pe)) = rx.blocking_recv() {
                 let verdict = callback(pe.event, &mut ctx);
 
                 // Signal the supervisor with the verdict.
@@ -380,7 +408,7 @@ pub(crate) fn spawn_policy_fn(
         })
         .expect("failed to spawn policy-fn thread");
 
-    tx
+    PolicyFnWorker { tx, thread: Some(thread) }
 }
 
 // ============================================================
@@ -391,6 +419,46 @@ pub(crate) fn spawn_policy_fn(
 mod tests {
     use super::*;
 
+    #[test]
+    fn worker_drop_waits_for_queued_callbacks() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let calls = Arc::new(AtomicUsize::new(0));
+        let seen = calls.clone();
+        let callback: PolicyCallback = Arc::new(move |_e, _c| {
+            std::thread::sleep(std::time::Duration::from_millis(100));
+            seen.fetch_add(1, Ordering::SeqCst);
+            Verdict::Allow
+        });
+        let live = Arc::new(RwLock::new(test_live()));
+        let worker = spawn_policy_fn(
+            callback,
+            live,
+            test_live(),
+            Arc::new(RwLock::new(HashMap::new())),
+            Arc::new(crate::seccomp::state::DeniedSet::default()),
+        );
+        let tx = worker.sender();
+        let event = SyscallEvent {
+            syscall: "close".into(),
+            category: SyscallCategory::File,
+            pid: 1,
+            parent_pid: None,
+            host: None,
+            port: None,
+            size: None,
+            argv: None,
+            denied: false,
+            path: None,
+            path2: None,
+            fd: None,
+            flags: None,
+            protocol: None,
+        };
+        tx.send(PolicyMsg::Event(PolicyEvent { event, gate: None })).unwrap();
+        drop(worker);
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
     fn test_live() -> LivePolicy {
         LivePolicy {
             allowed_ips: ["127.0.0.1", "10.0.0.1"]

crates/sandlock-core/src/sandbox.rs

@@ -270,6 +270,7 @@ struct Runtime {
     child_pid: Option<i32>,
     pidfd: Option<std::os::fd::OwnedFd>,
     notif_handle: Option<JoinHandle<()>>,
+    policy_fn_worker: Option<crate::policy_fn::PolicyFnWorker>,
     throttle_handle: Option<JoinHandle<()>>,
     loadavg_handle: Option<JoinHandle<()>>,
     control_handle: Option<JoinHandle<()>>,
@@ -930,6 +931,7 @@ impl Sandbox {
 
         let rt = self.rt_mut();
         if let Some(h) = rt.notif_handle.take() { h.abort(); }
+        rt.policy_fn_worker = None;
         if let Some(h) = rt.throttle_handle.take() { h.abort(); }
         if let Some(h) = rt.loadavg_handle.take() { h.abort(); }
         if let Some(h) = rt.control_handle.take() { h.abort(); }
@@ -1514,6 +1516,7 @@ impl Sandbox {
                 child_pid: Some(clone_pid),
                 pidfd: None,
                 notif_handle: None,
+                policy_fn_worker: None,
                 throttle_handle: None,
                 loadavg_handle: None,
                 _stdout_read: None,
@@ -1614,6 +1617,7 @@ impl Sandbox {
             child_pid: None,
             pidfd: None,
             notif_handle: None,
+            policy_fn_worker: None,
             throttle_handle: None,
             loadavg_handle: None,
             control_handle: None,
@@ -2240,10 +2244,11 @@ impl Sandbox {
                 let denied = policy_fn_state.denied.clone();
                 let pid_overrides = net_state.pid_ip_overrides.clone();
                 policy_fn_state.live_policy = Some(live.clone());
-                let tx = crate::policy_fn::spawn_policy_fn(
+                let worker = crate::policy_fn::spawn_policy_fn(
                     callback.clone(), live, ceiling, pid_overrides, denied,
                 );
-                policy_fn_state.event_tx = Some(tx);
+                policy_fn_state.event_tx = Some(worker.sender());
+                self.rt_mut().policy_fn_worker = Some(worker);
             }
 
             let chroot_state = ChrootState::new();
@@ -2492,6 +2497,7 @@ impl Drop for Sandbox {
             }
 
             if let Some(h) = rt.notif_handle.take() { h.abort(); }
+            rt.policy_fn_worker = None;
             if let Some(h) = rt.throttle_handle.take() { h.abort(); }
             if let Some(h) = rt.loadavg_handle.take() { h.abort(); }
             if let Some(h) = rt.control_handle.take() { h.abort(); }

crates/sandlock-core/src/seccomp/notif.rs

@@ -2068,26 +2068,28 @@ async fn emit_policy_event(
 
     let verdict = if is_held {
         let (gate_tx, gate_rx) = tokio::sync::oneshot::channel();
-        let _ = tx.send(crate::policy_fn::PolicyEvent {
+        let _ = tx.send(crate::policy_fn::PolicyMsg::Event(crate::policy_fn::PolicyEvent {
             event,
             gate: Some(gate_tx),
-        });
+        }));
         let received = match tokio::time::timeout(std::time::Duration::from_secs(5), gate_rx).await {
             Ok(Ok(verdict)) => Some(verdict),
             _ => None, // timeout or channel closed
         };
         resolve_held_gate(received)
     } else {
-        let _ = tx.send(crate::policy_fn::PolicyEvent {
+        let _ = tx.send(crate::policy_fn::PolicyMsg::Event(crate::policy_fn::PolicyEvent {
             event,
             gate: None,
-        });
+        }));
         None
     };
     // Emit the remaining sendmmsg destinations as observation-only events.
     // The verdict above already covers the whole syscall; gate: None is correct.
     for extra in sendmmsg_extras {
-        let _ = tx.send(crate::policy_fn::PolicyEvent { event: extra, gate: None });
+        let _ = tx.send(crate::policy_fn::PolicyMsg::Event(
+            crate::policy_fn::PolicyEvent { event: extra, gate: None },
+        ));
     }
 
     verdict

crates/sandlock-core/src/seccomp/state.rs

@@ -674,7 +674,7 @@ impl DeniedSet {
 /// Dynamic policy callback state.
 pub struct PolicyFnState {
     /// Event sender for dynamic policy callback (None if no policy_fn).
-    pub event_tx: Option<tokio::sync::mpsc::UnboundedSender<crate::policy_fn::PolicyEvent>>,
+    pub event_tx: Option<tokio::sync::mpsc::UnboundedSender<crate::policy_fn::PolicyMsg>>,
     /// Shared live policy for dynamic updates (None if no policy_fn).
     pub live_policy: Option<std::sync::Arc<std::sync::RwLock<crate::policy_fn::LivePolicy>>>,
     /// Dynamically denied paths and inode identities from policy_fn / fs_deny.