test(judge): verify managed worker faults and sealed programs
This commit is contained in:
@@ -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::<Vec<_>>()
|
||||
);
|
||||
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<u32> = 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<u32> = 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");
|
||||
}
|
||||
|
||||
@@ -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::<u64>(),
|
||||
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");
|
||||
}
|
||||
|
||||
@@ -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::<Vec<_>>()
|
||||
== ["Uid:", "1000", "1000", "1000", "1000"])
|
||||
);
|
||||
assert_eq!(
|
||||
fs::read_dir("/sys/class/net")
|
||||
.unwrap()
|
||||
.map(|e| e.unwrap().file_name())
|
||||
.collect::<Vec<_>>(),
|
||||
["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})
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user