feat(sandbox): add immutable Python and Java execution profiles

backend/tests/test_judging_image_builder.py

@@ -0,0 +1,63 @@
+"""Image construction data tests; actual publication/execution needs the independent VM."""
+
+import importlib.util
+import io
+import json
+import tarfile
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+MODULE = Path(__file__).resolve().parents[2] / "scripts/build_judging_images.py"
+SPEC = importlib.util.spec_from_file_location("judging_image_builder", MODULE)
+builder = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(builder)
+
+
+def test_catalog_uses_wire_path_order_including_dot_and_directory_boundaries(tmp_path):
+    image = tmp_path / "image"
+    image.mkdir()
+    builder.put(image, "usr/lib/module/a.py", b"data")
+    builder.put(image, "usr/lib/module.py", b"module")
+    support = tmp_path / "support"
+    library = support / "usr/lib/ld-linux-x86-64.so.2"
+    library.parent.mkdir(parents=True)
+    library.write_bytes(b"synthetic ordinary loader data")
+    args = SimpleNamespace(support_dir=support, support_manifest_sha256="a" * 64)
+    builder.finish(
+        args,
+        builder.PINS["python313"],
+        image,
+        {"usr/lib/ld-linux-x86-64.so.2": {"kind": "file", "sha256": builder.sha(library)}},
+    )
+    manifest = json.loads((image / "manifest.json").read_bytes())
+    paths = [item["path"] for item in manifest["files"]]
+    assert paths == sorted(paths)
+    assert paths.index("usr/lib/module.py") < paths.index("usr/lib/module/a.py")
+
+
+def test_public_archive_links_cannot_escape_or_cycle():
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w") as archive:
+        for name, target in [("python/outside", "../../secret"), ("python/cycle", "cycle")]:
+            item = tarfile.TarInfo(name)
+            item.type = tarfile.SYMTYPE
+            item.linkname = target
+            archive.addfile(item)
+    buffer.seek(0)
+    with tarfile.open(fileobj=buffer, mode="r:") as archive:
+        members = {item.name: item for item in archive.getmembers()}
+        for name in members:
+            with pytest.raises(ValueError):
+                builder.member_bytes(archive, members, name)
+
+
+def test_corrupt_archive_is_refused_before_output_creation(tmp_path):
+    archive = tmp_path / "public.tar.gz"
+    archive.write_bytes(b"untrusted replacement")
+    output = tmp_path / "image"
+    args = SimpleNamespace(language="python313", archive=archive, output=output)
+    with pytest.raises(ValueError, match="fixed upstream digest"):
+        builder.build(args)
+    assert not output.exists()

judge/crates/contracts/src/judging_v2.rs

@@ -311,7 +311,7 @@ impl CompiledProgram {
         Ok(())
     }
 }
-fn class_name(path: &str) -> bool {
+pub fn class_name(path: &str) -> bool {
     let Some(stem) = path.strip_suffix(".class") else {
         return false;
     };

judge/crates/sandbox/Cargo.toml

@@ -15,6 +15,7 @@ kf-contracts.workspace = true
 clap.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+sha2 = "=0.10.9"
 rustix = { version = "1.1.5", features = ["net", "process", "fs", "thread", "mount", "system", "stdio", "pipe", "event"] }
 
 # Fixed BPF compilation/installation through safe APIs; libc supplies syscall
@@ -29,6 +30,3 @@ workspace = true
 [[example]]
 name = "identity_probe"
 test = false
-
-[dev-dependencies]
-sha2 = "=0.10.9"

judge/crates/sandbox/runtime/python_check.py

@@ -0,0 +1,28 @@
+"""Trusted, isolated syntax checker; never evaluates submitted code or writes pyc."""
+
+import sys
+
+
+def main():
+    try:
+        with open("/work/main.py", "rb") as source:
+            text = source.read(262145)
+        if len(text) > 262144:
+            print("source exceeds syntax checker byte ceiling", file=sys.stderr)
+            return 65
+        # compile(bytes) honors Python's BOM/encoding declaration without
+        # interpreting any top-level statement or importing user modules.
+        compile(text, "/work/main.py", "exec", dont_inherit=True)
+    except (SyntaxError, UnicodeError, ValueError) as error:
+        # Source belongs to the submitting user. Hidden test input is never
+        # supplied to this stage; diagnostic capture remains OS bounded.
+        print(f"{type(error).__name__}: {error}", file=sys.stderr)
+        return 1
+    except OSError:
+        print("trusted syntax checker input unavailable", file=sys.stderr)
+        return 70
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

judge/crates/sandbox/src/acceptance/isolation.rs

@@ -133,6 +133,7 @@ fn matrix_case(case: &str) -> io::Result<serde_json::Value> {
         ExecutionProfile::Native => "native",
         ExecutionProfile::CompileC => "compile_c",
         ExecutionProfile::CompileCpp => "compile_cpp",
+        _ => return Err(io::Error::other("GNU isolation probe profile")),
     };
     if profile == ExecutionProfile::Native {
         request.argv.extend([mode.into(), label.into()]);

judge/crates/sandbox/src/artifacts.rs

@@ -5,7 +5,7 @@
 //! Outputs are resolved only through a verified, frozen target's work FD.
 //! Snapshotting is not publication, input mounting, stdio wiring or an exec permit.
 
-use crate::spec::{ValidatedSpec, path};
+use crate::spec::{ValidatedSpec, artifact_path};
 use rustix::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd};
 use rustix::fs::{FileType, MemfdFlags, Mode, OFlags, ResolveFlags, SealFlags, Stat};
 use std::fs::File;
@@ -215,6 +215,7 @@ impl ArtifactRoot {
 pub(crate) struct OutputPlan {
     paths: Vec<String>,
     budget: SnapshotBudget,
+    java_classes: bool,
 }
 
 impl OutputPlan {
@@ -222,6 +223,7 @@ impl OutputPlan {
         Self {
             paths: spec.spec().outputs.clone(),
             budget: SnapshotBudget::for_spec(spec),
+            java_classes: spec.spec().profile == crate::spec::ExecutionProfile::JavaCompile,
         }
     }
 
@@ -231,12 +233,43 @@ impl OutputPlan {
         {
             return Err(refused("invalid output directory or count"));
         }
+        let paths = if self.java_classes {
+            let mut paths = Vec::new();
+            let mut count = 0;
+            // The entire group is frozen and drained. Discover only bounded,
+            // flat compiler outputs through the held private work directory.
+            for entry in std::fs::read_dir(format!("/proc/self/fd/{}", work.as_raw_fd()))? {
+                count += 1;
+                if count > kf_contracts::judging_v2::MAX_CLASS_FILES + 1 {
+                    return Err(refused("class export count"));
+                }
+                let name = entry?
+                    .file_name()
+                    .into_string()
+                    .map_err(|_| refused("class export name"))?;
+                if name == "Main.java" {
+                    continue;
+                }
+                if !kf_contracts::judging_v2::class_name(&name) {
+                    return Err(refused("class export path/type"));
+                }
+                paths.push(name);
+            }
+            paths.sort();
+            paths
+        } else {
+            self.paths.clone()
+        };
         let mut budget = self.budget;
-        let mut files = Vec::with_capacity(self.paths.len());
-        for path in &self.paths {
+        let mut files = Vec::with_capacity(paths.len());
+        for path in &paths {
+            let file = snapshot(work, path, &mut budget, None, false)?;
+            if self.java_classes && file.bytes() == 0 {
+                return Err(refused("empty class export"));
+            }
             files.push(OutputSnapshot {
                 path: path.clone(),
-                file: snapshot(work, path, &mut budget, None, false)?,
+                file,
             });
         }
         // Errors drop every partial FD; there is no partially published batch.
@@ -309,7 +342,7 @@ fn snapshot(
     owner: Option<u32>,
     executable: bool,
 ) -> io::Result<SnapshotFile> {
-    path(relative, "artifact").map_err(|_| refused("noncanonical artifact path"))?;
+    artifact_path(relative, "artifact").map_err(|_| refused("noncanonical artifact path"))?;
     // Atomic containment; reject unsupported openat2 or races, no fallback or
     // retry. O_PATH inspects special files without opening their device/pipe.
     let fd = rustix::fs::openat2(

judge/crates/sandbox/src/execution.rs

@@ -154,12 +154,17 @@ pub(crate) fn authorize(config: &ValidatedRuntimeConfig) -> io::Result<()> {
     if st.st_uid != 0 || st.st_gid != 0 || st.st_mode & 0o7022 != 0 {
         return Err(invalid("image manifest ownership"));
     }
-    let m: serde_json::Value = serde_json::from_slice(&read_pinned(manifest, 8192)?)
-        .map_err(|_| invalid("image manifest JSON"))?;
+    let bytes = read_pinned(manifest, crate::image::MAX_CATALOG_BYTES)?;
+    if let Some(identity) = &c.runtime_identity {
+        crate::image::verify(&image, identity, &bytes)?;
+    } else {
+        let m: serde_json::Value =
+            serde_json::from_slice(&bytes).map_err(|_| invalid("image manifest JSON"))?;
         if m != serde_json::json!({"image_kind":"linux_gnu_cxx_x86_64_v1","architecture":"x86_64","revision":1})
         {
             return Err(invalid("unsupported image manifest"));
         }
+    }
     let own = std::fs::read_to_string("/proc/self/cgroup")?;
     let path = own
         .lines()

judge/crates/sandbox/src/image.rs

@@ -0,0 +1,266 @@
+//! Root-owned, immutable directory images pinned by their complete catalog.
+//! All descriptor walking, ownership and byte hashing stays in Sandbox.
+use kf_contracts::{
+    judging::{Digest, Label},
+    judging_v2::{ImageKind, RuntimeIdentity, RuntimeVersion},
+};
+use rustix::{
+    fd::{AsRawFd, OwnedFd},
+    fs::{FileType, Mode, OFlags},
+};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest as _, Sha256};
+use std::{
+    collections::BTreeMap,
+    fs::File,
+    io::{self, Read},
+};
+
+pub const MAX_CATALOG_BYTES: u64 = 1024 * 1024;
+const MAX_FILES: usize = 4096;
+const MAX_IMAGE_BYTES: u64 = 1024 * 1024 * 1024;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ImageFile {
+    pub path: String,
+    pub bytes: u64,
+    pub sha256: Digest,
+    pub executable: bool,
+}
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ImageCatalog {
+    pub format_version: u16,
+    pub architecture: String,
+    pub image_kind: ImageKind,
+    pub image_revision: u32,
+    pub toolchain_id: Label,
+    pub runtime_version: RuntimeVersion,
+    pub directories: Vec<String>,
+    pub files: Vec<ImageFile>,
+}
+fn refused() -> io::Error {
+    io::Error::new(
+        io::ErrorKind::PermissionDenied,
+        "image catalog/integrity refused",
+    )
+}
+pub fn catalog_digest(data: &[u8]) -> Digest {
+    format!("{:x}", Sha256::digest(data))
+        .try_into()
+        .expect("SHA256")
+}
+impl ImageCatalog {
+    pub fn validate(&self, identity: &RuntimeIdentity) -> io::Result<()> {
+        let language = match self.image_kind {
+            ImageKind::GnuCxx => kf_contracts::judging_v2::Language::C17,
+            ImageKind::CPython313 => kf_contracts::judging_v2::Language::Python313,
+            ImageKind::OpenJdk21 => kf_contracts::judging_v2::Language::Java21,
+        };
+        identity
+            .validate_for_language(language)
+            .map_err(|_| refused())?;
+        if self.format_version != 2
+            || self.architecture != "x86_64"
+            || self.image_kind != identity.image_kind
+            || self.image_revision != identity.image_revision
+            || self.toolchain_id != identity.toolchain_id
+            || self.runtime_version != identity.runtime_version
+            || self.files.is_empty()
+            || self.files.len() > MAX_FILES
+            || self.directories.len() > MAX_FILES
+            || self.directories.windows(2).any(|v| v[0] >= v[1])
+            || self.files.windows(2).any(|v| v[0].path >= v[1].path)
+        {
+            return Err(refused());
+        }
+        let mut total = 0_u64;
+        for name in self
+            .directories
+            .iter()
+            .chain(self.files.iter().map(|v| &v.path))
+        {
+            crate::spec::path(name, "image").map_err(|_| refused())?;
+            if name == "manifest.json"
+                || name
+                    .split('/')
+                    .any(|v| matches!(v, "site-packages" | "__pycache__"))
+                || name.ends_with(".pth")
+                || name.ends_with("/sitecustomize.py")
+                || name.ends_with("/usercustomize.py")
+            {
+                return Err(refused());
+            }
+        }
+        for file in &self.files {
+            total = total
+                .checked_add(file.bytes)
+                .filter(|v| *v <= MAX_IMAGE_BYTES)
+                .ok_or_else(refused)?;
+        }
+        if self
+            .files
+            .iter()
+            .any(|f| self.directories.contains(&f.path))
+            || ["dev", "proc", "tmp", "work"]
+                .iter()
+                .any(|p| !self.directories.iter().any(|d| d == p))
+        {
+            return Err(refused());
+        }
+        let required: &[&str] = match self.image_kind {
+            ImageKind::CPython313 => &[
+                "usr/bin/python3.13",
+                "usr/lib/kf/python_check.py",
+                "usr/lib/python3.13/os.py",
+                "lib64/ld-linux-x86-64.so.2",
+            ],
+            ImageKind::OpenJdk21 => &[
+                "usr/lib/jvm/bin/java",
+                "usr/lib/jvm/bin/javac",
+                "usr/lib/jvm/lib/modules",
+                "usr/lib/jvm/lib/ct.sym",
+                "lib64/ld-linux-x86-64.so.2",
+            ],
+            ImageKind::GnuCxx => &["usr/bin/gcc", "usr/bin/g++", "lib64/ld-linux-x86-64.so.2"],
+        };
+        if required
+            .iter()
+            .any(|p| !self.files.iter().any(|f| f.path == *p && f.bytes != 0))
+        {
+            return Err(refused());
+        }
+        Ok(())
+    }
+}
+
+/// Every inode is checked, including unexpected entries and empty directories.
+/// No symlinks, hard links, writable files, special files or mount crossings.
+pub(crate) fn verify(image: &OwnedFd, identity: &RuntimeIdentity, bytes: &[u8]) -> io::Result<()> {
+    if bytes.len() as u64 > MAX_CATALOG_BYTES || catalog_digest(bytes) != identity.image_digest {
+        return Err(refused());
+    }
+    let catalog: ImageCatalog = serde_json::from_slice(bytes).map_err(|_| refused())?;
+    catalog.validate(identity)?;
+    let mut files: BTreeMap<_, _> = catalog.files.iter().map(|f| (f.path.as_str(), f)).collect();
+    let mut dirs: std::collections::BTreeSet<_> =
+        catalog.directories.iter().map(String::as_str).collect();
+    let root = rustix::fs::fstat(image)?;
+    if root.st_uid != 0 || root.st_gid != 0 || root.st_mode & 0o7777 != 0o555 {
+        return Err(refused());
+    }
+    walk(image, "", root.st_dev, &mut files, &mut dirs)?;
+    if !files.is_empty() || !dirs.is_empty() {
+        return Err(refused());
+    }
+    Ok(())
+}
+fn walk<'a>(
+    directory: &OwnedFd,
+    prefix: &str,
+    device: u64,
+    files: &mut BTreeMap<&'a str, &'a ImageFile>,
+    dirs: &mut std::collections::BTreeSet<&'a str>,
+) -> io::Result<()> {
+    if prefix.split('/').count() > 32 {
+        return Err(refused());
+    }
+    for entry in std::fs::read_dir(format!("/proc/self/fd/{}", directory.as_raw_fd()))? {
+        let name = entry?.file_name();
+        let name = name.to_str().ok_or_else(refused)?;
+        let path = if prefix.is_empty() {
+            name.into()
+        } else {
+            format!("{prefix}/{name}")
+        };
+        let fd = rustix::fs::openat(
+            directory,
+            name,
+            OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::empty(),
+        )?;
+        let st = rustix::fs::fstat(&fd)?;
+        if st.st_uid != 0 || st.st_gid != 0 || st.st_dev != device {
+            return Err(refused());
+        }
+        match FileType::from_raw_mode(st.st_mode) {
+            FileType::Directory => {
+                if st.st_mode & 0o7777 != 0o555 || !dirs.remove(path.as_str()) {
+                    return Err(refused());
+                }
+                let child = rustix::fs::openat(
+                    directory,
+                    name,
+                    OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+                    Mode::empty(),
+                )?;
+                let after = rustix::fs::fstat(&child)?;
+                if st.st_dev != after.st_dev || st.st_ino != after.st_ino {
+                    return Err(refused());
+                }
+                walk(&child, &path, device, files, dirs)?;
+            }
+            FileType::RegularFile => {
+                if st.st_nlink != 1 || st.st_size < 0 {
+                    return Err(refused());
+                }
+                if path == "manifest.json" {
+                    if st.st_mode & 0o7777 != 0o444 {
+                        return Err(refused());
+                    }
+                    continue;
+                }
+                let expected = files.remove(path.as_str()).ok_or_else(refused)?;
+                if st.st_size as u64 != expected.bytes
+                    || st.st_mode & 0o7777 != if expected.executable { 0o555 } else { 0o444 }
+                {
+                    return Err(refused());
+                }
+                let read = rustix::fs::open(
+                    format!("/proc/self/fd/{}", fd.as_raw_fd()),
+                    OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NONBLOCK | OFlags::NOCTTY,
+                    Mode::empty(),
+                )?;
+                let after = rustix::fs::fstat(&read)?;
+                if st.st_dev != after.st_dev || st.st_ino != after.st_ino {
+                    return Err(refused());
+                }
+                let mut reader = File::from(read);
+                let mut hash = Sha256::new();
+                let mut count = 0_u64;
+                let mut buffer = [0_u8; 65536];
+                loop {
+                    let n = reader.read(&mut buffer)?;
+                    if n == 0 {
+                        break;
+                    }
+                    count += n as u64;
+                    if count > expected.bytes {
+                        return Err(refused());
+                    }
+                    hash.update(&buffer[..n]);
+                }
+                if count != expected.bytes
+                    || format!("{:x}", hash.finalize()) != expected.sha256.as_str()
+                {
+                    return Err(refused());
+                }
+                let final_stat = reader.metadata()?;
+                use std::os::unix::fs::MetadataExt;
+                if final_stat.dev() != st.st_dev
+                    || final_stat.ino() != st.st_ino
+                    || final_stat.nlink() != 1
+                    || final_stat.len() != expected.bytes
+                    || final_stat.mode() & 0o7777 != st.st_mode & 0o7777
+                    || final_stat.uid() != 0
+                    || final_stat.gid() != 0
+                {
+                    return Err(refused());
+                }
+            }
+            _ => return Err(refused()),
+        }
+    }
+    Ok(())
+}

judge/crates/sandbox/src/isolation/inputs.rs

@@ -1,5 +1,5 @@
 //! Fixed pre-exec read-only bindings. No task code exists during this stage.
-use crate::spec::path;
+use crate::spec::artifact_path;
 use rustix::fd::OwnedFd;
 use rustix::fs::{FileType, Mode, OFlags, ResolveFlags, SealFlags, StatVfsMountFlags};
 use rustix::mount::MountFlags;
@@ -37,7 +37,7 @@ pub(crate) fn bind(relative: &str, fd: &OwnedFd, bytes: u64, executable: bool) -
     if rustix::process::getpid().as_raw_pid() != 1 || !rustix::process::geteuid().is_root() {
         return Err(denied("input binding requires privileged namespace init"));
     }
-    path(relative, "input").map_err(|_| denied("input binding path"))?;
+    artifact_path(relative, "input").map_err(|_| denied("input binding path"))?;
     let source = rustix::fs::fstat(fd)?;
     let seals =
         SealFlags::WRITE | SealFlags::GROW | SealFlags::SHRINK | SealFlags::EXEC | SealFlags::SEAL;

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

@@ -169,7 +169,7 @@ pub(crate) fn compile(profile: ExecutionProfile) -> io::Result<BpfProgram> {
             thread_required,
         )?],
     ];
-    if profile != ExecutionProfile::Native {
+    if profile.gnu_compiler() {
         clones.push(vec![condition(
             0,
             SeccompCmpOp::Eq,
@@ -252,11 +252,20 @@ pub(crate) fn compile(profile: ExecutionProfile) -> io::Result<BpfProgram> {
         condition(2, SeccompCmpOp::MaskedEq(!flags), 0)?,
     ]);
     allowed.insert(libc::SYS_fcntl, rules(fcntl)?);
-    // Only non-mutating tty/pipe queries; no terminal injection/device control.
+    // Fixed tty/pipe queries. CPython's actual fopen path also needs FIOCLEX:
+    // it only sets close-on-exec on a descriptor already owned by this task.
+    // FIONCLEX, terminal injection and arbitrary device ioctls stay denied.
+    let mut ioctls = vec![libc::TCGETS, libc::FIONREAD];
+    if matches!(
+        profile,
+        ExecutionProfile::PythonCheck | ExecutionProfile::PythonRun
+    ) {
+        ioctls.extend([libc::TCGETS2, libc::FIOCLEX]);
+    }
     allowed.insert(
         libc::SYS_ioctl,
         rules(
-            [libc::TCGETS, libc::FIONREAD]
+            ioctls
                 .into_iter()
                 .map(|op| Ok(vec![condition(1, SeccompCmpOp::Eq, op)?]))
                 .collect::<io::Result<_>>()?,
@@ -403,6 +412,15 @@ pub(crate) fn install(profile: ExecutionProfile) -> io::Result<()> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    const PROFILES: [ExecutionProfile; 7] = [
+        ExecutionProfile::Native,
+        ExecutionProfile::CompileC,
+        ExecutionProfile::CompileCpp,
+        ExecutionProfile::PythonCheck,
+        ExecutionProfile::PythonRun,
+        ExecutionProfile::JavaCompile,
+        ExecutionProfile::JavaRun,
+    ];
     // Evaluate generated classic BPF against synthetic seccomp_data; this is
     // an ordinary policy regression, not evidence of kernel installation.
     fn evaluate(p: &[sock_filter], arch: u32, nr: u32, args: [u64; 6]) -> u32 {
@@ -435,11 +453,7 @@ mod tests {
     }
     #[test]
     fn abi_unknown_calls_and_clone_flags_fail_closed() {
-        for profile in [
-            ExecutionProfile::Native,
-            ExecutionProfile::CompileC,
-            ExecutionProfile::CompileCpp,
-        ] {
+        for profile in PROFILES {
             let p = compile(profile).unwrap();
             let deny = libc::SECCOMP_RET_ERRNO | libc::EPERM as u32;
             assert_eq!(
@@ -476,16 +490,12 @@ mod tests {
     #[test]
     fn compile_profile_exception_does_not_widen_native() {
         let flags = (libc::CLONE_VM | libc::CLONE_VFORK | libc::SIGCHLD) as u64;
-        for profile in [
-            ExecutionProfile::Native,
-            ExecutionProfile::CompileC,
-            ExecutionProfile::CompileCpp,
-        ] {
+        for profile in PROFILES {
             let p = compile(profile).unwrap();
-            let expected = if profile == ExecutionProfile::Native {
-                libc::SECCOMP_RET_ERRNO | libc::EPERM as u32
-            } else {
+            let expected = if profile.gnu_compiler() {
                 libc::SECCOMP_RET_ALLOW
+            } else {
+                libc::SECCOMP_RET_ERRNO | libc::EPERM as u32
             };
             assert_eq!(
                 evaluate(
@@ -505,11 +515,7 @@ mod tests {
 
     #[test]
     fn dangerous_calls_and_pointer_or_high_bit_argument_bypasses_are_denied() {
-        for profile in [
-            ExecutionProfile::Native,
-            ExecutionProfile::CompileC,
-            ExecutionProfile::CompileCpp,
-        ] {
+        for profile in PROFILES {
             let p = compile(profile).unwrap();
             let deny = libc::SECCOMP_RET_ERRNO | libc::EPERM as u32;
             for nr in [
@@ -574,4 +580,51 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn python_descriptor_compatibility_never_grants_other_ioctl_operations() {
+        for profile in PROFILES {
+            let program = compile(profile).unwrap();
+            let denied = libc::SECCOMP_RET_ERRNO | libc::EPERM as u32;
+            let expected = if matches!(
+                profile,
+                ExecutionProfile::PythonCheck | ExecutionProfile::PythonRun
+            ) {
+                libc::SECCOMP_RET_ALLOW
+            } else {
+                denied
+            };
+            for request in [libc::FIOCLEX, libc::TCGETS2] {
+                assert_eq!(
+                    evaluate(
+                        &program,
+                        AUDIT_X86_64,
+                        libc::SYS_ioctl as u32,
+                        [3, request, 0, 0, 0, 0]
+                    ),
+                    expected
+                );
+                assert_eq!(
+                    evaluate(
+                        &program,
+                        AUDIT_X86_64,
+                        libc::SYS_ioctl as u32,
+                        [3, (1 << 32) | request, 0, 0, 0, 0]
+                    ),
+                    denied
+                );
+            }
+            for request in [libc::FIONCLEX, libc::TIOCSTI] {
+                assert_eq!(
+                    evaluate(
+                        &program,
+                        AUDIT_X86_64,
+                        libc::SYS_ioctl as u32,
+                        [3, request, 0, 0, 0, 0]
+                    ),
+                    denied
+                );
+            }
+        }
+    }
 }

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

@@ -130,10 +130,11 @@ impl WireMessage for NamespaceCommand {
             }
             .valid(),
             Self::PrepareInputs { count, bytes } => {
-                *count <= 64 && *bytes <= crate::artifacts::MAX_SNAPSHOT_BYTES
+                *count <= kf_contracts::judging_v2::MAX_CLASS_FILES as u32
+                    && *bytes <= crate::artifacts::MAX_SNAPSHOT_BYTES
             }
             Self::AttachInput { path, bytes, .. } => {
-                crate::spec::path(path, "input").is_ok()
+                crate::spec::artifact_path(path, "input").is_ok()
                     && *bytes <= crate::artifacts::MAX_SNAPSHOT_BYTES
             }
             Self::PrepareIdentity { uid, gid } => crate::isolation::identity::valid(*uid, *gid),
@@ -518,6 +519,53 @@ impl Write for FixedBuffer {
 mod tests {
     use super::*;
 
+    #[test]
+    fn class_bindings_allow_flat_inner_names_and_a_bounded_complete_set() {
+        for name in [
+            "Main.class",
+            "Main$1.class",
+            "Main$Inner.class",
+            "Helper.class",
+        ] {
+            assert!(
+                NamespaceCommand::AttachInput {
+                    path: name.into(),
+                    bytes: 1,
+                    executable: false,
+                }
+                .valid()
+            );
+        }
+        for name in [
+            "../Main.class",
+            "pkg/Main$Inner.class",
+            "Main$1.class/child",
+        ] {
+            assert!(
+                !NamespaceCommand::AttachInput {
+                    path: name.into(),
+                    bytes: 1,
+                    executable: false,
+                }
+                .valid()
+            );
+        }
+        assert!(
+            NamespaceCommand::PrepareInputs {
+                count: 256,
+                bytes: 1024
+            }
+            .valid()
+        );
+        assert!(
+            !NamespaceCommand::PrepareInputs {
+                count: 257,
+                bytes: 1024
+            }
+            .valid()
+        );
+    }
+
     fn pair() -> (ControlChannel, ControlChannel) {
         let (a, b) = UnixStream::pair().unwrap();
         let peer = ExpectedPeer {

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

@@ -510,10 +510,8 @@ pub(crate) fn execute(
     }
     let stdout = SnapshotFile::from_capture(pump.stdout());
     let stderr = SnapshotFile::from_capture(pump.stderr());
-    let compile_failed = matches!(
-        spec.spec().profile,
-        crate::spec::ExecutionProfile::CompileC | crate::spec::ExecutionProfile::CompileCpp
-    ) && (main != Some(Termination::Exited { code: 0 })
+    let compile_failed = spec.spec().profile.prepares_program()
+        && (main != Some(Termination::Exited { code: 0 })
             || cancelled.is_some()
             || !limits.is_empty()
             || !errors.is_empty());

judge/crates/sandbox/src/lib.rs

@@ -1,9 +1,12 @@
 //! Linux execution boundary: policy, kernel gates, bounded supervision and recovery.
 //! Ordinary operator/Worker entries require trusted authority and active isolation.
 
+#[cfg(target_os = "linux")]
+pub mod image;
 pub mod launcher;
 pub mod outcome;
 pub mod probe;
+pub mod profiles;
 pub mod runtime;
 pub mod spec;
 

judge/crates/sandbox/src/outcome.rs

@@ -281,14 +281,28 @@ impl ExecutionReport {
         check_capture(&self.stderr, limits.stderr_bytes)?;
         let mut artifact_paths = std::collections::BTreeSet::new();
         let mut artifact_bytes = 0_u64;
+        let java_classes = spec.spec().profile == crate::spec::ExecutionProfile::JavaCompile;
+        if self.artifacts.len()
+            > if java_classes {
+                kf_contracts::judging_v2::MAX_CLASS_FILES
+            } else {
+                16
+            }
+            || (java_classes && self.artifacts.windows(2).any(|v| v[0].path >= v[1].path))
+        {
+            return Err(ReportError("artifact_count_or_order"));
+        }
         for artifact in &self.artifacts {
             artifact_bytes = artifact_bytes
                 .checked_add(artifact.bytes)
                 .ok_or(ReportError("artifacts"))?;
-            if !spec.spec().outputs.contains(&artifact.path)
+            if !(spec.spec().outputs.contains(&artifact.path)
+                || (spec.spec().profile == crate::spec::ExecutionProfile::JavaCompile
+                    && kf_contracts::judging_v2::class_name(&artifact.path)))
                 || !artifact_paths.insert(artifact.path.as_str())
                 || !artifact_id(&artifact.artifact_id)
                 || artifact.bytes > limits.file_bytes
+                || (java_classes && artifact.bytes == 0)
                 || artifact_bytes > limits.workdir_bytes
             {
                 return Err(ReportError("artifacts"));

judge/crates/sandbox/src/profiles.rs

@@ -0,0 +1,133 @@
+//! Closed interpreter/JDK commands. Operator image selection remains separate;
+//! validation alone neither acquires authority nor proves OS isolation.
+use crate::spec::{ExecutionProfile as P, ExecutionSpec, ResourceLimits};
+use kf_contracts::{judging::RunLimits, judging_v2 as v2};
+
+fn contract_limits(r: &ResourceLimits) -> RunLimits {
+    RunLimits {
+        cpu_time_us: r.cpu_time_us,
+        wall_time_us: r.wall_time_us,
+        memory_bytes: r.memory_bytes,
+        stack_bytes: r.stack_bytes,
+        processes: r.processes,
+        open_files: r.open_files,
+        stdout_bytes: r.stdout_bytes,
+        stderr_bytes: r.stderr_bytes,
+        file_bytes: r.file_bytes,
+        workdir_bytes: r.workdir_bytes,
+    }
+}
+
+/// Explicit heap, native overhead and fixed CPU/thread ergonomics. The kernel
+/// still bounds the complete JVM; printed OOM text is never kernel evidence.
+fn jvm_options(limits: &ResourceLimits) -> Result<Vec<String>, &'static str> {
+    let heap = v2::jvm_heap_bytes(&contract_limits(limits))? / (1024 * 1024);
+    Ok(vec![
+        "-XX:+UseSerialGC".into(),
+        "-XX:ActiveProcessorCount=1".into(),
+        "-XX:-UseContainerSupport".into(),
+        "-XX:CICompilerCount=2".into(),
+        "-XX:-UsePerfData".into(),
+        "-XX:+DisableAttachMechanism".into(),
+        "-XX:MaxMetaspaceSize=64m".into(),
+        "-XX:CompressedClassSpaceSize=32m".into(),
+        "-XX:ReservedCodeCacheSize=32m".into(),
+        "-Xss256k".into(),
+        "-Xms16m".into(),
+        format!("-Xmx{heap}m"),
+        "-Djava.io.tmpdir=/tmp".into(),
+        "-Dfile.encoding=UTF-8".into(),
+        "-Duser.language=en".into(),
+        "-Duser.country=US".into(),
+        "-Duser.timezone=UTC".into(),
+    ])
+}
+
+pub fn argv(profile: P, limits: &ResourceLimits) -> Result<Vec<String>, &'static str> {
+    let (language, preparing) = match profile {
+        P::PythonCheck => (v2::Language::Python313, true),
+        P::PythonRun => (v2::Language::Python313, false),
+        P::JavaCompile => (v2::Language::Java21, true),
+        P::JavaRun => (v2::Language::Java21, false),
+        _ => return Err("profile has no managed command"),
+    };
+    v2::validate_startup_budget(language, &contract_limits(limits), preparing)?;
+    match profile {
+        P::PythonCheck | P::PythonRun => Ok([
+            "/usr/bin/python3.13",
+            "-I",
+            "-S",
+            "-B",
+            if profile == P::PythonCheck {
+                "/usr/lib/kf/python_check.py"
+            } else {
+                "/work/main.py"
+            },
+        ]
+        .map(String::from)
+        .to_vec()),
+        P::JavaCompile => {
+            let mut args = vec!["/usr/lib/jvm/bin/javac".into()];
+            args.extend(jvm_options(limits)?.into_iter().map(|a| format!("-J{a}")));
+            args.extend(
+                [
+                    "-proc:none",
+                    "-implicit:none",
+                    "-encoding",
+                    "UTF-8",
+                    "--release",
+                    "21",
+                    "-classpath",
+                    "/work",
+                    "-sourcepath",
+                    "/work",
+                    "-d",
+                    "/work",
+                    "/work/Main.java",
+                ]
+                .map(String::from),
+            );
+            Ok(args)
+        }
+        P::JavaRun => {
+            let mut args = vec!["/usr/lib/jvm/bin/java".into()];
+            args.extend(jvm_options(limits)?);
+            args.extend(["-cp", "/work", "Main"].map(String::from));
+            Ok(args)
+        }
+        _ => unreachable!(),
+    }
+}
+
+pub(crate) fn validate(spec: &ExecutionSpec) -> Result<(), &'static str> {
+    if spec.argv != argv(spec.profile, &spec.limits)?
+        || !spec.env.is_empty()
+        || spec.cwd != "."
+        || !spec.outputs.is_empty()
+    {
+        return Err("managed profile differs from trusted command");
+    }
+    if spec.profile == P::JavaRun {
+        if spec.inputs.is_empty()
+            || spec.inputs.len() > v2::MAX_CLASS_FILES
+            || !spec.inputs.iter().any(|f| f.path == "Main.class")
+            || spec.inputs.iter().any(|f| !v2::class_name(&f.path))
+            || spec.inputs.windows(2).any(|f| f[0].path >= f[1].path)
+        {
+            return Err("invalid Java class input set");
+        }
+    } else {
+        let expected = if spec.profile == P::JavaCompile {
+            "Main.java"
+        } else {
+            "main.py"
+        };
+        if spec.inputs.len() != 1 || spec.inputs[0].path != expected {
+            return Err("managed profile requires fixed source");
+        }
+    }
+    if spec.profile.prepares_program() && spec.stdin_artifact.is_some() {
+        return Err("preparation does not consume test input");
+    }
+    Ok(())
+}

judge/crates/sandbox/src/rpc.rs

@@ -56,7 +56,7 @@ impl WireMessage for Reply {
     fn valid(&self) -> bool {
         match self {
             Self::Completed { report } => {
-                report.artifacts.len() <= 16
+                report.artifacts.len() <= kf_contracts::judging_v2::MAX_CLASS_FILES
                     && report.errors.len() <= 16
                     && report.limits.len() <= 8
             }

judge/crates/sandbox/src/runtime.rs

@@ -16,6 +16,40 @@ pub const MAX_CONFIG_BYTES: u64 = 65_536;
 pub enum ImageKind {
     #[serde(rename = "linux_gnu_cxx_x86_64_v1")]
     LinuxGnuCxxX86_64V1,
+    #[serde(rename = "linux_cpython313_x86_64_v1")]
+    LinuxCPython313X86_64V1,
+    #[serde(rename = "linux_openjdk21_x86_64_v1")]
+    LinuxOpenJdk21X86_64V1,
+}
+
+impl ImageKind {
+    pub fn supports(self, profile: crate::spec::ExecutionProfile) -> bool {
+        use crate::spec::ExecutionProfile as P;
+        matches!(
+            (self, profile),
+            (
+                Self::LinuxGnuCxxX86_64V1,
+                P::Native | P::CompileC | P::CompileCpp
+            ) | (Self::LinuxCPython313X86_64V1, P::PythonCheck | P::PythonRun)
+                | (Self::LinuxOpenJdk21X86_64V1, P::JavaCompile | P::JavaRun)
+        )
+    }
+    pub fn contract_kind(self) -> kf_contracts::judging_v2::ImageKind {
+        use kf_contracts::judging_v2::ImageKind as K;
+        match self {
+            Self::LinuxGnuCxxX86_64V1 => K::GnuCxx,
+            Self::LinuxCPython313X86_64V1 => K::CPython313,
+            Self::LinuxOpenJdk21X86_64V1 => K::OpenJdk21,
+        }
+    }
+    pub fn language(self) -> kf_contracts::judging_v2::Language {
+        use kf_contracts::judging_v2::Language as L;
+        match self {
+            Self::LinuxGnuCxxX86_64V1 => L::C17,
+            Self::LinuxCPython313X86_64V1 => L::Python313,
+            Self::LinuxOpenJdk21X86_64V1 => L::Java21,
+        }
+    }
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -51,6 +85,9 @@ pub struct RuntimeConfig {
     pub tasks_root: String,
     pub image_root: String,
     pub image_kind: ImageKind,
+    /// Explicit catalog fingerprint; absent only for the original GNU image.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub runtime_identity: Option<kf_contracts::judging_v2::RuntimeIdentity>,
     pub delegated_cgroup_root: String,
     pub identities: IdentityPool,
     pub max_concurrent_runs: u32,
@@ -88,6 +125,7 @@ impl ValidatedRuntimeConfig {
     pub fn spec_policy(&self) -> SpecPolicy {
         SpecPolicy::with_limits_ceiling(self.0.per_task_ceiling.clone())
             .expect("validated immutable limits ceiling")
+            .for_image(self.0.image_kind)
     }
 
     /// Slot mapping only; the supervisor must lease slots until the old task
@@ -129,6 +167,18 @@ fn fits(each: u64, slots: u32, reserve: u64, total: u64) -> bool {
 
 impl RuntimeConfig {
     pub fn validate(self) -> Result<ValidatedRuntimeConfig, ConfigError> {
+        match &self.runtime_identity {
+            Some(identity) if identity.image_kind == self.image_kind.contract_kind() => identity
+                .validate_for_language(self.image_kind.language())
+                .map_err(|_| invalid("runtime_identity", "invalid_fixed_identity"))?,
+            None if self.image_kind == ImageKind::LinuxGnuCxxX86_64V1 => {}
+            _ => {
+                return Err(invalid(
+                    "runtime_identity",
+                    "missing_or_mismatched_identity",
+                ));
+            }
+        }
         for (field, value) in [
             ("state_root", &self.state_root),
             ("tasks_root", &self.tasks_root),
@@ -343,12 +393,21 @@ pub fn preflight(
     }
     // A controlled directory image includes a GNU C/C++ toolchain, runtime
     // loader and immutable manifest. Content/ownership checks come later.
-    for (subject, relative) in [
+    let binaries: &[(&str, &str)] = match policy.image_kind {
+        ImageKind::LinuxGnuCxxX86_64V1 => &[("gcc", "usr/bin/gcc"), ("g++", "usr/bin/g++")],
+        ImageKind::LinuxCPython313X86_64V1 => &[
+            ("python", "usr/bin/python3.13"),
+            ("python_checker", "usr/lib/kf/python_check.py"),
+        ],
+        ImageKind::LinuxOpenJdk21X86_64V1 => &[
+            ("java", "usr/lib/jvm/bin/java"),
+            ("javac", "usr/lib/jvm/bin/javac"),
+        ],
+    };
+    for &(subject, relative) in binaries.iter().chain(&[
         ("image_manifest", "manifest.json"),
-        ("gcc", "usr/bin/gcc"),
-        ("g++", "usr/bin/g++"),
         ("dynamic_loader", "lib64/ld-linux-x86-64.so.2"),
-    ] {
+    ]) {
         if !Path::new(&policy.image_root).join(relative).is_file() {
             refuse("image_file_not_visible", subject);
         }

judge/crates/sandbox/src/spec.rs

@@ -18,16 +18,42 @@ pub enum ExecutionProfile {
     Native,
     CompileC,
     CompileCpp,
+    PythonCheck,
+    PythonRun,
+    JavaCompile,
+    JavaRun,
 }
 
 impl ExecutionProfile {
     /// Trusted image-local compiler tools only; never inherit the host PATH.
     pub fn compiler_search_path(self) -> Option<&'static str> {
         match self {
-            Self::Native => None,
+            Self::Native
+            | Self::PythonCheck
+            | Self::PythonRun
+            | Self::JavaCompile
+            | Self::JavaRun => None,
             Self::CompileC | Self::CompileCpp => Some("/usr/bin"),
         }
     }
+
+    pub fn prepares_program(self) -> bool {
+        matches!(
+            self,
+            Self::CompileC | Self::CompileCpp | Self::PythonCheck | Self::JavaCompile
+        )
+    }
+
+    pub fn gnu_compiler(self) -> bool {
+        matches!(self, Self::CompileC | Self::CompileCpp)
+    }
+
+    pub fn managed_runtime(self) -> bool {
+        matches!(
+            self,
+            Self::PythonCheck | Self::PythonRun | Self::JavaCompile | Self::JavaRun
+        )
+    }
 }
 
 /// Every field is explicit: omission cannot silently remove a resource limit.
@@ -66,7 +92,9 @@ impl Default for ResourceLimits {
 impl ResourceLimits {
     pub fn for_profile(profile: ExecutionProfile) -> Self {
         match profile {
-            ExecutionProfile::Native => Self::default(),
+            ExecutionProfile::Native
+            | ExecutionProfile::PythonCheck
+            | ExecutionProfile::PythonRun => Self::default(),
             ExecutionProfile::CompileC | ExecutionProfile::CompileCpp => Self {
                 cpu_time_us: 10_000_000,
                 wall_time_us: 30_000_000,
@@ -78,6 +106,20 @@ impl ResourceLimits {
                 workdir_bytes: 256 * MIB,
                 ..Self::default()
             },
+            ExecutionProfile::JavaCompile => Self {
+                memory_bytes: 512 * MIB,
+                processes: 32,
+                open_files: 128,
+                cpu_time_us: 10_000_000,
+                wall_time_us: 30_000_000,
+                ..Self::default()
+            },
+            ExecutionProfile::JavaRun => Self {
+                memory_bytes: 384 * MIB,
+                processes: 32,
+                open_files: 64,
+                ..Self::default()
+            },
         }
     }
 
@@ -129,12 +171,14 @@ pub struct ExecutionSpec {
 #[derive(Clone, Debug)]
 pub struct SpecPolicy {
     limits_ceiling: ResourceLimits,
+    image_kind: Option<crate::runtime::ImageKind>,
 }
 
 impl Default for SpecPolicy {
     fn default() -> Self {
         Self {
             limits_ceiling: ResourceLimits::ceilings(),
+            image_kind: None,
         }
     }
 }
@@ -144,8 +188,14 @@ impl SpecPolicy {
         validate_limits(&limits, &ResourceLimits::ceilings())?;
         Ok(Self {
             limits_ceiling: limits,
+            image_kind: None,
         })
     }
+
+    pub(crate) fn for_image(mut self, kind: crate::runtime::ImageKind) -> Self {
+        self.image_kind = Some(kind);
+        self
+    }
 }
 
 /// An immutable validated request, not proof that the environment is safe.
@@ -162,7 +212,15 @@ impl ValidatedSpec {
     /// image-local PATH/TMPDIR defaults through their environment whitelist.
     pub fn execution_environment(&self) -> BTreeMap<String, String> {
         let mut env = BTreeMap::from([
-            ("LANG".into(), "C".into()),
+            (
+                "LANG".into(),
+                if self.0.profile.managed_runtime() {
+                    "C.UTF-8"
+                } else {
+                    "C"
+                }
+                .into(),
+            ),
             ("TZ".into(), "UTC".into()),
             ("TMPDIR".into(), "/tmp".into()),
         ]);
@@ -238,6 +296,14 @@ pub(crate) fn path(value: &str, field: &'static str) -> Result<(), SpecError> {
     Ok(())
 }
 
+pub(crate) fn artifact_path(value: &str, field: &'static str) -> Result<(), SpecError> {
+    if kf_contracts::judging_v2::class_name(value) {
+        Ok(())
+    } else {
+        path(value, field)
+    }
+}
+
 fn overlaps(left: &str, right: &str) -> bool {
     left == right
         || left
@@ -308,16 +374,36 @@ impl ExecutionSpec {
                 return Err(error("env", ValidationCode::InvalidEnvironment));
             }
         }
+        if self.profile.managed_runtime() && (!self.env.is_empty() || self.cwd != ".") {
+            return Err(error("env_or_cwd", ValidationCode::InvalidEnvironment));
+        }
+        if policy
+            .image_kind
+            .is_some_and(|kind| !kind.supports(self.profile))
+        {
+            return Err(error("profile", ValidationCode::UnsupportedExecutable));
+        }
         if self.cwd != "." {
             path(&self.cwd, "cwd")?;
         }
-        if self.inputs.len() > 64 || self.outputs.len() > 16 {
+        if self.inputs.len()
+            > if self.profile == ExecutionProfile::JavaRun {
+                kf_contracts::judging_v2::MAX_CLASS_FILES
+            } else {
+                64
+            }
+            || self.outputs.len() > 16
+        {
             return Err(error("artifacts", ValidationCode::TooManyArtifacts));
         }
         let mut input_paths = BTreeSet::new();
         for input in &self.inputs {
             identifier(&input.artifact_id, "inputs.artifact_id")?;
+            if self.profile == ExecutionProfile::JavaRun {
+                artifact_path(&input.path, "inputs.path")?;
+            } else {
                 path(&input.path, "inputs.path")?;
+            }
             if input_paths
                 .iter()
                 .any(|other: &&str| overlaps(other, &input.path))
@@ -353,6 +439,14 @@ impl ExecutionSpec {
                 .is_some_and(|relative| input_paths.contains(relative)),
             ExecutionProfile::CompileC => executable == "/usr/bin/gcc",
             ExecutionProfile::CompileCpp => executable == "/usr/bin/g++",
+            ExecutionProfile::PythonCheck
+            | ExecutionProfile::PythonRun
+            | ExecutionProfile::JavaCompile
+            | ExecutionProfile::JavaRun => {
+                crate::profiles::validate(&self)
+                    .map_err(|_| error("profile", ValidationCode::InvalidArgv))?;
+                true
+            }
         };
         if !approved {
             return Err(error("argv[0]", ValidationCode::UnsupportedExecutable));

judge/crates/sandbox/tests/managed_profiles.rs

@@ -0,0 +1,195 @@
+//! Ordinary lexical/authority policy tests. Actual runtime/kernel evidence is VM-only.
+use kf_contracts::judging_v2::{ImageKind as ContractKind, RuntimeIdentity};
+use kf_sandbox::{
+    profiles,
+    runtime::{ImageKind, RuntimeConfig},
+    spec::*,
+};
+
+fn request(profile: ExecutionProfile) -> ExecutionSpec {
+    let limits = ResourceLimits::for_profile(profile);
+    ExecutionSpec {
+        schema_version: kf_contracts::SchemaVersion,
+        run_id: "managed-policy".into(),
+        profile,
+        argv: profiles::argv(profile, &limits).unwrap(),
+        env: Default::default(),
+        cwd: ".".into(),
+        inputs: if profile == ExecutionProfile::JavaRun {
+            [
+                "Helper.class",
+                "Main$1.class",
+                "Main$Inner.class",
+                "Main.class",
+            ]
+            .iter()
+            .enumerate()
+            .map(|(i, path)| InputArtifact {
+                artifact_id: format!("class-{i}"),
+                path: (*path).into(),
+            })
+            .collect()
+        } else {
+            vec![InputArtifact {
+                artifact_id: "source".into(),
+                path: if profile == ExecutionProfile::JavaCompile {
+                    "Main.java"
+                } else {
+                    "main.py"
+                }
+                .into(),
+            }]
+        },
+        stdin_artifact: None,
+        outputs: vec![],
+        limits,
+    }
+}
+
+#[test]
+fn commands_environment_and_source_or_class_sets_are_closed() {
+    for profile in [
+        ExecutionProfile::PythonCheck,
+        ExecutionProfile::PythonRun,
+        ExecutionProfile::JavaCompile,
+        ExecutionProfile::JavaRun,
+    ] {
+        let original = request(profile);
+        let valid = original.clone().validate(&SpecPolicy::default()).unwrap();
+        assert_eq!(valid.execution_environment()["LANG"], "C.UTF-8");
+        assert!(!valid.execution_environment().contains_key("PATH"));
+        for arg in [
+            "-c",
+            "-m",
+            "-javaagent:/work/user.jar",
+            "-cp",
+            "@options",
+            "/bin/sh",
+        ] {
+            let mut modified = original.clone();
+            modified.argv.push(arg.into());
+            assert!(modified.validate(&SpecPolicy::default()).is_err());
+        }
+        for variable in [
+            "PATH",
+            "PYTHONPATH",
+            "PYTHONHOME",
+            "JAVA_TOOL_OPTIONS",
+            "JDK_JAVA_OPTIONS",
+            "CLASSPATH",
+            "LD_PRELOAD",
+            "LANG",
+        ] {
+            let mut modified = original.clone();
+            modified.env.insert(variable.into(), "C".into());
+            assert!(modified.validate(&SpecPolicy::default()).is_err());
+        }
+        for input in [
+            "../Main.class",
+            "pkg/Main.class",
+            "user.jar",
+            "Main$1.class/child",
+            "main.pyc",
+            "options",
+        ] {
+            let mut modified = original.clone();
+            modified.inputs[0].path = input.into();
+            assert!(modified.validate(&SpecPolicy::default()).is_err());
+        }
+        let mut changed = original.clone();
+        changed.cwd = "tmp".into();
+        assert!(changed.validate(&SpecPolicy::default()).is_err());
+        let mut changed = original;
+        changed.outputs.push("user.jar".into());
+        assert!(changed.validate(&SpecPolicy::default()).is_err());
+    }
+    let mut missing = request(ExecutionProfile::JavaRun);
+    missing.inputs.pop();
+    assert!(missing.validate(&SpecPolicy::default()).is_err());
+    let mut duplicate = request(ExecutionProfile::JavaRun);
+    duplicate.inputs.push(duplicate.inputs[0].clone());
+    assert!(duplicate.validate(&SpecPolicy::default()).is_err());
+}
+
+#[test]
+fn startup_heap_and_budget_parameters_cannot_be_supplied_by_request() {
+    for profile in [ExecutionProfile::JavaCompile, ExecutionProfile::JavaRun] {
+        let original = request(profile);
+        assert!(
+            original
+                .argv
+                .iter()
+                .any(|a| a.contains("ActiveProcessorCount=1"))
+        );
+        assert!(
+            original
+                .argv
+                .iter()
+                .any(|a| a.contains("DisableAttachMechanism"))
+        );
+        assert!(original.argv.iter().any(|a| a.contains("UseSerialGC")));
+        let mut changed = original.clone();
+        changed.limits.memory_bytes += 1024 * 1024;
+        assert!(changed.validate(&SpecPolicy::default()).is_err());
+        let mut tiny = original.limits;
+        tiny.memory_bytes = 128 * 1024 * 1024;
+        assert!(profiles::argv(profile, &tiny).is_err());
+    }
+    let mut small = request(ExecutionProfile::PythonRun);
+    small.limits.open_files = 8;
+    assert!(small.validate(&SpecPolicy::default()).is_err());
+}
+
+#[test]
+fn root_image_kind_never_promotes_legacy_authority() {
+    let legacy: RuntimeConfig = serde_json::from_str(include_str!(
+        "../../../../deploy/sandbox.runtime.example.json"
+    ))
+    .unwrap();
+    let old = legacy.clone().validate().unwrap();
+    assert!(
+        request(ExecutionProfile::PythonRun)
+            .validate(&old.spec_policy())
+            .is_err()
+    );
+    for (kind, contract, profile, id, version) in [
+        (
+            ImageKind::LinuxCPython313X86_64V1,
+            ContractKind::CPython313,
+            ExecutionProfile::PythonRun,
+            "cpython313-v1",
+            "3.13.16",
+        ),
+        (
+            ImageKind::LinuxOpenJdk21X86_64V1,
+            ContractKind::OpenJdk21,
+            ExecutionProfile::JavaRun,
+            "openjdk21-v1",
+            "21.0.12.1+1-LTS",
+        ),
+    ] {
+        let mut config = legacy.clone();
+        config.image_kind = kind;
+        assert!(config.clone().validate().is_err());
+        config.runtime_identity = Some(RuntimeIdentity {
+            toolchain_id: id.to_owned().try_into().unwrap(),
+            image_kind: contract,
+            image_revision: 1,
+            image_digest: "a".repeat(64).try_into().unwrap(),
+            runtime_version: version.to_owned().try_into().unwrap(),
+        });
+        let validated = config.clone().validate().unwrap();
+        assert!(request(profile).validate(&validated.spec_policy()).is_ok());
+        assert!(
+            request(if profile == ExecutionProfile::PythonRun {
+                ExecutionProfile::JavaRun
+            } else {
+                ExecutionProfile::PythonRun
+            })
+            .validate(&validated.spec_policy())
+            .is_err()
+        );
+        config.runtime_identity.as_mut().unwrap().image_revision = 0;
+        assert!(config.validate().is_err());
+    }
+}

judge/crates/sandbox/tests/managed_runtime_vm.rs

@@ -0,0 +1,415 @@
+#![cfg(all(target_os = "linux", feature = "vm-acceptance"))]
+//! Actual new profile execution through the ordinary default Root entry.
+//! This does not substitute for the later ordinary Worker/browser pipeline.
+#[allow(dead_code)]
+mod support;
+use kf_contracts::judging_v2::RuntimeIdentity;
+use kf_sandbox::{
+    image::{ImageCatalog, catalog_digest},
+    profiles,
+    runtime::ImageKind,
+    spec::*,
+};
+use std::{
+    fs,
+    os::unix::fs::{MetadataExt, PermissionsExt},
+    path::PathBuf,
+    process::{Command, Stdio},
+};
+
+struct Fixture {
+    root: support::Root,
+    dir: PathBuf,
+    config: PathBuf,
+}
+impl Fixture {
+    fn new(language: &str, kind: ImageKind) -> Self {
+        support::require_vm();
+        let mut root = support::Root::new();
+        root.config.max_concurrent_runs = 1;
+        root.config.identities.count = 1;
+        root.config.aggregate.task_cpu_quota_us = 100_000;
+        root.config.per_task_ceiling.cpu_time_us = 30_000_000;
+        root.config.per_task_ceiling.wall_time_us = 60_000_000;
+        root.config.image_root = format!("/var/lib/kf-judge/images/{language}");
+        root.config.image_kind = kind;
+        let bytes = fs::read(format!("{}/manifest.json", root.config.image_root)).unwrap();
+        let catalog: ImageCatalog = serde_json::from_slice(&bytes).unwrap();
+        root.config.runtime_identity = Some(RuntimeIdentity {
+            toolchain_id: catalog.toolchain_id.clone(),
+            image_kind: catalog.image_kind,
+            image_revision: catalog.image_revision,
+            image_digest: catalog_digest(&bytes),
+            runtime_version: catalog.runtime_version.clone(),
+        });
+        root.config.clone().validate().unwrap();
+        catalog
+            .validate(root.config.runtime_identity.as_ref().unwrap())
+            .unwrap();
+        let image = PathBuf::from(&root.config.image_root);
+        for (path, mode, bytes) in std::iter::once((String::new(), 0o555, None))
+            .chain(catalog.directories.iter().map(|p| (p.clone(), 0o555, None)))
+            .chain(catalog.files.iter().map(|p| {
+                (
+                    p.path.clone(),
+                    if p.executable { 0o555 } else { 0o444 },
+                    Some(p.bytes),
+                )
+            }))
+        {
+            let metadata = fs::symlink_metadata(image.join(&path)).unwrap();
+            assert_eq!(
+                (metadata.uid(), metadata.gid(), metadata.mode() & 0o7777),
+                (0, 0, mode),
+                "image metadata {path}"
+            );
+            if let Some(bytes) = bytes {
+                assert!(metadata.is_file(), "{path}");
+                assert_eq!(metadata.nlink(), 1, "{path}");
+                assert_eq!(metadata.len(), bytes, "{path}");
+            }
+        }
+        println!("KF_MANAGED_CATALOG_METADATA_STATUS=0 {language}");
+        let dir = PathBuf::from(format!("/kf/managed-{}-{language}", std::process::id()));
+        fs::create_dir(&dir).unwrap();
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o700)).unwrap();
+        for name in ["inputs", "outputs"] {
+            fs::create_dir(dir.join(name)).unwrap();
+        }
+        let config = dir.join("runtime.json");
+        let fixture = Self { root, dir, config };
+        fixture.save_config();
+        println!(
+            "KF_MANAGED_IMAGE_PROOF {}",
+            serde_json::json!({"language":language,"identity":fixture.root.config.runtime_identity})
+        );
+        fixture
+    }
+    fn save_config(&self) {
+        fs::write(&self.config, serde_json::to_vec(&self.root.config).unwrap()).unwrap();
+        fs::set_permissions(&self.config, fs::Permissions::from_mode(0o600)).unwrap();
+    }
+    fn input(&self, name: &str, bytes: &[u8]) {
+        fs::write(self.dir.join("inputs").join(name), bytes).unwrap();
+        fs::set_permissions(
+            self.dir.join("inputs").join(name),
+            fs::Permissions::from_mode(0o600),
+        )
+        .unwrap();
+    }
+    fn request(&self, id: &str, profile: ExecutionProfile) -> ExecutionSpec {
+        let mut limits = ResourceLimits::for_profile(profile);
+        // Same finite acceptance ceiling as the existing real GNU regression;
+        // this fixture does not change any submitted job's configured limits.
+        limits.cpu_time_us = 30_000_000;
+        limits.wall_time_us = 60_000_000;
+        ExecutionSpec {
+            schema_version: kf_contracts::SchemaVersion,
+            run_id: id.into(),
+            profile,
+            argv: profiles::argv(profile, &limits).unwrap(),
+            env: Default::default(),
+            cwd: ".".into(),
+            inputs: vec![InputArtifact {
+                artifact_id: "source".into(),
+                path: if profile == ExecutionProfile::JavaCompile {
+                    "Main.java"
+                } else {
+                    "main.py"
+                }
+                .into(),
+            }],
+            stdin_artifact: None,
+            outputs: vec![],
+            limits,
+        }
+    }
+    fn command(&self, spec: &ExecutionSpec) -> Command {
+        fs::write(
+            self.dir.join("spec.json"),
+            serde_json::to_vec(spec).unwrap(),
+        )
+        .unwrap();
+        let mut command = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap());
+        command
+            .env_clear()
+            .args(["run", "--config"])
+            .arg(&self.config)
+            .arg("--spec")
+            .arg(self.dir.join("spec.json"))
+            .arg("--artifacts")
+            .arg(self.dir.join("inputs"))
+            .arg("--output")
+            .arg(self.dir.join("outputs"));
+        command
+    }
+    fn execute(&self, spec: &ExecutionSpec) -> serde_json::Value {
+        let output = if std::env::var("KF_MANAGED_TRACE").as_deref() == Ok("1") {
+            let mut child = self
+                .command(spec)
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped())
+                .spawn()
+                .unwrap();
+            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
+            let mut tracer = None;
+            while child.try_wait().unwrap().is_none() && std::time::Instant::now() < deadline {
+                let mut parents = vec![child.id()];
+                for _ in 0..4 {
+                    let mut children = vec![];
+                    for pid in parents {
+                        if let Ok(data) =
+                            fs::read_to_string(format!("/proc/{pid}/task/{pid}/children"))
+                        {
+                            children.extend(
+                                data.split_whitespace()
+                                    .filter_map(|v| v.parse::<u32>().ok()),
+                            );
+                        }
+                    }
+                    if let Some(pid) = children.iter().find(|pid| {
+                        fs::read_to_string(format!("/proc/{pid}/comm"))
+                            .is_ok_and(|s| s.trim() == "python3.13")
+                    }) {
+                        tracer = Some(
+                            Command::new("/kf/bin/strace")
+                                .env_clear()
+                                .args([
+                                    "-ff",
+                                    "-qq",
+                                    "-e",
+                                    "trace=open,openat,openat2,fcntl,ioctl",
+                                    "-e",
+                                    "status=failed",
+                                    "-o",
+                                ])
+                                .arg(self.dir.join("trace"))
+                                .arg("-p")
+                                .arg(pid.to_string())
+                                .spawn()
+                                .unwrap(),
+                        );
+                        break;
+                    }
+                    parents = children;
+                }
+                if tracer.is_some() {
+                    break;
+                }
+                std::thread::sleep(std::time::Duration::from_millis(2));
+            }
+            let output = child.wait_with_output().unwrap();
+            if let Some(mut tracer) = tracer {
+                let _ = tracer.wait();
+            }
+            output
+        } else {
+            self.command(spec).output().unwrap()
+        };
+        if std::env::var("KF_MANAGED_TRACE").as_deref() == Ok("1") {
+            for entry in fs::read_dir(&self.dir).unwrap() {
+                let entry = entry.unwrap();
+                if entry.file_name().to_string_lossy().starts_with("trace.") {
+                    let data = fs::read_to_string(entry.path()).unwrap();
+                    for line in data.lines().filter(|s| s.contains("EPERM")) {
+                        println!(
+                            "KF_MANAGED_DENIED_SYSCALL {}",
+                            line.chars().take(1024).collect::<String>()
+                        );
+                    }
+                }
+            }
+        }
+        assert!(
+            output.status.success(),
+            "{} {}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+        let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap();
+        println!(
+            "KF_MANAGED_RUN_PROOF {}",
+            serde_json::json!({"spec":spec,"result":value})
+        );
+        for name in ["stdout", "stderr"] {
+            if let Ok(bytes) = fs::read(self.dir.join("outputs").join(&spec.run_id).join(name)) {
+                println!(
+                    "KF_MANAGED_CAPTURE {}",
+                    serde_json::json!({"run_id":spec.run_id,"name":name,"prefix":String::from_utf8_lossy(&bytes[..bytes.len().min(4096)])})
+                );
+            }
+        }
+        assert_eq!(value["report"]["cleanup"], "complete");
+        assert_eq!(value["report"]["errors"], serde_json::json!([]));
+        self.empty();
+        value
+    }
+    fn reject(&self, spec: &ExecutionSpec, code: &str) {
+        let output = self.command(spec).output().unwrap();
+        assert_eq!(
+            output.status.code(),
+            Some(2),
+            "{}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+        assert!(output.stdout.is_empty());
+        let value: serde_json::Value = serde_json::from_slice(&output.stderr).unwrap();
+        assert_eq!(value["error"], code);
+        println!("KF_MANAGED_REFUSAL_PROOF {value}");
+        self.empty();
+    }
+    fn stdout(&self, id: &str) -> Vec<u8> {
+        fs::read(self.dir.join("outputs").join(id).join("stdout")).unwrap()
+    }
+    fn image_refusals(&mut self, spec: &ExecutionSpec, binary: &str) {
+        let image = PathBuf::from(&self.root.config.image_root);
+        let executable = image.join(binary);
+        let backup = self.dir.join("trusted-runtime-backup");
+        fs::rename(&executable, &backup).unwrap();
+        self.reject(spec, "runtime_not_ready");
+        fs::rename(&backup, &executable).unwrap();
+        fs::set_permissions(&executable, fs::Permissions::from_mode(0o777)).unwrap();
+        self.reject(spec, "runtime_not_ready");
+        fs::set_permissions(&executable, fs::Permissions::from_mode(0o555)).unwrap();
+        let original = self.root.config.runtime_identity.clone().unwrap();
+        self.root
+            .config
+            .runtime_identity
+            .as_mut()
+            .unwrap()
+            .runtime_version = "1.0.0".to_owned().try_into().unwrap();
+        self.save_config();
+        self.reject(spec, "untrusted_runtime_config");
+        self.root.config.runtime_identity = Some(original.clone());
+        self.root
+            .config
+            .runtime_identity
+            .as_mut()
+            .unwrap()
+            .image_digest = "e".repeat(64).try_into().unwrap();
+        self.save_config();
+        self.reject(spec, "runtime_not_ready");
+        self.root.config.runtime_identity = Some(original);
+        self.save_config();
+        let loader = image.join("lib64/ld-linux-x86-64.so.2");
+        fs::rename(&loader, &backup).unwrap();
+        self.reject(spec, "runtime_not_ready");
+        fs::rename(&backup, &loader).unwrap();
+        let original = fs::read(&loader).unwrap();
+        let mut corrupted = original.clone();
+        let last = corrupted.len() - 1;
+        corrupted[last] ^= 1;
+        fs::write(&loader, corrupted).unwrap();
+        self.reject(spec, "runtime_not_ready");
+        fs::write(&loader, original).unwrap();
+        println!(
+            "KF_MANAGED_IMAGE_REFUSALS_STATUS=0 {}",
+            self.root.config.image_root
+        );
+    }
+    fn empty(&self) {
+        for name in ["tasks", "captures", "identities"] {
+            let p = PathBuf::from("/var/lib/kf-judge").join(name);
+            if p.exists() {
+                assert_eq!(fs::read_dir(p).unwrap().count(), 0);
+            }
+        }
+        assert!(
+            !fs::read_dir(&self.root.path).unwrap().any(|e| e
+                .unwrap()
+                .file_type()
+                .unwrap()
+                .is_dir())
+        );
+    }
+}
+impl Drop for Fixture {
+    fn drop(&mut self) {
+        let _ = fs::remove_dir_all(&self.dir);
+    }
+}
+
+#[test]
+#[ignore = "independent no-NIC/no-share VM with read-back total and fixture caps"]
+fn python_profile_check_run_and_identity_refusals() {
+    let mut f = Fixture::new("python313", ImageKind::LinuxCPython313X86_64V1);
+    f.input("source", b"import sys\nprint(sys.version.split()[0])\nprint('top-level')\nopen('marker','w').write('owned')\n");
+    let check = f.request("py-check", ExecutionProfile::PythonCheck);
+    let r = f.execute(&check);
+    assert_eq!(r["outcome"], serde_json::json!({"kind":"exited","code":0}));
+    assert!(f.stdout("py-check").is_empty());
+    assert!(r["report"]["artifacts"].as_array().unwrap().is_empty());
+    let run = f.request("py-run", ExecutionProfile::PythonRun);
+    let r = f.execute(&run);
+    assert_eq!(r["outcome"], serde_json::json!({"kind":"exited","code":0}));
+    assert_eq!(f.stdout("py-run"), b"3.13.16\ntop-level\n");
+    f.input("source", b"# coding: latin-1\nprint('caf\xe9')\n");
+    assert_eq!(
+        f.execute(&f.request("py-encoding", ExecutionProfile::PythonCheck))["outcome"]["code"],
+        0
+    );
+    f.input("source", b"if !invalid:\n    pass\n");
+    assert_eq!(
+        f.execute(&f.request("py-syntax", ExecutionProfile::PythonCheck))["outcome"]["code"],
+        1
+    );
+    let mut wrong = run.clone();
+    wrong.argv.push("-c".into());
+    wrong.run_id = "py-argv-refused".into();
+    f.reject(&wrong, "invalid_execution_spec");
+    f.image_refusals(&run, "usr/bin/python3.13");
+}
+
+#[test]
+#[ignore = "independent no-NIC/no-share VM with read-back total and fixture caps"]
+fn java_profile_compiles_class_set_and_runs_same_set() {
+    let mut f = Fixture::new("java21", ImageKind::LinuxOpenJdk21X86_64V1);
+    f.input("source", br#"public class Main { static class Inner { } public static void main(String[] args) { System.out.println(System.getProperty("java.runtime.version")); Runnable r = new Runnable() { public void run() { System.out.println("java-ok"); } }; r.run(); new Inner(); } } class Helper {}"#);
+    let compile = f.request("java-compile", ExecutionProfile::JavaCompile);
+    let r = f.execute(&compile);
+    assert_eq!(r["outcome"], serde_json::json!({"kind":"exited","code":0}));
+    let artifacts = r["report"]["artifacts"].as_array().unwrap();
+    let names: Vec<_> = artifacts
+        .iter()
+        .map(|a| a["path"].as_str().unwrap())
+        .collect();
+    assert_eq!(
+        names,
+        [
+            "Helper.class",
+            "Main$1.class",
+            "Main$Inner.class",
+            "Main.class"
+        ]
+    );
+    let mut run = f.request("java-run", ExecutionProfile::JavaRun);
+    run.inputs.clear();
+    for (i, name) in names.iter().enumerate() {
+        let id = format!("class-{i}");
+        let bytes = fs::read(
+            f.dir
+                .join("outputs/java-compile")
+                .join(format!("artifact-{i}")),
+        )
+        .unwrap();
+        f.input(&id, &bytes);
+        run.inputs.push(InputArtifact {
+            artifact_id: id,
+            path: (*name).into(),
+        });
+    }
+    let r = f.execute(&run);
+    assert_eq!(r["outcome"], serde_json::json!({"kind":"exited","code":0}));
+    assert_eq!(f.stdout("java-run"), b"21.0.12.1+1-LTS\njava-ok\n");
+    f.input("source", b"public class Main { invalid }\n");
+    let r = f.execute(&f.request("java-syntax", ExecutionProfile::JavaCompile));
+    assert_eq!(r["outcome"]["code"], 1);
+    assert!(r["report"]["artifacts"].as_array().unwrap().is_empty());
+    let mut wrong = run;
+    wrong
+        .env
+        .insert("JAVA_TOOL_OPTIONS".into(), "-javaagent:/work/x.jar".into());
+    wrong.run_id = "java-env-refused".into();
+    f.reject(&wrong, "invalid_execution_spec");
+    f.image_refusals(&wrong, "usr/lib/jvm/bin/java");
+}

scripts/build_judging_images.py

@@ -0,0 +1,323 @@
+"""Offline curated directory images from pinned public archives and verified libraries.
+
+Build as an ordinary user. Publish as root with 0555 directories / 0444 data /
+0555 executables in a private VM; the resulting catalog fingerprint belongs in
+the protected runtime config. This tool never executes the target runtimes,
+submitted source, package hooks, or network installers.
+"""
+
+import argparse
+import gzip
+import hashlib
+import json
+import re
+import subprocess
+import tarfile
+import tempfile
+from pathlib import Path, PurePosixPath
+
+PINS = {
+    "python313": {
+        "sha256": "4595c5589fff7bf0cb158d9a88a797e0d791fa33830770fcb7bf3f4b104feeae",
+        "version": "3.13.16",
+        "prefix": "python/",
+        "toolchain_id": "cpython313-v1",
+        "kind": "linux_cpython313_x86_64_v1",
+        "source": "https://github.com/astral-sh/python-build-standalone/releases/download/20261003/cpython-3.13.16%2B20261003-x86_64-unknown-linux-gnu-install_only_stripped.tar.gz",
+    },
+    "java21": {
+        "sha256": "ce79869e1307ed8ee1e2baa86a412b1eb5b75d10a01006d788a6f968bcfaee94",
+        "version": "21.0.12.1+1-LTS",
+        "prefix": "jdk-21.0.12.1+1/",
+        "toolchain_id": "openjdk21-v1",
+        "kind": "linux_openjdk21_x86_64_v1",
+        "source": "https://github.com/adoptium/temurin21-binaries/releases/download/jdk-21.0.12.1%2B1/OpenJDK21U-jdk_x64_linux_hotspot_21.0.12.1_1.tar.gz",
+    },
+}
+
+
+def sha(path):
+    with path.open("rb") as source:
+        return hashlib.file_digest(source, "sha256").hexdigest()
+
+
+def canonical(name):
+    return (
+        len(name) <= 4096
+        and not name.startswith("/")
+        and all(
+            re.fullmatch(r"[A-Za-z0-9_.-]{1,255}", p) and p not in (".", "..")
+            for p in name.split("/")
+        )
+    )
+
+
+def member_bytes(archive, members, name, seen=()):
+    if name in seen or len(seen) > 8 or name not in members:
+        raise ValueError("public archive link outside bounded member set")
+    item = members[name]
+    if item.issym() or item.islnk():
+        parts = []
+        target = PurePosixPath(item.linkname)
+        if target.is_absolute():
+            raise ValueError("absolute public archive link")
+        target = target if item.islnk() else PurePosixPath(name).parent / target
+        for part in target.parts:
+            if part == "..":
+                if not parts:
+                    raise ValueError("public archive link escapes root")
+                parts.pop()
+            elif part != ".":
+                parts.append(part)
+        return member_bytes(archive, members, "/".join(parts), (*seen, name))
+    if not item.isfile() or item.size > 256 * 1024 * 1024:
+        raise ValueError("unsupported public archive member")
+    with archive.extractfile(item) as source:
+        data = source.read(item.size + 1)
+    if len(data) != item.size:
+        raise ValueError("public archive length")
+    return data, bool(item.mode & 0o111)
+
+
+def put(root, name, data, executable=False):
+    if not canonical(name):
+        raise ValueError("noncanonical image path")
+    target = root / name
+    if target.exists():
+        raise ValueError("duplicate image path")
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_bytes(data)
+    target.chmod(0o555 if executable else 0o444)
+
+
+def selected(language, relative):
+    if language == "python313":
+        if relative == "bin/python3.13":
+            return "usr/bin/python3.13"
+        if relative.startswith("lib/"):
+            if not relative.startswith("lib/python3.13/") and relative not in {
+                "lib/libpython3.so",
+                "lib/libpython3.13.so",
+                "lib/libpython3.13.so.1.0",
+            }:
+                return None
+            parts = PurePosixPath(relative).parts
+            if any(
+                p
+                in {
+                    "site-packages",
+                    "__pycache__",
+                    "ensurepip",
+                    "test",
+                    "tests",
+                    "idlelib",
+                    "tkinter",
+                }
+                for p in parts
+            ):
+                return None
+            if relative.endswith((".pyc", ".pth", ".a")) or "_tkinter" in relative:
+                return None
+            if relative.endswith(("/sitecustomize.py", "/usercustomize.py")):
+                return None
+            return "usr/" + relative
+        return None
+    if relative in {"bin/java", "bin/javac", "release", "NOTICE"}:
+        return "usr/lib/jvm/" + relative
+    if relative.startswith(("lib/", "conf/", "legal/")):
+        if relative.endswith("src.zip") or PurePosixPath(relative).name in {
+            "libawt_xawt.so",
+            "libjawt.so",
+            "libsplashscreen.so",
+            "libjsound.so",
+        }:
+            return None
+        return "usr/lib/jvm/" + relative
+    return None
+
+
+def support_file(args, name, records):
+    root = args.support_dir.resolve(strict=True)
+    source = (root / "usr/lib" / name).resolve(strict=True)
+    relative = source.relative_to(root).as_posix()
+    expected = records.get(relative)
+    if expected is None or expected.get("kind") != "file" or sha(source) != expected["sha256"]:
+        raise ValueError("support library differs from verified public manifest")
+    return source.read_bytes()
+
+
+def dependencies(path):
+    with path.open("rb") as source:
+        if source.read(4) != b"\x7fELF":
+            return []
+    result = subprocess.run(
+        ["readelf", "-d", str(path)],
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=15,
+    )
+    return re.findall(r"\(NEEDED\).*\[([^\]]+)\]", result.stdout)
+
+
+def build(args):
+    pin = PINS[args.language]
+    if sha(args.archive) != pin["sha256"]:
+        raise ValueError("runtime archive does not match fixed upstream digest")
+    if sha(args.support_manifest) != args.support_manifest_sha256:
+        raise ValueError("support manifest fingerprint")
+    support = json.loads(args.support_manifest.read_text())
+    records = {v["path"]: v for v in support["files"]}
+    root = args.output
+    root.mkdir(mode=0o700, parents=False, exist_ok=False)
+    # Inflate once into a bounded anonymous file for random member/link reads.
+    # No archive member path is extracted during this conversion.
+    with tempfile.TemporaryFile() as unpacked:
+        with gzip.open(args.archive, "rb") as compressed:
+            total = 0
+            while data := compressed.read(1024 * 1024):
+                total += len(data)
+                if total > 512 * 1024 * 1024:
+                    raise ValueError("public archive inflated byte ceiling")
+                unpacked.write(data)
+        unpacked.seek(0)
+        with tarfile.open(fileobj=unpacked, mode="r:") as archive:
+            copy_archive(args, pin, root, archive)
+    if args.language == "python313":
+        checker = (
+            Path(__file__).resolve().parents[1] / "judge/crates/sandbox/runtime/python_check.py"
+        )
+        put(root, "usr/lib/kf/python_check.py", checker.read_bytes())
+    finish(args, pin, root, records)
+
+
+def copy_archive(args, pin, root, archive):
+    members = {}
+    for item in archive.getmembers():
+        name = item.name.rstrip("/")
+        if (
+            name in members
+            or PurePosixPath(name).is_absolute()
+            or ".." in PurePosixPath(name).parts
+        ):
+            raise ValueError("duplicate/noncanonical public archive member")
+        members[name] = item
+    for name in sorted(members):
+        relative = name.removeprefix(pin["prefix"])
+        if relative == name or members[name].isdir():
+            continue
+        destination = selected(args.language, relative)
+        if destination is not None:
+            data, executable = member_bytes(archive, members, name)
+            put(root, destination, data, executable)
+
+
+def finish(args, pin, root, records):
+    # Flatten immutable image-local loader/libs; no symlink or host source is
+    # retained. Resolve ELF requirements from the image plus verified support.
+    put(
+        root,
+        "lib64/ld-linux-x86-64.so.2",
+        support_file(args, "ld-linux-x86-64.so.2", records),
+        True,
+    )
+    known = {p.name for p in root.rglob("*") if p.is_file()}
+    pending = [p for p in root.rglob("*") if p.is_file()]
+    copied = []
+    while pending:
+        source = pending.pop()
+        for name in dependencies(source):
+            if name.startswith("$ORIGIN/"):
+                target = (source.parent / name.removeprefix("$ORIGIN/")).resolve(strict=True)
+                relative = target.relative_to(root.resolve()).as_posix()
+                if not canonical(relative) or not target.is_file() or target.is_symlink():
+                    raise ValueError("image-relative ELF dependency is not a listed regular file")
+                continue
+            if not canonical(name) or "/" in name:
+                raise ValueError("noncanonical ELF dependency")
+            if name in known:
+                continue
+            data = support_file(args, name, records)
+            put(root, "usr/lib/" + name, data, True)
+            known.add(name)
+            copied.append({"path": "usr/lib/" + name, "sha256": hashlib.sha256(data).hexdigest()})
+            pending.append(root / "usr/lib" / name)
+    for anchor in ["work", "tmp", "proc", "dev"]:
+        (root / anchor).mkdir()
+    files = []
+    directories = []
+    for path in sorted(root.rglob("*")):
+        relative = path.relative_to(root).as_posix()
+        if path.is_symlink() or not canonical(relative):
+            raise ValueError("unexpected image inode/path")
+        if path.is_file():
+            files.append(
+                {
+                    "path": relative,
+                    "bytes": path.stat().st_size,
+                    "sha256": sha(path),
+                    "executable": bool(path.stat().st_mode & 0o111),
+                }
+            )
+        elif path.is_dir():
+            directories.append(relative)
+        else:
+            raise ValueError("special image inode")
+    if len(files) > 4096 or sum(f["bytes"] for f in files) > 1024**3:
+        raise ValueError("image exceeds catalog capacity")
+    catalog = {
+        "format_version": 2,
+        "architecture": "x86_64",
+        "image_kind": pin["kind"],
+        "image_revision": 1,
+        "toolchain_id": pin["toolchain_id"],
+        "runtime_version": pin["version"],
+        "directories": sorted(directories),
+        "files": sorted(files, key=lambda f: f["path"]),
+    }
+    data = (
+        json.dumps(catalog, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n"
+    ).encode()
+    if len(data) > 1024**2:
+        raise ValueError("catalog exceeds bounded transport")
+    put(root, "manifest.json", data)
+    for directory in sorted((p for p in root.rglob("*") if p.is_dir()), reverse=True):
+        directory.chmod(0o555)
+    root.chmod(0o555)
+    identity = {
+        "toolchain_id": pin["toolchain_id"],
+        "image_kind": pin["kind"],
+        "image_revision": 1,
+        "runtime_version": pin["version"],
+        "image_digest": hashlib.sha256(data).hexdigest(),
+    }
+    print(
+        json.dumps(
+            {
+                "runtime_identity": identity,
+                "archive": pin,
+                "support_manifest_sha256": args.support_manifest_sha256,
+                "files": len(files),
+                "bytes": sum(f["bytes"] for f in files),
+                "support_libraries": copied,
+                "source_execution": False,
+                "root_publication_and_VM_acceptance": False,
+            },
+            indent=2,
+        )
+    )
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("language", choices=PINS)
+    parser.add_argument("--archive", required=True, type=Path)
+    parser.add_argument("--support-dir", required=True, type=Path)
+    parser.add_argument("--support-manifest", required=True, type=Path)
+    parser.add_argument("--support-manifest-sha256", required=True)
+    parser.add_argument("--output", required=True, type=Path)
+    build(parser.parse_args())
+
+
+if __name__ == "__main__":
+    main()