diff --git a/judge/crates/sandbox/tests/managed_pipeline_vm.rs b/judge/crates/sandbox/tests/managed_pipeline_vm.rs index db03265..9b8784f 100644 --- a/judge/crates/sandbox/tests/managed_pipeline_vm.rs +++ b/judge/crates/sandbox/tests/managed_pipeline_vm.rs @@ -30,7 +30,12 @@ fn id(n: u32) -> String { } fn budget(profile: ExecutionProfile) -> v1::RunLimits { let mut b = ResourceLimits::for_profile(profile); - if profile.prepares_program() { + if profile == ExecutionProfile::JavaCompile { + // Actual javac of the supported 256-class boundary under single-vCPU + // TCG needs a finite larger fixture budget. Production limits stay fixed. + b.cpu_time_us = 60_000_000; + b.wall_time_us = 120_000_000; + } else if profile.prepares_program() { b.cpu_time_us = 30_000_000; b.wall_time_us = 60_000_000; } else if matches!( @@ -55,8 +60,8 @@ impl Fixture { root.config.max_concurrent_runs = 1; root.config.identities.count = 1; root.config.aggregate.task_cpu_quota_us = 100000; - root.config.per_task_ceiling.cpu_time_us = 30_000_000; - root.config.per_task_ceiling.wall_time_us = 60_000_000; + root.config.per_task_ceiling.cpu_time_us = 60_000_000; + root.config.per_task_ceiling.wall_time_us = 120_000_000; let mut capabilities = Vec::new(); for (name, kind, languages) in [ ( @@ -182,6 +187,16 @@ impl Fixture { points: &[(&[u8], &[u8])], verdict: v1::Verdict, legacy: bool, + ) -> serde_json::Value { + self.execute_with(language, source, points, verdict, legacy, |_| {}) + } + fn make_job( + &mut self, + language: v2::Language, + source: &[u8], + points: &[(&[u8], &[u8])], + legacy: bool, + edit: impl FnOnce(&mut serde_json::Value), ) -> serde_json::Value { self.next += 10; let mut m: serde_json::Value = serde_json::from_str(include_str!( @@ -221,6 +236,7 @@ impl Fixture { }) .collect::>() ); + edit(&mut m); let canonical = if legacy { m["schema_version"] = serde_json::json!(1); m["toolchain"] = serde_json::json!({"toolchain_id":"gnu-cxx-v1","image_kind":"linux_gnu_cxx_x86_64_v1","image_revision":1}); @@ -237,6 +253,9 @@ impl Fixture { &self.dir.join("job.json"), &serde_json::to_vec(&job).unwrap(), ); + job + } + fn worker_command(&self, legacy: bool) -> Command { let mut command = Command::new(std::env::var_os("KF_WORKER_TEST_BIN").unwrap()); command .env_clear() @@ -259,7 +278,55 @@ impl Fixture { .arg("--admission") .arg(self.dir.join("admission.json")); } - let output = command.output().unwrap(); + command + } + fn assert_drained(&self) { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let stopped = Command::new(std::env::current_exe().unwrap()) + .env_clear() + .uid(1000) + .gid(1000) + .args([ + "--ignored", + "--exact", + "ordinary_pipeline_stopped_client", + "--test-threads=1", + "--nocapture", + ]) + .output() + .unwrap(); + if stopped.status.success() + && String::from_utf8_lossy(&stopped.stdout).contains("KF_PIPELINE_STOPPED=1") + { + break; + } + assert!( + Instant::now() < deadline, + "global drain refused: {stopped:?}" + ); + std::thread::sleep(Duration::from_millis(25)); + } + for name in ["tasks", "captures", "identities"] { + assert_eq!( + fs::read_dir(format!("/var/lib/kf-judge/{name}")) + .unwrap() + .count(), + 0 + ); + } + } + fn execute_with( + &mut self, + language: v2::Language, + source: &[u8], + points: &[(&[u8], &[u8])], + verdict: v1::Verdict, + legacy: bool, + edit: impl FnOnce(&mut serde_json::Value), + ) -> serde_json::Value { + let job = self.make_job(language, source, points, legacy, edit); + let output = self.worker_command(legacy).output().unwrap(); if !output.stderr.is_empty() { println!( "KF_PIPELINE_WORKER_DIAGNOSTIC {}", @@ -299,7 +366,11 @@ impl Fixture { if legacy { 1 } else { 2 } ); let stages = actual["stages"].as_array().unwrap(); - assert!(!stages.is_empty()); + assert!( + !stages.is_empty() + || (verdict == v1::Verdict::SystemError + && actual["result"]["report"]["failure_code"] == "artifact_unavailable") + ); let mut ids = std::collections::HashSet::new(); for s in stages { assert!(s["helper_pid"].as_u64().unwrap() > 1); @@ -307,38 +378,20 @@ impl Fixture { assert_eq!(s["report"]["errors"], serde_json::json!([])); assert_eq!(s["report"]["cleanup"], "complete"); } - assert_eq!( - stages.len(), - if verdict == v1::Verdict::CompilationError { - 1 - } else { - points.len() + 1 - } - ); - assert_eq!(fs::read_dir(self.dir.join("staging")).unwrap().count(), 0); - let stopped = Command::new(std::env::current_exe().unwrap()) - .env_clear() - .uid(1000) - .gid(1000) - .args([ - "--ignored", - "--exact", - "ordinary_pipeline_stopped_client", - "--test-threads=1", - "--nocapture", - ]) - .output() - .unwrap(); - assert!(stopped.status.success(), "{stopped:?}"); - assert!(String::from_utf8_lossy(&stopped.stdout).contains("KF_PIPELINE_STOPPED=1")); - for name in ["tasks", "captures", "identities"] { + if verdict == v1::Verdict::SystemError { + assert!(stages.len() <= points.len() + 1); + } else { assert_eq!( - fs::read_dir(format!("/var/lib/kf-judge/{name}")) - .unwrap() - .count(), - 0 + stages.len(), + if verdict == v1::Verdict::CompilationError { + 1 + } else { + points.len() + 1 + } ); } + assert_eq!(fs::read_dir(self.dir.join("staging")).unwrap().count(), 0); + self.assert_drained(); println!( "KF_MANAGED_PIPELINE_PROOF {}", serde_json::json!({"worker_uid":1000,"language":language,"legacy":legacy,"job":job,"actual":actual,"staging_empty":true,"global_stopped":true}) @@ -507,3 +560,599 @@ fn ordinary_worker_runs_versioned_and_legacy_gnu() { ); println!("KF_MANAGED_GNU_PIPELINE_STATUS=0"); } + +#[test] +#[ignore = "actual ordinary Worker language/encoding/class boundary and artifact failures only in capped VM"] +fn ordinary_worker_encoding_and_class_bounds() { + let mut f = Fixture::new(); + let mut expected = "café".as_bytes().to_vec(); + expected.extend_from_slice(b"\0\xff\n"); + f.execute( + v2::Language::Python313, + b"# coding: latin-1\nimport os,sys\nassert not os.path.exists('old')\nopen('old','w').close()\nsys.stdout.buffer.write('caf\xe9'.encode('utf-8')+sys.stdin.buffer.read())\n", + &[(b"", "café".as_bytes()), (b"\0\xff\n", &expected)], + v1::Verdict::Accepted, + false, + ); + let java = r#"import java.io.*; public class Main extends Entry {} class Entry {static int count;public static void main(String[] a)throws Exception{if(++count!=1||new File("old").exists())throw new AssertionError();new File("old").createNewFile();System.out.print("\0你好");System.in.transferTo(System.out);}}"#; + let mut expected = "\0你好".as_bytes().to_vec(); + expected.extend_from_slice(b"\0\xff\n"); + let inherited = f.execute( + v2::Language::Java21, + java.as_bytes(), + &[(b"", "\0你好".as_bytes()), (b"\0\xff\n", &expected)], + v1::Verdict::Accepted, + false, + ); + assert_eq!( + inherited["result"]["report"]["compilation"]["program"]["files"] + .as_array() + .unwrap() + .len(), + 2 + ); + let mut classes = String::from( + "public class Main {static int n;public static void main(String[] a){if(++n!=1)throw new AssertionError();System.out.print(A254.value());}}\n", + ); + for i in 0..255 { + classes.push_str(&format!( + "class A{i:03} {{static String value(){{return \"last\";}}}}\n" + )); + } + let boundary = f.execute( + v2::Language::Java21, + classes.as_bytes(), + &[(b"", b"last"), (b"x", b"last")], + v1::Verdict::Accepted, + false, + ); + assert_eq!( + boundary["result"]["report"]["compilation"]["program"]["files"] + .as_array() + .unwrap() + .len(), + 256 + ); + classes.push_str("class Extra {}\n"); + let oversized = f.execute( + v2::Language::Java21, + classes.as_bytes(), + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + assert!(oversized["result"]["report"]["compilation"]["program"].is_null()); + assert!( + oversized["result"]["report"]["compilation"]["diagnostics"] + .as_str() + .unwrap() + .contains("256") + ); + f.execute( + v2::Language::Java21, + r#"public class Main {public static void main(String[] a){}} class Aé {}"#.as_bytes(), + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + let long_name = format!( + "public class Main {{public static void main(String[] a){{}}}} class A{} {{}}", + "x".repeat(123) + ); + f.execute( + v2::Language::Java21, + long_name.as_bytes(), + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + f.execute( + v2::Language::Python313, + b"# coding: unknown_codec\npass\n", + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + println!("KF_MANAGED_ENCODING_CLASS_BOUNDS_STATUS=0"); +} + +#[test] +#[ignore = "focused current-binary ordinary managed WA, syntax and entry errors in capped VM"] +fn ordinary_worker_rechecks_managed_grades_and_entry_errors() { + let mut f = Fixture::new(); + let points: &[(&[u8], &[u8])] = &[(b"", b""), (b"\0\xff\n", b"wrong")]; + for (language, source) in [ + (v2::Language::Python313, &b"import os,sys\nassert not os.path.exists('previous')\nassert not os.path.exists('expected')\nassert not os.path.exists('/kf')\nopen('previous','w').close()\nsys.stdout.buffer.write(sys.stdin.buffer.read())\n"[..]), + (v2::Language::Java21, &b"import java.io.*; public class Main {static int n;public static void main(String[] a)throws Exception{if(++n!=1||new File(\"previous\").exists()||new File(\"expected\").exists()||new File(\"/kf\").exists()||Main.class.getResource(\"Main.java\")!=null)throw new AssertionError();new File(\"previous\").createNewFile();System.in.transferTo(System.out);}}"[..]), + ] { + let judged=f.execute(language,source,points,v1::Verdict::WrongAnswer,false); + assert_eq!(judged["result"]["report"]["cases"][0]["verdict"],"accepted"); + assert_eq!(judged["result"]["report"]["cases"][1]["verdict"],"wrong_answer"); + } + f.execute( + v2::Language::Python313, + b"open('top-level','w').close()\ndef invalid(:\n", + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + let missing = f.execute( + v2::Language::Java21, + b"public class Main {static {if(System.nanoTime()>0){while(true){}}}}", + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + assert_eq!(missing["stages"].as_array().unwrap().len(), 1); + assert!( + missing["result"]["report"]["compilation"]["diagnostics"] + .as_str() + .unwrap() + .contains("Main") + ); + assert!(missing["result"]["report"]["compilation"]["program"].is_null()); + println!("KF_MANAGED_CURRENT_GRADES_STATUS=0"); +} + +#[test] +#[ignore = "actual ordinary Worker preparation limits and artifact failures only in capped VM"] +fn ordinary_worker_preparation_limits_and_artifact_failures() { + let mut f = Fixture::new(); + for (language, source, cpu) in [ + (v2::Language::Python313, b"x=0\n".repeat(60_000), 500_000), + ( + v2::Language::Java21, + b"public class Main {public static void main(String[] a){}}".to_vec(), + 1_000_000, + ), + ] { + let limited = f.execute_with( + language, + &source, + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + |m| { + m["compile_limits"]["cpu_time_us"] = serde_json::json!(cpu); + }, + ); + assert_eq!( + limited["result"]["report"]["compilation"]["status"], + "compilation_limit" + ); + assert!(limited["result"]["report"]["compilation"]["program"].is_null()); + assert!( + limited["result"]["case_run_ids"] + .as_array() + .unwrap() + .is_empty() + ); + } + // Manifest identities/digests remain valid. Corrupt only actual fixture bytes + // to prove the ordinary Worker fails closed at each read boundary. + for language in [v2::Language::Python313, v2::Language::Java21] { + let source: &[u8] = if language == v2::Language::Python313 { + b"import sys\nsys.stdout.buffer.write(b'answer')\n" + } else { + b"public class Main {public static void main(String[] a){System.out.print(\"answer\");}}" + }; + for (field, index, stage_count) in [ + ("source", None, 0), + ("input", Some(0), 1), + ("expected", Some(0), 2), + ] { + let artifacts = f.dir.join("artifacts"); + let actual = f.execute_with( + language, + source, + &[(b"input", b"answer")], + v1::Verdict::SystemError, + false, + |m| { + let a = match index { + None => &m[field], + Some(i) => &m["cases"][i][field], + }; + let path = artifacts.join(a["digest"].as_str().unwrap()); + let mut bytes = fs::read(&path).unwrap(); + bytes[0] ^= 1; + store(&path, &bytes); + }, + ); + assert_eq!(actual["stages"].as_array().unwrap().len(), stage_count); + assert_eq!( + actual["result"]["report"]["failure_code"], + "artifact_unavailable" + ); + if stage_count < 2 { + assert!(actual["result"]["report"]["run_cpu_time_us"].is_null()); + } + } + } + // A corrupted read must not poison the next accepted job or its slot. + f.execute( + v2::Language::Python313, + b"print('fresh')\n", + &[(b"", b"fresh\n")], + v1::Verdict::Accepted, + false, + ); + f.execute( + v2::Language::Java21, + b"public class Main {public static void main(String[] a){System.out.print(\"fresh\");}}", + &[(b"", b"fresh")], + v1::Verdict::Accepted, + false, + ); + println!("KF_MANAGED_LANGUAGE_BOUNDS_STATUS=0"); +} + +#[test] +#[ignore = "actual ordinary Worker kernel limits and spoofed language errors only in capped VM"] +fn ordinary_worker_trusted_limits_and_spoofed_oom() { + let mut f = Fixture::new(); + for (source, verdict) in [ + ( + &b"raise MemoryError('fake kernel OOM')\n"[..], + v1::Verdict::RuntimeError, + ), + (&b"while True: pass\n"[..], v1::Verdict::TimeLimitExceeded), + ( + &b"import os\nx=b'x'*65536\nwhile True: os.write(1,x)\n"[..], + v1::Verdict::OutputLimitExceeded, + ), + ( + &b"blocks=[]\nwhile True: blocks.append(bytearray(8*1024*1024))\n"[..], + v1::Verdict::MemoryLimitExceeded, + ), + ] { + let result = f.execute( + v2::Language::Python313, + source, + &[(b"", b"")], + verdict, + false, + ); + assert_eq!( + result["result"]["report"]["compilation"]["status"], + "succeeded" + ); + } + for (source,verdict) in [ + (&b"public class Main {public static void main(String[] a){throw new OutOfMemoryError(\"fake kernel OOM\");}}"[..],v1::Verdict::RuntimeError), + (&b"public class Main {public static void main(String[] a){while(true){}}}"[..],v1::Verdict::TimeLimitExceeded), + (&b"public class Main {public static void main(String[] a)throws Exception{byte[] b=new byte[65536];while(true)System.out.write(b);}}"[..],v1::Verdict::OutputLimitExceeded), + (&b"import sun.misc.Unsafe; import java.lang.reflect.*; public class Main {public static void main(String[] a)throws Exception{Field f=Unsafe.class.getDeclaredField(\"theUnsafe\");f.setAccessible(true);Unsafe u=(Unsafe)f.get(null);while(true){long p=u.allocateMemory(16777216);u.setMemory(p,16777216,(byte)1);}}}"[..],v1::Verdict::MemoryLimitExceeded), + ] {let result=f.execute(v2::Language::Java21,source,&[(b"",b"")],verdict,false);assert_eq!(result["result"]["report"]["compilation"]["status"],"succeeded");} + f.execute( + v2::Language::Python313, + b"print('after-limits')\n", + &[(b"", b"after-limits\n")], + v1::Verdict::Accepted, + false, + ); + f.execute(v2::Language::Java21,b"public class Main {public static void main(String[] a){System.out.print(\"after-limits\");}}",&[(b"",b"after-limits")],v1::Verdict::Accepted,false); + println!("KF_MANAGED_WORKER_LIMITS_STATUS=0"); +} + +#[test] +#[ignore = "actual Worker death with live Python fork/JVM threads and proven publication recovery only in capped VM"] +fn ordinary_worker_death_drains_tree_and_recovers_publication() { + let mut f = Fixture::new(); + for language in [v2::Language::Python313, v2::Language::Java21] { + let source: &[u8] = if language == v2::Language::Python313 { + b"import os,time\np=os.fork()\nif p==0:\n while True: time.sleep(1)\nopen('started','w').write('python-fork-live')\nwhile True: time.sleep(1)\n" + } else { + b"import java.nio.file.*; public class Main {public static void main(String[] a)throws Exception{for(int i=0;i<3;i++)new Thread(()->{while(true)try{Thread.sleep(1000);}catch(InterruptedException e){}}).start();Files.writeString(Path.of(\"started\"),\"java-threads-live\");while(true)Thread.sleep(1000);}}" + }; + let job = f.make_job(language, source, &[(b"", b"")], false, |_| {}); + let run_id = format!("r-{}-0", job["binding"]["attempt_id"].as_str().unwrap()); + let group = f.root.path.join(format!("kf-{run_id}")); + let mut worker = f + .worker_command(false) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let worker_pid = worker.id(); + let deadline = Instant::now() + Duration::from_secs(100); + let (members, marker, threads) = loop { + if let Some(status) = worker.try_wait().unwrap() { + let output = worker.wait_with_output().unwrap(); + panic!( + "worker ended before live marker ({status}): {} {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + let members: Vec = fs::read_to_string(group.join("cgroup.procs")) + .unwrap_or_default() + .split_whitespace() + .map(|p| p.parse().unwrap()) + .collect(); + if let Some((marker, threads)) = members.iter().find_map(|pid| { + let executable = fs::read_link(format!("/proc/{pid}/exe")).ok()?; + let expected = if language == v2::Language::Python313 { + "python3.13" + } else { + "java" + }; + if executable.file_name()?.to_str()? != expected { + return None; + } + let path = PathBuf::from(format!("/proc/{pid}/root/work/started")); + let marker = fs::read_to_string(path).ok()?; + let threads = fs::read_dir(format!("/proc/{pid}/task")).ok()?.count(); + let live = if language == v2::Language::Python313 { + members.len() >= 2 && marker == "python-fork-live" + } else { + threads >= 4 && marker == "java-threads-live" + }; + live.then_some((marker, threads)) + }) { + break (members, marker, threads); + } + assert!( + Instant::now() < deadline, + "actual user tree not observed in {group:?}" + ); + std::thread::sleep(Duration::from_millis(10)); + }; + assert!( + fs::read_dir(f.dir.join("staging")).unwrap().count() > 0, + "live ordinary publication required" + ); + worker.kill().unwrap(); + let killed = worker.wait_with_output().unwrap(); + assert!(!killed.status.success()); + f.assert_drained(); + assert!(!group.exists()); + for pid in &members { + assert!( + !PathBuf::from(format!("/proc/{pid}")).exists(), + "descendant {pid} remains" + ); + } + assert!( + fs::read_dir(f.dir.join("staging")).unwrap().count() > 0, + "dead publication must remain until authenticated ordinary recovery" + ); + println!( + "KF_MANAGED_WORKER_DEATH_PROOF {}", + serde_json::json!({"job":job,"language":language,"worker_uid":1000,"worker_pid":worker_pid,"observed_target_pids":members,"observed_threads":threads,"actual_marker":marker,"worker_exit":killed.status.to_string(),"global_stopped":true,"root_owned_zero":true,"dead_publication_present":true}) + ); + // A fresh ordinary Worker recovers the dead owner's journal, then proves + // its own prepare/run and exact bytes without inheriting the killed tree. + let (fresh, expected): (&[u8], &[u8]) = if language == v2::Language::Python313 { + ( + b"import os\nassert not os.path.exists('started')\nprint('recovered')\n", + b"recovered\n", + ) + } else { + (b"import java.io.*; public class Main {public static void main(String[] a){if(new File(\"started\").exists())throw new AssertionError();System.out.print(\"recovered\");}}",b"recovered") + }; + f.execute( + language, + fresh, + &[(b"", expected)], + v1::Verdict::Accepted, + false, + ); + } + println!("KF_MANAGED_WORKER_DEATH_STATUS=0"); +} + +#[test] +#[ignore = "actual ordinary Worker production adapter cancellation during check/javac and live trees in capped VM"] +fn ordinary_worker_control_cancels_preparation_and_live_runs() { + let mut f = Fixture::new(); + for language in [v2::Language::Python313, v2::Language::Java21] { + for preparing in [true, false] { + let source = if preparing { + if language == v2::Language::Python313 { + b"x=0\n".repeat(60_000) + } else { + let mut s = + String::from("public class Main {public static void main(String[] a){}}\n"); + for i in 0..1200 { + s.push_str(&format!("class Many{i} {{int i{i}=1;}}\n")); + } + s.into_bytes() + } + } else if language == v2::Language::Python313 { + b"import os,time\np=os.fork()\nif p==0:\n while True: time.sleep(1)\nopen('started','w').write('python-fork-live')\nwhile True: time.sleep(1)\n".to_vec() + } else { + b"import java.nio.file.*; public class Main {public static void main(String[] a)throws Exception{for(int i=0;i<3;i++)new Thread(()->{while(true)try{Thread.sleep(1000);}catch(InterruptedException e){}}).start();Files.writeString(Path.of(\"started\"),\"java-threads-live\");while(true)Thread.sleep(1000);}}".to_vec() + }; + let job = f.make_job( + language, + &source, + &[(b"", b""), (b"next", b"next")], + false, + |_| {}, + ); + let stage_id = if preparing { + format!("c-{}", job["binding"]["attempt_id"].as_str().unwrap()) + } else { + format!("r-{}-0", job["binding"]["attempt_id"].as_str().unwrap()) + }; + let group = f.root.path.join(format!("kf-{stage_id}")); + let mut worker = Command::new(std::env::var_os("KF_WORKER_CONTROL_TEST_BIN").unwrap()) + .env_clear() + .uid(1000) + .gid(1000) + .env("KF_SANDBOX_INDEPENDENT_VM", "1") + .env("KF_WORKER_CONTROL_DIR", &f.dir) + .args([ + "--ignored", + "--exact", + "runtime_cancel_from_trusted_control", + "--test-threads=1", + "--nocapture", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let worker_pid = worker.id(); + let deadline = Instant::now() + Duration::from_secs(100); + let (members, observed) = loop { + assert!( + worker.try_wait().unwrap().is_none(), + "worker ended before observed cancellable stage" + ); + let members: Vec = fs::read_to_string(group.join("cgroup.procs")) + .unwrap_or_default() + .split_whitespace() + .map(|p| p.parse().unwrap()) + .collect(); + let observed = members.iter().find_map(|pid| { + if preparing { + let exe = fs::read_link(format!("/proc/{pid}/exe")).ok()?; + let expected = if language == v2::Language::Python313 { + "python3.13" + } else { + "javac" + }; + if exe.file_name()?.to_str()? != expected { + return None; + } + let cmd = fs::read(format!("/proc/{pid}/cmdline")).ok()?; + let required = if language == v2::Language::Python313 { + "/usr/lib/kf/python_check.py" + } else { + "/work/Main.java" + }; + if !cmd.split(|b| *b == 0).any(|a| a == required.as_bytes()) { + return None; + } + Some(format!("isolated-{expected}-exec-with-fixed-source")) + } else { + let marker = + fs::read_to_string(format!("/proc/{pid}/root/work/started")).ok()?; + let threads = fs::read_dir(format!("/proc/{pid}/task")).ok()?.count(); + if (language == v2::Language::Python313 + && members.len() >= 2 + && marker == "python-fork-live") + || (language == v2::Language::Java21 + && threads >= 4 + && marker == "java-threads-live") + { + Some(marker) + } else { + None + } + } + }); + if let Some(observed) = observed { + break (members, observed); + } + assert!( + Instant::now() < deadline, + "requested actual stage was not observed" + ); + std::thread::sleep(Duration::from_millis(5)); + }; + store(&f.dir.join("cancel"), b"cancel\n"); + let output = worker.wait_with_output().unwrap(); + fs::remove_file(f.dir.join("cancel")).unwrap(); + assert!(output.status.success(), "{output:?}"); + let text = String::from_utf8(output.stdout).unwrap(); + let proof: serde_json::Value = serde_json::from_str( + text.lines() + .find_map(|l| l.split_once("KF_WORKER_CONTROL_PROOF ").map(|(_, p)| p)) + .unwrap_or_else(|| panic!("successful control test omitted proof: {text}")), + ) + .unwrap(); + assert_eq!(proof["result"]["report"]["verdict"], "cancelled"); + assert_eq!(proof["result"]["report"]["cleanup_complete"], true); + assert_eq!( + proof["stages"].as_array().unwrap().len(), + if preparing { 1 } else { 2 } + ); + assert_eq!( + proof["result"]["skipped"].as_array().unwrap().len(), + if preparing { 2 } else { 1 } + ); + f.assert_drained(); + assert!(!group.exists()); + for pid in &members { + assert!(!PathBuf::from(format!("/proc/{pid}")).exists()); + } + assert_eq!(fs::read_dir(f.dir.join("staging")).unwrap().count(), 0); + println!( + "KF_MANAGED_WORKER_CONTROL_PROOF {}", + serde_json::json!({"language":language,"preparing":preparing,"job":job,"worker_uid":1000,"worker_pid":worker_pid,"observed_target_pids":members,"observed":observed,"actual":proof,"staging_empty":true,"global_stopped":true}) + ); + } + let (fresh, expected): (&[u8], &[u8]) = if language == v2::Language::Python313 { + (b"print('after-cancel')\n", b"after-cancel\n") + } else { + (b"public class Main {public static void main(String[] a){System.out.print(\"after-cancel\");}}",b"after-cancel") + }; + f.execute( + language, + fresh, + &[(b"", expected)], + v1::Verdict::Accepted, + false, + ); + } + println!("KF_MANAGED_WORKER_CONTROL_STATUS=0"); +} + +#[test] +#[ignore = "actual ordinary Worker checked Python/compiled Java program cannot be replaced or forged in capped VM"] +fn ordinary_worker_uses_sealed_program_after_backing_mutation() { + let mut f = Fixture::new(); + for language in [v2::Language::Python313, v2::Language::Java21] { + let source: &[u8] = if language == v2::Language::Python313 { + b"import os,sys\nassert not os.path.exists('old')\nopen('old','w').close()\nsys.stdout.buffer.write(sys.stdin.buffer.read())\n" + } else { + b"import java.io.*; public class Main {static int n;public static void main(String[] a)throws Exception{if(++n!=1||new File(\"old\").exists())throw new AssertionError();new File(\"old\").createNewFile();Helper.copy();}} class Helper {static void copy()throws Exception{System.in.transferTo(System.out);}}" + }; + let job = f.make_job( + language, + source, + &[(b"", b""), (b"\0\xff\n", b"\0\xff\n")], + false, + |_| {}, + ); + let output = Command::new(std::env::var_os("KF_WORKER_CONTROL_TEST_BIN").unwrap()) + .env_clear() + .uid(1000) + .gid(1000) + .env("KF_SANDBOX_INDEPENDENT_VM", "1") + .env("KF_WORKER_CONTROL_DIR", &f.dir) + .args([ + "--ignored", + "--exact", + "runtime_checked_program_survives_backing_mutation_and_rejects_forgery", + "--test-threads=1", + "--nocapture", + ]) + .output() + .unwrap(); + assert!(output.status.success(), "{output:?}"); + let text = String::from_utf8(output.stdout).unwrap(); + let proof: serde_json::Value = serde_json::from_str( + text.lines() + .find_map(|l| { + l.split_once("KF_WORKER_SEALED_PROGRAM_PROOF ") + .map(|(_, p)| p) + }) + .unwrap_or_else(|| panic!("successful sealed-program test omitted proof: {text}")), + ) + .unwrap(); + assert_eq!(proof["rejected_metadata_mutations"], 14); + assert_eq!(proof["backing_changed_after_real_preparation"], true); + assert_eq!(proof["stages"].as_array().unwrap().len(), 3); + f.assert_drained(); + assert_eq!(fs::read_dir(f.dir.join("staging")).unwrap().count(), 0); + println!( + "KF_MANAGED_WORKER_SEALED_PROGRAM_PROOF {}", + serde_json::json!({"language":language,"job":job,"worker_uid":1000,"actual":proof,"staging_empty":true,"global_stopped":true}) + ); + } + println!("KF_MANAGED_WORKER_SEALED_PROGRAM_STATUS=0"); +} diff --git a/judge/crates/sandbox/tests/managed_runtime_vm.rs b/judge/crates/sandbox/tests/managed_runtime_vm.rs index 5a383de..6593a83 100644 --- a/judge/crates/sandbox/tests/managed_runtime_vm.rs +++ b/judge/crates/sandbox/tests/managed_runtime_vm.rs @@ -876,3 +876,135 @@ public class Main { } println!("KF_MANAGED_FILE_WALL_STATUS=0 java21"); } + +#[test] +#[ignore = "actual held Java class-directory kernel links/seals/bounds only in independently capped VM"] +fn java_class_exports_reject_links_replacement_and_partial_batches() { + use kf_sandbox::launcher::root::RootedTarget; + use rustix::fd::{AsFd, AsRawFd}; + use std::io::Read; + use std::os::unix::fs::symlink; + let f = Fixture::new("java21", ImageKind::LinuxOpenJdk21X86_64V1); + let mut request = f.request("java-export-refusals", ExecutionProfile::JavaCompile); + request.limits.file_bytes = 16 * 1024; + request.limits.workdir_bytes = 64 * 1024; + request.limits.stdout_bytes = 1; + request.limits.stderr_bytes = 1; + let config = f.root.validated(); + let spec = request.validate(&config.spec_policy()).unwrap(); + let target = RootedTarget::prepare( + &config, + &spec, + &PathBuf::from(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap()), + ) + .unwrap(); + let work = PathBuf::from(format!("/proc/{}/root/work", target.init_pid())); + let candidate = work.join("Main.class"); + let canary = f.dir.join("outside-canary"); + fs::write(&canary, b"never-export").unwrap(); + let inode = fs::metadata(&canary).unwrap().ino(); + let same_mount_canary = work.join("Canary.class"); + fs::write(&same_mount_canary, b"never-export").unwrap(); + let work_inode = fs::metadata(&same_mount_canary).unwrap().ino(); + let descriptors = || fs::read_dir("/proc/self/fd").unwrap().count(); + let before = descriptors(); + for variant in 0..7 { + match variant { + 0 => symlink(&canary, &candidate).unwrap(), + 1 => fs::hard_link(&same_mount_canary, &candidate).unwrap(), + 2 => fs::create_dir(&candidate).unwrap(), + 3 => symlink("/proc/self/fd/0", &candidate).unwrap(), + 4 => fs::write(&candidate, []).unwrap(), + 5 => fs::write(&candidate, [0; 16 * 1024 + 1]).unwrap(), + 6 => { + fs::write(&candidate, b"class").unwrap(); + fs::rename(&candidate, work.join("Main$Held.class")).unwrap(); + symlink(&canary, &candidate).unwrap(); + } + _ => unreachable!(), + } + assert!(target.snapshot_outputs().is_err(), "variant {variant}"); + assert_eq!(fs::metadata(&canary).unwrap().ino(), inode); + assert_eq!(fs::read(&canary).unwrap(), b"never-export"); + assert_eq!(fs::metadata(&same_mount_canary).unwrap().ino(), work_inode); + assert_eq!(fs::read(&same_mount_canary).unwrap(), b"never-export"); + if variant == 2 { + fs::remove_dir(&candidate).unwrap(); + } else { + fs::remove_file(&candidate).unwrap(); + } + if variant == 6 { + fs::remove_file(work.join("Main$Held.class")).unwrap(); + } + assert_eq!(descriptors(), before); + } + fs::remove_file(&same_mount_canary).unwrap(); + fs::write(&candidate, [1; 16 * 1024]).unwrap(); + fs::write(work.join("Main$1.class"), [2; 16 * 1024]).unwrap(); + for name in ["Third.class", "Fourth.class", "Fifth.class"] { + fs::File::create(work.join(name)) + .unwrap() + .set_len(16 * 1024) + .unwrap(); + } + assert!(target.snapshot_outputs().is_err(), "total logical bytes"); + assert_eq!(descriptors(), before, "partial batch must drop every FD"); + for name in ["Third.class", "Fourth.class", "Fifth.class"] { + fs::remove_file(work.join(name)).unwrap(); + } + let sealed = target.snapshot_outputs().unwrap(); + assert_eq!(sealed.len(), 2); + assert_eq!( + sealed.iter().map(|o| o.file.bytes()).sum::(), + 32 * 1024 + ); + for output in &sealed { + let mut bytes = vec![]; + output + .file + .reader() + .unwrap() + .read_to_end(&mut bytes) + .unwrap(); + let expected = if output.path == "Main.class" { + vec![1; 16 * 1024] + } else { + vec![2; 16 * 1024] + }; + assert_eq!(bytes, expected); + fs::write(work.join(&output.path), b"changed backing").unwrap(); + let fd = rustix::fs::open( + format!("/proc/self/fd/{}", output.file.as_fd().as_raw_fd()), + rustix::fs::OFlags::RDWR | rustix::fs::OFlags::CLOEXEC, + rustix::fs::Mode::empty(), + ) + .unwrap(); + assert_eq!( + rustix::io::pwrite(&fd, b"tamper", 0), + Err(rustix::io::Errno::PERM) + ); + assert_eq!(rustix::fs::ftruncate(&fd, 0), Err(rustix::io::Errno::PERM)); + bytes.clear(); + output + .file + .reader() + .unwrap() + .read_to_end(&mut bytes) + .unwrap(); + assert_eq!(bytes, expected); + } + drop(sealed); + assert_eq!(descriptors(), before); + assert!( + target + .refuse_remaining_isolation() + .cleanup_failure + .is_none() + ); + f.empty(); + fs::remove_file(&canary).unwrap(); + println!( + "KF_JAVA_CLASS_DIRECTORY_PROOF symlink=1 hardlink=1 directory=1 magiclink=1 empty=1 file_limit=1 total_limit=1 replacement=1 canary_inode_bytes_preserved=1 partial_fd_cleanup=1 kernel_seals=1 backing_mutation_safe=1 owned_zero=1" + ); + println!("KF_JAVA_CLASS_DIRECTORY_STATUS=0"); +} diff --git a/judge/crates/worker/tests/runtime_control_vm.rs b/judge/crates/worker/tests/runtime_control_vm.rs new file mode 100644 index 0000000..82f2722 --- /dev/null +++ b/judge/crates/worker/tests/runtime_control_vm.rs @@ -0,0 +1,267 @@ +#![cfg(target_os = "linux")] +//! Ordinary production adapter controlled by a trusted, independent VM fixture. +//! This ignored test never runs submitted source on the development host. +use kf_contracts::{judging as v1, judging_v2 as v2, judging_versions::NodeAdmission}; +use kf_worker::{ + pipeline::LocalArtifacts, + pipeline_v2::{Job, RuntimeAttempt}, + runtime_specs::AuthorizedJob, +}; +use std::{ + fs, + os::unix::fs::MetadataExt, + path::PathBuf, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::{Duration, Instant}, +}; + +fn fixture_dir() -> PathBuf { + assert_eq!( + std::env::var("KF_SANDBOX_INDEPENDENT_VM").as_deref(), + Ok("1") + ); + assert_eq!( + fs::read_to_string("/kf/INDEPENDENT_VM").unwrap().trim(), + "isolated-test-guest" + ); + assert!( + fs::read_to_string("/proc/self/status") + .unwrap() + .lines() + .any(|l| l.split_whitespace().collect::>() + == ["Uid:", "1000", "1000", "1000", "1000"]) + ); + assert_eq!( + fs::read_dir("/sys/class/net") + .unwrap() + .map(|e| e.unwrap().file_name()) + .collect::>(), + ["lo"] + ); + for path in ["/mnt/hgfs", "/home/kronecker", "/root/.ssh"] { + assert!(!std::path::Path::new(path).exists()); + } + let mounts = fs::read_to_string("/proc/self/mountinfo").unwrap(); + assert!( + ["hgfs", "9p", "virtiofs"] + .iter() + .all(|m| !mounts.contains(m)) + ); + let dir = PathBuf::from(std::env::var_os("KF_WORKER_CONTROL_DIR").unwrap()); + assert_eq!( + fs::metadata(dir.join("admission.json")).unwrap().uid(), + 1000 + ); + dir +} + +#[test] +#[ignore = "actual ordinary Worker cancellation only in independently capped no-NIC/no-share VM"] +fn runtime_cancel_from_trusted_control() { + let dir = fixture_dir(); + let admission: NodeAdmission = + v1::decode_wire(&fs::read(dir.join("admission.json")).unwrap()).unwrap(); + let job: Job = v1::decode_wire(&fs::read(dir.join("job.json")).unwrap()).unwrap(); + let authorized = AuthorizedJob::admit( + job.manifest.clone(), + job.binding.clone(), + job.manifest_digest.clone(), + job.attempt_number, + &admission, + ) + .unwrap(); + let cancel = Arc::new(AtomicBool::new(false)); + let finished = Arc::new(AtomicBool::new(false)); + let watch_cancel = cancel.clone(); + let watch_finished = finished.clone(); + let control = dir.join("cancel"); + let watcher = std::thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(110); + while !watch_finished.load(Ordering::Acquire) { + if fs::read(&control).is_ok_and(|b| b == b"cancel\n") { + watch_cancel.store(true, Ordering::Release); + return; + } + assert!( + Instant::now() < deadline, + "trusted fixture did not deliver cancellation" + ); + std::thread::sleep(Duration::from_millis(5)); + } + }); + let mut attempt = RuntimeAttempt::new( + authorized, + LocalArtifacts::open(&dir.join("artifacts")).unwrap(), + "/var/lib/kf-judge/factory/control.sock".into(), + dir.join("staging"), + &cancel, + ) + .unwrap(); + let result = kf_engine::submission_v2::judge_submission( + &mut attempt, + &job.manifest, + job.binding, + job.manifest_digest, + ) + .unwrap(); + finished.store(true, Ordering::Release); + watcher.join().unwrap(); + assert!(cancel.load(Ordering::Acquire)); + assert_eq!(result.report.verdict, v1::Verdict::Cancelled); + assert!(result.report.cleanup_complete); + assert!(result.report.failure_code.is_none()); + assert!(result.skipped.len() <= job.manifest.cases.len()); + println!( + "KF_WORKER_CONTROL_PROOF {}", + serde_json::to_string(&kf_worker::pipeline_v2::Judged { + result, + stages: attempt.into_stages() + }) + .unwrap() + ); +} + +#[test] +#[ignore = "actual prepared program seals and metadata fencing only in capped no-NIC/no-share VM"] +fn runtime_checked_program_survives_backing_mutation_and_rejects_forgery() { + use kf_engine::submission_v2::{PreparationPort, RunPort}; + let dir = fixture_dir(); + let admission: NodeAdmission = + v1::decode_wire(&fs::read(dir.join("admission.json")).unwrap()).unwrap(); + let job: Job = v1::decode_wire(&fs::read(dir.join("job.json")).unwrap()).unwrap(); + let authorized = AuthorizedJob::admit( + job.manifest.clone(), + job.binding.clone(), + job.manifest_digest.clone(), + job.attempt_number, + &admission, + ) + .unwrap(); + let cancel = AtomicBool::new(false); + let mut attempt = RuntimeAttempt::new( + authorized, + LocalArtifacts::open(&dir.join("artifacts")).unwrap(), + "/var/lib/kf-judge/factory/control.sock".into(), + dir.join("staging"), + &cancel, + ) + .unwrap(); + let compiled = attempt + .prepare( + &job.manifest, + &job.binding, + &format!("c-{}", job.binding.attempt_id.as_uuid()), + ) + .unwrap(); + assert_eq!(compiled.report.status, v1::CompilationStatus::Succeeded); + assert!(compiled.cleanup_complete); + let program = compiled.report.program.clone().unwrap(); + let run_id = format!("r-{}-0", job.binding.attempt_id.as_uuid()); + let mut bads = Vec::new(); + let mut bad = program.clone(); + bad.binding.fencing_version += 1; + bads.push(bad); + let mut bad = program.clone(); + bad.binding.attempt_id = "bbbbbbbb-bbbb-4bbb-8bbb-000000000001" + .to_string() + .try_into() + .unwrap(); + bads.push(bad); + let mut bad = program.clone(); + bad.source.digest = "f".repeat(64).try_into().unwrap(); + bads.push(bad); + let mut bad = program.clone(); + bad.toolchain.image_revision += 1; + bads.push(bad); + let mut bad = program.clone(); + bad.kind = v2::ProgramKind::NativeExecutable; + bads.push(bad); + let mut bad = program.clone(); + bad.files[0].path = "../Main.class".into(); + bads.push(bad); + let mut bad = program.clone(); + bad.files[0].artifact.bytes = 0; + bads.push(bad); + let mut bad = program.clone(); + bad.files[0].artifact.digest = "f".repeat(64).try_into().unwrap(); + bads.push(bad); + let mut bad = program.clone(); + bad.files.push(bad.files[0].clone()); + bads.push(bad); + let mut bad = program.clone(); + bad.files.clear(); + bads.push(bad); + let mut bad = program.clone(); + bad.source.bytes = v1::MAX_SOURCE_BYTES + 1; + bads.push(bad); + for path in ["main.pyc", "program.zip"] { + let mut bad = program.clone(); + bad.files[0].path = path.into(); + bads.push(bad); + } + let mut bad = program.clone(); + let file = bad.files[0].clone(); + bad.files = (0..257) + .map(|i| { + let mut file = file.clone(); + file.path = if i == 0 { + program.files[0].path.clone() + } else { + format!("Extra{i}.class") + }; + file + }) + .collect(); + bads.push(bad); + let rejected = bads.len(); + for bad in bads { + let error = attempt + .run( + &bad, + &job.manifest.cases[0], + &job.manifest.run_limits, + &run_id, + ) + .err() + .expect("forged program must be refused before Root"); + assert_eq!(error.code, v1::FailureCode::ArtifactUnavailable); + assert!(error.cleanup_complete); + } + // Only the VM fixture's original backing artifact is changed. The ordinary + // adapter must use its exact previously checked/compiled sealed capabilities. + let backing = dir + .join("artifacts") + .join(job.manifest.source.digest.as_str()); + fs::write( + &backing, + b"corrupted source backing after actual preparation", + ) + .unwrap(); + let mut observations = Vec::new(); + for (i, case) in job.manifest.cases.iter().enumerate() { + let observation = attempt + .run( + &program, + case, + &job.manifest.run_limits, + &format!("r-{}-{i}", job.binding.attempt_id.as_uuid()), + ) + .unwrap(); + assert_eq!( + observation.kind, + kf_engine::ExecutionKind::Exited { code: 0 } + ); + let expected = fs::read(dir.join("artifacts").join(case.expected.digest.as_str())).unwrap(); + assert_eq!(observation.stdout, expected); + observations.push(serde_json::json!({"case_id":case.case_id,"cpu_time_us":observation.cpu_time_us,"memory_peak_bytes":observation.memory_peak_bytes,"exact_bytes":true})); + } + let stages = attempt.into_stages(); + assert_eq!(stages.len(), job.manifest.cases.len() + 1); + println!( + "KF_WORKER_SEALED_PROGRAM_PROOF {}", + serde_json::json!({"program":program,"compilation":compiled.report,"stages":stages,"observations":observations,"rejected_metadata_mutations":rejected,"backing_changed_after_real_preparation":true,"same_rejected_run_id_later_executed":true}) + ); +}