fix(sandbox): unwind every prepared owner before releasing resources

judge/crates/sandbox/src/acceptance.rs

@@ -82,12 +82,33 @@ fn prefix(file: &crate::artifacts::SnapshotFile) -> io::Result<String> {
 }
 pub fn run_case(case: &str) -> io::Result<serde_json::Value> {
     verify_guest(true)?;
+    let rollback = case.strip_prefix("cancel-").filter(|s| {
+        matches!(
+            *s,
+            "blocked"
+                | "namespace"
+                | "root"
+                | "proc"
+                | "inputs"
+                | "identity"
+                | "seccomp"
+                | "io"
+                | "exec"
+        )
+    });
+    let failing = case.strip_prefix("fail-").filter(|s| {
+        matches!(
+            *s,
+            "namespace" | "root" | "proc" | "inputs" | "rlimits" | "identity" | "seccomp" | "io"
+        )
+    });
     let mode = match case {
         "binary" | "stdout" | "stderr" | "wall" | "leftover" | "early" | "disk" | "export" => {
             MonitorMode::Normal
         }
         "cancel" => MonitorMode::CancelAfter(100000),
         "monitor" => MonitorMode::FailAfter(100000),
+        _ if rollback.is_some() || failing.is_some() => MonitorMode::Normal,
         _ => return Err(io::Error::other("unknown fixed case")),
     };
     let id = format!("execution-{case}");
@@ -141,9 +162,36 @@ pub fn run_case(case: &str) -> io::Result<serde_json::Value> {
     if case == "export" {
         request.outputs = vec!["valid.bin".into(), "escape".into()];
     }
+    if case == "fail-io" {
+        request.limits.file_bytes = 65536;
+        request.limits.stdout_bytes = 131072;
+        request.limits.stderr_bytes = 131072;
+        request.stdin_artifact = None;
+    }
+    if case == "fail-inputs" {
+        fs::remove_file(format!("{}/program-001", fixture.publications))?;
+    }
     let spec = request
         .validate(&SpecPolicy::default())
         .map_err(io::Error::other)?;
+    if let Some(stage) = rollback {
+        return crate::launcher::rollback::cancel_stage(
+            stage,
+            &config,
+            &spec,
+            &std::env::current_exe()?,
+            &root,
+        );
+    }
+    if let Some(stage) = failing {
+        return crate::launcher::rollback::fail_stage(
+            stage,
+            &config,
+            &spec,
+            &std::env::current_exe()?,
+            &root,
+        );
+    }
     let completed =
         run::execute(&config, &spec, &std::env::current_exe()?, &root, mode).map_err(|e| {
             io::Error::other(format!(

judge/crates/sandbox/src/cgroup.rs

@@ -174,6 +174,7 @@ impl DelegatedCgroup {
             name,
             limits,
             removed: false,
+            preserve: false,
             baseline: None,
         };
         let setup = (|| {
@@ -210,10 +211,20 @@ pub struct TaskCgroup {
     name: String,
     limits: AppliedLimits,
     removed: bool,
+    preserve: bool,
     baseline: Option<crate::accounting::AccountingSnapshot>,
 }
 
 impl TaskCgroup {
+    pub(crate) fn is_removed(&self) -> bool {
+        self.removed
+    }
+    /// A failed owned wait/cleanup must keep its marker for restart recovery;
+    /// Drop may retry stopping, but may not erase the evidence or reuse identity.
+    pub(crate) fn preserve_on_drop(&mut self) {
+        self.preserve = true;
+    }
+
     pub fn accounting_snapshot(&self) -> io::Result<crate::accounting::AccountingSnapshot> {
         self.live()?;
         self.applied_limits()?;
@@ -402,7 +413,11 @@ impl TaskCgroup {
 impl Drop for TaskCgroup {
     fn drop(&mut self) {
         if !self.removed {
-            let _ = self.kill_and_remove(Duration::from_millis(250));
+            if self.preserve {
+                let _ = self.terminate_tree(Duration::from_millis(250));
+            } else {
+                let _ = self.kill_and_remove(Duration::from_millis(250));
+            }
         }
     }
 }

judge/crates/sandbox/src/launcher/blocked.rs

@@ -303,7 +303,25 @@ impl PreparedTarget {
             let root = DelegatedCgroup::open(config)?;
             let mut group = root.create_task(spec)?;
             if let Err(mut error) = group.enroll(target) {
-                error.cleanup_error = group.kill_and_remove(CLEANUP).err();
+                let stop = group.terminate_tree(CLEANUP).err();
+                let reap = target.terminate().err();
+                #[cfg(feature = "vm-acceptance")]
+                let orphan = if super::reap::is_owned_scope() {
+                    super::reap::reap_current().err()
+                } else {
+                    None
+                };
+                #[cfg(not(feature = "vm-acceptance"))]
+                let orphan: Option<io::Error> = None;
+                let remove = if stop.is_none() && reap.is_none() && orphan.is_none() {
+                    group.remove_empty().err()
+                } else {
+                    None
+                };
+                error.cleanup_error = stop.or(reap).or(orphan).or(remove);
+                if error.cleanup_error.is_some() {
+                    group.preserve_on_drop();
+                }
                 return Err(error);
             }
             Ok::<_, crate::cgroup::CgroupError>(group)
@@ -350,12 +368,14 @@ impl PreparedTarget {
         stage: ExecutionStage,
         error: Option<&io::Error>,
     ) -> InitializationFailure {
-        let group_error = self.cgroup.terminate_tree(CLEANUP).err();
+        let group_error = if self.cgroup.is_removed() {
+            None
+        } else {
+            self.cgroup.terminate_tree(CLEANUP).err()
+        };
         let reap_error = self.target.terminate().err();
         #[cfg(feature = "vm-acceptance")]
-        let orphan_error = if rustix::process::child_subreaper().ok().flatten()
+        let orphan_error = if super::reap::is_owned_scope() {
-            == Some(rustix::process::getpid())
-        {
             super::reap::reap_current().err()
         } else {
             None
@@ -368,6 +388,13 @@ impl PreparedTarget {
             } else {
                 None
             };
+        if group_error.is_some()
+            || reap_error.is_some()
+            || orphan_error.is_some()
+            || remove_error.is_some()
+        {
+            self.cgroup.preserve_on_drop();
+        }
         InitializationFailure {
             failure: failure(stage, ErrorCode::CapabilityUnavailable, error),
             cleanup_failure: group_error
@@ -391,9 +418,14 @@ impl Drop for PreparedTarget {
         if self.cgroup.terminate_tree(CLEANUP).is_ok() && self.target.terminate().is_ok() {
             #[cfg(feature = "vm-acceptance")]
             if super::reap::is_owned_scope() && super::reap::reap_current().is_err() {
+                self.cgroup.preserve_on_drop();
                 return;
             }
-            let _ = self.cgroup.remove_empty();
+            if self.cgroup.remove_empty().is_err() {
+                self.cgroup.preserve_on_drop();
+            }
+        } else if !self.cgroup.is_removed() {
+            self.cgroup.preserve_on_drop();
         }
     }
 }

judge/crates/sandbox/src/launcher/mod.rs

@@ -33,4 +33,6 @@ pub(crate) mod reap;
 #[cfg(all(target_os = "linux", feature = "vm-acceptance"))]
 pub(crate) mod capture;
 #[cfg(all(target_os = "linux", feature = "vm-acceptance"))]
+pub(crate) mod rollback;
+#[cfg(all(target_os = "linux", feature = "vm-acceptance"))]
 pub(crate) mod run;

judge/crates/sandbox/src/launcher/reap.rs

@@ -44,8 +44,8 @@ impl OwnedChildren {
     /// The precondition excludes unrelated children, including other tasks.
     pub(crate) fn finish(&mut self) -> io::Result<u32> {
         let count = reap_current()?;
-        self.finished = true;
         rustix::process::set_child_subreaper(self.previous)?;
+        self.finished = true;
         OWNED_SCOPE.store(false, Ordering::Relaxed);
         Ok(count)
     }

judge/crates/sandbox/src/launcher/rollback.rs

@@ -0,0 +1,120 @@
+//! Fixed VM-only preparation cancellation/exception matrix using actual owners.
+use super::{
+    blocked::PreparedTarget, identity::IdentityReducedTarget, inputs::InputBoundTarget,
+    namespaces::NamespacedTarget, proc_dev::RestrictedRootTarget, reap::OwnedChildren,
+    root::RootedTarget, seccomp::FilteredTarget,
+};
+use crate::{artifacts::ArtifactRoot, runtime::ValidatedRuntimeConfig, spec::ValidatedSpec};
+use std::{io, path::Path};
+fn fail(e: super::blocked::InitializationFailure) -> io::Error {
+    io::Error::other(format!(
+        "{:?} {:?} cleanup {:?}",
+        e.failure.stage, e.failure.errno, e.cleanup_failure
+    ))
+}
+pub(crate) fn cancel_stage(
+    stage: &str,
+    config: &ValidatedRuntimeConfig,
+    spec: &ValidatedSpec,
+    binary: &Path,
+    artifacts: &ArtifactRoot,
+) -> io::Result<serde_json::Value> {
+    let mut children = OwnedChildren::begin()?;
+    match stage {
+        "blocked" => {
+            let mut t = PreparedTarget::prepare(config, spec, binary).map_err(fail)?;
+            let first = t.refuse_at(crate::outcome::ExecutionStage::Preparation, None);
+            let second = t.refuse_at(crate::outcome::ExecutionStage::Preparation, None);
+            if first.cleanup_failure.is_some() || second.cleanup_failure.is_some() {
+                return Err(io::Error::other("idempotent stopped owner cleanup"));
+            }
+        }
+        "namespace" => {
+            drop(NamespacedTarget::prepare(config, spec, binary).map_err(fail)?);
+        }
+        "root" => {
+            drop(RootedTarget::prepare(config, spec, binary).map_err(fail)?);
+        }
+        "proc" => {
+            drop(RestrictedRootTarget::prepare(config, spec, binary).map_err(fail)?);
+        }
+        "inputs" => {
+            drop(InputBoundTarget::prepare(config, spec, binary, artifacts).map_err(fail)?);
+        }
+        "identity" => {
+            drop(IdentityReducedTarget::prepare(config, spec, binary, artifacts).map_err(fail)?);
+        }
+        "seccomp" => {
+            drop(FilteredTarget::prepare(config, spec, binary, artifacts).map_err(fail)?);
+        }
+        "io" | "exec" => {
+            let mut t = FilteredTarget::prepare(config, spec, binary, artifacts).map_err(fail)?;
+            super::capture::private_supervisor()?;
+            let (mut lease, dir) = super::capture::CaptureLease::prepare(config, spec)?;
+            let (pump, stdio) = crate::io::PipePump::prepare(t.take_stdin(), spec, dir)?;
+            if stage == "exec" {
+                t.root_mut().target.prepare_execution(spec, stdio)?;
+            } else {
+                drop(stdio);
+            }
+            let result = t.refuse_at(crate::outcome::ExecutionStage::Exec, None);
+            if result.cleanup_failure.is_some() {
+                return Err(fail(result));
+            }
+            drop(pump);
+            lease.cleanup()?;
+            lease.cleanup()?;
+        }
+        _ => return Err(io::Error::other("fixed preparation stage")),
+    }
+    let n = children.finish()?;
+    Ok(
+        serde_json::json!({"stage":stage,"cancel_before_exec":true,"remaining_owned_children":0,"reaped_after_drop":n,"idempotent":matches!(stage,"blocked"|"io"|"exec")}),
+    )
+}
+pub(crate) fn fail_stage(
+    stage: &str,
+    config: &ValidatedRuntimeConfig,
+    spec: &ValidatedSpec,
+    binary: &Path,
+    artifacts: &ArtifactRoot,
+) -> io::Result<serde_json::Value> {
+    if stage == "io" {
+        let Err(e) = super::run::execute(
+            config,
+            spec,
+            binary,
+            artifacts,
+            super::run::MonitorMode::Normal,
+        ) else {
+            return Err(io::Error::other("expected pipe budget failure"));
+        };
+        if e.failure.stage != crate::outcome::ExecutionStage::Exec || e.cleanup_failure.is_some() {
+            return Err(fail(e));
+        }
+        return Ok(
+            serde_json::json!({"stage":stage,"failure":e.failure,"cleanup_complete":true,"user_exec_released":false}),
+        );
+    }
+    let mut children = OwnedChildren::begin()?;
+    let result = if stage == "inputs" {
+        let target = RestrictedRootTarget::prepare(config, spec, binary).map_err(fail)?;
+        InputBoundTarget::bind(target, artifacts).map(|t| {
+            drop(t);
+        })
+    } else {
+        FilteredTarget::prepare(config, spec, binary, artifacts).map(|t| {
+            drop(t);
+        })
+    };
+    let Err(e) = result else {
+        return Err(io::Error::other("expected actual preparation failure"));
+    };
+    if e.cleanup_failure.is_some() {
+        return Err(fail(e));
+    }
+    let n = children.finish()?;
+    Ok(
+        serde_json::json!({"stage":stage,"failure":e.failure,"cleanup_complete":true,"remaining_owned_children":0,"reaped_after_failure":n,"user_exec_released":false}),
+    )
+}

judge/crates/sandbox/src/launcher/run.rs

@@ -109,11 +109,7 @@ pub(crate) fn execute(
     let (work, mut lease, mut pump) = match setup {
         Ok(s) => s,
         Err(e) => {
-            return Err(target
+            return Err(target.refuse_at(ExecutionStage::Exec, Some(&e)));
-                .root_mut()
-                .target
-                .prepared
-                .refuse_at(ExecutionStage::Exec, Some(&e)));
         }
     };
     let init = target.root().init_pid();
@@ -174,7 +170,6 @@ pub(crate) fn execute(
                             ErrorCode::MeasurementUnavailable,
                             &e.source,
                         ));
-                        monitoring_failed = true;
                         break;
                     }
                 };
@@ -340,7 +335,28 @@ pub(crate) fn execute(
         order.push("owned_reaped");
         // If a main status was only available through adopted wait4, its owned
         // pidfd waitid above records it before the general orphan drain.
-        let final_sample = budget.after_reap();
+        let final_sample = if errors.iter().any(|e| e.stage == ExecutionStage::Cleanup) {
+            // Keep counters unavailable if the required wait ownership proof
+            // failed. Never present an earlier sample as final-after-reap.
+            Err(crate::supervisor::SupervisionFailure {
+                source: io::Error::other("reaping incomplete"),
+                termination_error: None,
+                usage: ResourceUsage {
+                    cpu_time_us: Measurement::Unavailable {
+                        reason: MeasurementFailure::ReadFailed,
+                    },
+                    wall_time_us: latest.wall_time_us.clone(),
+                    memory_peak_bytes: Measurement::Unavailable {
+                        reason: MeasurementFailure::ReadFailed,
+                    },
+                    memory_events: Measurement::Unavailable {
+                        reason: MeasurementFailure::ReadFailed,
+                    },
+                },
+            })
+        } else {
+            budget.after_reap()
+        };
         order.push("final_accounting");
         match final_sample {
             Ok(e) => {
@@ -353,11 +369,7 @@ pub(crate) fn execute(
                     ErrorCode::MeasurementUnavailable,
                     &e.source,
                 ));
-                (
+                (e.usage, Vec::new(), reaped)
-                    if monitoring_failed { e.usage } else { latest },
-                    Vec::new(),
-                    reaped,
-                )
             }
         }
     };

judge/crates/sandbox/src/launcher/seccomp.rs

@@ -43,6 +43,14 @@ impl FilteredTarget {
     pub fn stdin(&self) -> Option<&crate::artifacts::SnapshotFile> {
         self.target.stdin()
     }
+    #[cfg(feature = "vm-acceptance")]
+    pub(crate) fn refuse_at(
+        self,
+        stage: ExecutionStage,
+        error: Option<&io::Error>,
+    ) -> InitializationFailure {
+        self.target.refuse_at(stage, error)
+    }
     pub fn refuse_remaining_isolation(self) -> InitializationFailure {
         self.target.refuse_at(ExecutionStage::Exec, None)
     }

judge/crates/sandbox/tests/fixtures/rollback_deny.c

@@ -0,0 +1,43 @@
+/* Fixed independent-guest syscall failure injector, never an operator helper. */
+#define _GNU_SOURCE
+#include <linux/audit.h>
+#include <linux/filter.h>
+#include <linux/seccomp.h>
+#include <errno.h>
+#include <stddef.h>
+#include <string.h>
+#include <stdlib.h>
+#include <sys/prctl.h>
+#include <sys/syscall.h>
+#include <unistd.h>
+int main(int argc,char **argv){
+ const char *vm=getenv("KF_SANDBOX_INDEPENDENT_VM");if(argc!=2||!vm||strcmp(vm,"1")||access("/kf/INDEPENDENT_VM",R_OK))return 2;
+ long nr=0;int special=0;char *mode=argv[1];
+ if(!strcmp(mode,"namespace")){nr=SYS_unshare;special=1;}
+ else if(!strcmp(mode,"root"))nr=SYS_pivot_root;
+ else if(!strcmp(mode,"proc"))nr=SYS_mknodat;
+ else if(!strcmp(mode,"rlimits")){nr=SYS_prlimit64;special=2;}
+ else if(!strcmp(mode,"identity"))nr=SYS_setgroups;
+ else if(!strcmp(mode,"seccomp"))nr=SYS_seccomp;
+ else return 3;
+ struct sock_filter code[16]={
+ BPF_STMT(BPF_LD|BPF_W|BPF_ABS,offsetof(struct seccomp_data,arch)),
+ BPF_JUMP(BPF_JMP|BPF_JEQ|BPF_K,AUDIT_ARCH_X86_64,1,0),
+ BPF_STMT(BPF_RET|BPF_K,SECCOMP_RET_KILL_PROCESS),
+ BPF_STMT(BPF_LD|BPF_W|BPF_ABS,offsetof(struct seccomp_data,nr)),
+ BPF_JUMP(BPF_JMP|BPF_JEQ|BPF_K,(unsigned)nr,0,1),
+ BPF_STMT(BPF_RET|BPF_K,SECCOMP_RET_ERRNO|EPERM),
+ BPF_STMT(BPF_RET|BPF_K,SECCOMP_RET_ALLOW)};
+ unsigned len=7;
+ if(special){
+  code[4]=(struct sock_filter)BPF_JUMP(BPF_JMP|BPF_JEQ|BPF_K,(unsigned)nr,0,3);
+  code[5]=(struct sock_filter)BPF_STMT(BPF_LD|BPF_W|BPF_ABS,offsetof(struct seccomp_data,args)+(special==1?0:16));
+  code[6]=(struct sock_filter)BPF_JUMP(BPF_JMP|BPF_JSET|BPF_K,special==1?0x20000000U:0xffffffffU,0,1);
+  code[7]=(struct sock_filter)BPF_STMT(BPF_RET|BPF_K,SECCOMP_RET_ERRNO|EPERM);
+  code[8]=(struct sock_filter)BPF_STMT(BPF_RET|BPF_K,SECCOMP_RET_ALLOW);len=9;
+ }
+ struct sock_fprog program={.len=(unsigned short)len,.filter=code};
+ if(prctl(PR_SET_NO_NEW_PRIVS,1,0,0,0)||prctl(PR_SET_SECCOMP,SECCOMP_MODE_FILTER,&program))return 4;
+ char c[32];if(strlen(mode)>15)return 5;strcpy(c,"fail-");strcat(c,mode);
+ execl("/kf/bin/kf-sandbox","kf-sandbox","vm-case",c,(char*)0);return 6;
+}

judge/crates/sandbox/tests/rollback_vm.rs

@@ -0,0 +1,109 @@
+#![cfg(all(target_os = "linux", feature = "vm-acceptance"))]
+#[allow(dead_code)]
+mod support;
+use std::{fs, path::Path, process::Command};
+fn check(case: &str, deny: bool) -> serde_json::Value {
+    support::require_vm();
+    let before = support::children();
+    let output = if deny {
+        Command::new("/kf/bin/rollback-deny")
+            .arg(case.strip_prefix("fail-").unwrap())
+            .env_clear()
+            .env("KF_SANDBOX_INDEPENDENT_VM", "1")
+            .output()
+            .unwrap()
+    } else {
+        Command::new(support::binary())
+            .args(["vm-case", case])
+            .env_clear()
+            .env("KF_SANDBOX_INDEPENDENT_VM", "1")
+            .output()
+            .unwrap()
+    };
+    assert!(
+        output.status.success(),
+        "{case}: {} {}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap();
+    assert_eq!(support::children(), before);
+    for path in [
+        "/var/lib/kf-judge/tasks",
+        "/var/lib/kf-judge/identities",
+        "/var/lib/kf-judge/captures",
+    ] {
+        if Path::new(path).exists() {
+            assert_eq!(fs::read_dir(path).unwrap().count(), 0, "{case} {path}");
+        }
+    }
+    for path in ["/sys/fs/cgroup", "/kf"] {
+        for entry in fs::read_dir(path).unwrap() {
+            let n = entry.unwrap().file_name();
+            assert!(
+                !n.to_string_lossy().starts_with("kf-acceptance-")
+                    && !n.to_string_lossy().starts_with("execution-publications-"),
+                "{case} {n:?}"
+            );
+        }
+    }
+    println!("KF_ROLLBACK_PROOF {value}");
+    value
+}
+#[test]
+#[ignore = "independent capped VM: cancellation/drop after every typed preparation checkpoint"]
+fn cancel_all_preparation_gates_without_exec_or_leaks() {
+    for stage in [
+        "blocked",
+        "namespace",
+        "root",
+        "proc",
+        "inputs",
+        "identity",
+        "seccomp",
+        "io",
+        "exec",
+    ] {
+        let v = check(&format!("cancel-{stage}"), false);
+        assert_eq!(v["cancel_before_exec"], true);
+        assert_eq!(v["remaining_owned_children"], 0);
+        if matches!(stage, "blocked" | "io" | "exec") {
+            assert_eq!(v["idempotent"], true);
+        }
+    }
+}
+#[test]
+#[ignore = "independent capped VM: actual kernel failures in six successive isolation stages"]
+fn real_kernel_stage_failures_rollback_owned_resources() {
+    for (stage, expected) in [
+        ("namespace", "namespaces"),
+        ("root", "root_filesystem"),
+        ("proc", "root_filesystem"),
+        ("rlimits", "preparation"),
+        ("identity", "identity"),
+        ("seccomp", "seccomp"),
+    ] {
+        let v = check(&format!("fail-{stage}"), true);
+        assert_eq!(v["failure"]["stage"], expected, "{v}");
+        assert_eq!(v["failure"]["errno"], 1, "{v}");
+        assert_eq!(v["cleanup_complete"], true);
+        assert_eq!(v["user_exec_released"], false);
+    }
+}
+#[test]
+#[ignore = "independent capped VM: post-root missing input and post-filter pipe budget failure"]
+fn late_data_failures_preserve_stage_and_clean() {
+    for case in ["fail-inputs", "fail-io"] {
+        let v = check(case, false);
+        assert_eq!(v["cleanup_complete"], true);
+        assert_eq!(v["user_exec_released"], false);
+        assert_eq!(
+            v["failure"]["stage"],
+            if case == "fail-inputs" {
+                "root_filesystem"
+            } else {
+                "exec"
+            }
+        );
+    }
+}