From 802e019699661824838004a3bf0b0f333b17f70b Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Fri, 9 Oct 2026 20:15:17 +0800 Subject: [PATCH] feat(worker): judge versioned sealed runtime programs --- backend/tests/test_coordinator.py | 1 + backend/tests/test_gnu_image_builder.py | 101 ++++ .../tests/test_multi_runtime_coordinator.py | 7 + judge/crates/coordinator/src/repository.rs | 14 +- judge/crates/engine/src/lib.rs | 1 + judge/crates/engine/src/submission.rs | 2 +- judge/crates/engine/src/submission_v2.rs | 478 ++++++++++++++++ judge/crates/sandbox/src/artifacts.rs | 110 +++- judge/crates/sandbox/src/image.rs | 43 +- judge/crates/sandbox/src/launcher/protocol.rs | 32 +- judge/crates/sandbox/src/rpc.rs | 4 +- .../sandbox/tests/indexed_staging_vm.rs | 142 +++++ .../sandbox/tests/managed_pipeline_vm.rs | 509 ++++++++++++++++++ judge/crates/worker/src/consumer.rs | 324 +++++++++-- judge/crates/worker/src/java_entry.rs | 212 ++++++++ judge/crates/worker/src/lib.rs | 9 + judge/crates/worker/src/main.rs | 42 ++ judge/crates/worker/src/pipeline.rs | 15 +- judge/crates/worker/src/pipeline_v2.rs | 414 ++++++++++++++ judge/crates/worker/src/runtime_specs.rs | 310 +++++++++++ scripts/build_gnu_image.py | 181 +++++++ 21 files changed, 2867 insertions(+), 84 deletions(-) create mode 100644 backend/tests/test_gnu_image_builder.py create mode 100644 judge/crates/engine/src/submission_v2.rs create mode 100644 judge/crates/sandbox/tests/indexed_staging_vm.rs create mode 100644 judge/crates/sandbox/tests/managed_pipeline_vm.rs create mode 100644 judge/crates/worker/src/java_entry.rs create mode 100644 judge/crates/worker/src/pipeline_v2.rs create mode 100644 judge/crates/worker/src/runtime_specs.rs create mode 100644 scripts/build_gnu_image.py diff --git a/backend/tests/test_coordinator.py b/backend/tests/test_coordinator.py index 4250280..9ac63e7 100644 --- a/backend/tests/test_coordinator.py +++ b/backend/tests/test_coordinator.py @@ -439,6 +439,7 @@ def test_actual_expiry_stopping_quarantine_retry_and_old_attempt( break await asyncio.sleep(0.05) assert pending[0]["binding"] == g["binding"] + assert pending == [dict(binding=g["binding"], case_count=2)] stopped = dict( schema_version=1, kind="judge.stopped", binding=g["binding"], cleanup_complete=False ) diff --git a/backend/tests/test_gnu_image_builder.py b/backend/tests/test_gnu_image_builder.py new file mode 100644 index 0000000..daacb70 --- /dev/null +++ b/backend/tests/test_gnu_image_builder.py @@ -0,0 +1,101 @@ +"""Offline package data only; no compiler, submitted source or image execution.""" + +import importlib.util +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +SPEC = importlib.util.spec_from_file_location( + "gnu_image_builder", Path(__file__).resolve().parents[2] / "scripts/build_gnu_image.py" +) +builder = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(builder) + + +def test_absolute_package_link_is_relocated_inside_verified_image(tmp_path): + root = tmp_path / "image" + target = root / "usr/lib/plugin.so" + target.parent.mkdir(parents=True) + target.write_bytes(b"synthetic pinned package bytes") + link = root / "usr/lib/plugin-link.so" + link.symlink_to("/usr/lib/plugin.so") + records = { + "usr/lib/plugin-link.so": {"kind": "symlink", "target": "/usr/lib/plugin.so"}, + "usr/lib/plugin.so": {"kind": "file", "sha256": builder.sha(target)}, + } + source, record = builder.resolve_verified(root, records, "usr/lib/plugin-link.so") + assert source == target + assert source.read_bytes() == b"synthetic pinned package bytes" + assert record == records["usr/lib/plugin.so"] + + +@pytest.mark.parametrize("target", ["../../../../outside", "plugin-link.so"]) +def test_package_link_escape_or_cycle_refused(tmp_path, target): + link = tmp_path / "usr/lib/plugin-link.so" + link.parent.mkdir(parents=True) + link.symlink_to(target) + records = {"usr/lib/plugin-link.so": {"kind": "symlink", "target": target}} + with pytest.raises(ValueError): + builder.resolve_verified(tmp_path, records, "usr/lib/plugin-link.so") + + +def test_replaced_or_unverified_ancestor_link_is_refused(tmp_path): + library = tmp_path / "usr/lib" + library.parent.mkdir() + library.symlink_to("/usr/lib") + records = {"usr/lib/plugin.so": {"kind": "file", "sha256": "a" * 64}} + with pytest.raises(ValueError, match="unverified GNU ancestor"): + builder.resolve_verified(tmp_path, records, "usr/lib/plugin.so") + records["usr/lib"] = {"kind": "symlink", "target": "/different/path"} + with pytest.raises(ValueError, match="differs from verified manifest"): + builder.resolve_verified(tmp_path, records, "usr/lib/plugin.so") + + +def test_verified_manifest_does_not_authorize_changed_package_bytes(tmp_path): + root = tmp_path / "support" + library = root / "usr/lib/ld-linux-x86-64.so.2" + library.parent.mkdir(parents=True) + library.write_bytes(b"original synthetic library") + manifest = { + "packages": [ + { + "name": "gcc", + "version": "16.2.1+r23+gd564253eb6c8-1", + "sha256": "ebac12ceef494c616a36fc6f0f792b365dc59b6ed46ca1a69f138aecc810b6ab", + }, + { + "name": "glibc", + "version": "2.44+r50+g1848099f063e-3", + "sha256": "071d807f4d3bf6b921f86b9a91e83a47ecc026c5821634efa46b09d890e06785", + }, + { + "name": "binutils", + "version": "2.47-2", + "sha256": "62efd213e87590f025cf861b646793d3cacb03e7e26187b73d33b00adaf7a977", + }, + ], + "files": [ + { + "path": "usr/lib/ld-linux-x86-64.so.2", + "kind": "file", + "bytes": library.stat().st_size, + "sha256": builder.sha(library), + } + ], + } + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text(json.dumps(manifest)) + library.write_bytes(b"changed synthetic library") + output = tmp_path / "output" + args = SimpleNamespace( + support_dir=root, + support_manifest=manifest_path, + support_manifest_sha256=builder.sha(manifest_path), + output=output, + ) + with pytest.raises(ValueError, match="differs from verified public package bytes"): + builder.build(args) + assert not (output / "manifest.json").exists() + assert not list(output.rglob("*.so*")) diff --git a/backend/tests/test_multi_runtime_coordinator.py b/backend/tests/test_multi_runtime_coordinator.py index 1e411f3..387de73 100644 --- a/backend/tests/test_multi_runtime_coordinator.py +++ b/backend/tests/test_multi_runtime_coordinator.py @@ -209,6 +209,13 @@ def test_actual_v2_cancel_requires_matching_stop_protocol_and_fence( headers={"Authorization": "Bearer " + platform}, ) ).status_code == 204 + pending = await client.get( + "/internal/pending-stops", headers=worker_headers(workers[0]) + ) + assert pending.status_code == 200 + assert pending.json() == [ + dict(schema_version=2, binding=grant["binding"], case_count=2) + ] stopped = dict( schema_version=2, kind="judge.stopped", diff --git a/judge/crates/coordinator/src/repository.rs b/judge/crates/coordinator/src/repository.rs index e04e40a..a30bc6e 100644 --- a/judge/crates/coordinator/src/repository.rs +++ b/judge/crates/coordinator/src/repository.rs @@ -460,8 +460,18 @@ impl Repository { } pub async fn pending_stops(&self, worker: Identifier) -> Result { let s = self.database.session().await?; - let rows=s.client.query("SELECT e.id,a.id attempt_id,a.fencing_version,jsonb_array_length(a.manifest->'cases') case_count FROM evaluations e JOIN judge_attempts a ON a.id=e.current_attempt_id WHERE e.state='cancelling' AND a.worker_id=$1 ORDER BY e.updated_at,e.id LIMIT 16",&[&worker.as_uuid()]).await?; - Ok(Value::Array(rows.iter().map(|r|json!({"binding":LeaseBinding{evaluation_id:identifier(r.get("id")),attempt_id:identifier(r.get("attempt_id")),worker_id:worker,fencing_version:r.get::<_,i64>("fencing_version")as u64},"case_count":r.get::<_,i32>("case_count")})).collect())) + let rows=s.client.query("SELECT e.id,a.id attempt_id,a.fencing_version,jsonb_array_length(a.manifest->'cases') case_count,(a.manifest->>'schema_version')::int schema_version FROM evaluations e JOIN judge_attempts a ON a.id=e.current_attempt_id WHERE e.state='cancelling' AND a.worker_id=$1 ORDER BY e.updated_at,e.id LIMIT 16",&[&worker.as_uuid()]).await?; + let entries = rows.iter().map(|r| { + let mut entry=json!({"binding":LeaseBinding{evaluation_id:identifier(r.get("id")),attempt_id:identifier(r.get("attempt_id")),worker_id:worker,fencing_version:r.get::<_,i64>("fencing_version")as u64},"case_count":r.get::<_,i32>("case_count")}); + match r.get::<_,i32>("schema_version") { + // Preserve the old pending-stop response and all v1 report bytes. + 1 => {}, + 2 => entry["schema_version"] = json!(2), + _ => return Err(Error::Invalid), + } + Ok(entry) + }).collect::>>()?; + Ok(Value::Array(entries)) } pub async fn recover(&self) -> Result { let mut s = self.database.session().await?; diff --git a/judge/crates/engine/src/lib.rs b/judge/crates/engine/src/lib.rs index b6c2570..c82094d 100644 --- a/judge/crates/engine/src/lib.rs +++ b/judge/crates/engine/src/lib.rs @@ -229,3 +229,4 @@ mod tests { } pub mod submission; +pub mod submission_v2; diff --git a/judge/crates/engine/src/submission.rs b/judge/crates/engine/src/submission.rs index 58d3392..d18d5e7 100644 --- a/judge/crates/engine/src/submission.rs +++ b/judge/crates/engine/src/submission.rs @@ -62,7 +62,7 @@ fn empty_compile(status: CompilationStatus) -> CompilationReport { memory_peak_bytes: None, } } -fn classify(o: &Observation, expected: &[u8]) -> Verdict { +pub(crate) fn classify(o: &Observation, expected: &[u8]) -> Verdict { match o.kind { ExecutionKind::Exited { code: 0 } => { if o.stdout == expected { diff --git a/judge/crates/engine/src/submission_v2.rs b/judge/crates/engine/src/submission_v2.rs new file mode 100644 index 0000000..aa6a8c8 --- /dev/null +++ b/judge/crates/engine/src/submission_v2.rs @@ -0,0 +1,478 @@ +//! Explicit v2 preparation and typed program execution; legacy wire stays separate. +//! Ports own clocks/cancellation/storage/execution; this module is pure domain. +use super::submission::{ + ControlPort, ExpectedPort, SkipReason, SkippedCase, StageFailure, classify, +}; +use crate::{ExecutionKind, Observation}; +use kf_contracts::judging_v2::*; +use serde::Serialize; +pub struct Compiled { + pub report: CompilationReport, + pub cleanup_complete: bool, + pub failure: Option, +} +pub trait PreparationPort { + fn prepare( + &mut self, + manifest: &JobManifest, + binding: &LeaseBinding, + run_id: &str, + ) -> Result; +} +pub trait RunPort { + fn run( + &mut self, + program: &CompiledProgram, + case: &TestCase, + limits: &RunLimits, + run_id: &str, + ) -> Result; +} +#[derive(Debug, Serialize)] +pub struct SubmissionResult { + pub report: JudgementReport, + pub compile_run_id: String, + pub case_run_ids: Vec, + pub skipped: Vec, +} +fn empty_compile(status: CompilationStatus) -> CompilationReport { + CompilationReport { + status, + program: None, + diagnostics: String::new().try_into().expect("empty diagnostics"), + diagnostics_truncated: false, + cpu_time_us: None, + memory_peak_bytes: None, + } +} +fn failure(code: FailureCode) -> StageFailure { + StageFailure { + code, + cleanup_complete: true, + } +} +fn gate(p: &impl ControlPort, m: &JobManifest) -> Result { + if p.elapsed_us() >= m.job_wall_time_us { + return Err(failure(FailureCode::JobDeadline)); + } + Ok(p.cancelled()) +} +pub fn judge_submission( + p: &mut P, + m: &JobManifest, + binding: LeaseBinding, + digest: Digest, +) -> Result { + m.validate()?; + binding.validate()?; + if binding.evaluation_id != m.evaluation_id { + return Err("submission binding differs"); + } + let compile_run_id = format!("c-{}", binding.attempt_id.as_uuid()); + let compilation = match gate(p, m) { + Ok(true) => Ok(Compiled { + report: empty_compile(CompilationStatus::Cancelled), + cleanup_complete: true, + failure: None, + }), + Err(e) => Err(e), + Ok(false) => p + .stage(Stage::Compiling, None) + .and_then(|()| p.prepare(m, &binding, &compile_run_id)), + }; + let mut cleanup_complete; + let mut failure_code; + let mut compilation = match compilation { + Ok(c) => { + c.report.validate()?; + if c.report.program.as_ref().is_some_and(|program| { + program.binding != binding + || program.source != m.source + || program.toolchain != m.toolchain + || program.kind != m.language.contract().program_kind + || (program.kind != ProgramKind::PythonSource + && (program + .files + .iter() + .any(|f| f.artifact.bytes > m.compile_limits.file_bytes) + || program.bytes() > m.compile_limits.workdir_bytes)) + }) { + return Err("preparation program differs from authorized manifest or lease"); + } + cleanup_complete = c.cleanup_complete; + failure_code = c.failure; + if !cleanup_complete { + failure_code = Some(FailureCode::CleanupFailed); + } + if failure_code.is_some() { + empty_compile(CompilationStatus::SystemError) + } else { + c.report + } + } + Err(e) => { + cleanup_complete = e.cleanup_complete; + failure_code = Some(if cleanup_complete { + e.code + } else { + FailureCode::CleanupFailed + }); + empty_compile(CompilationStatus::SystemError) + } + }; + if failure_code.is_none() && p.elapsed_us() >= m.job_wall_time_us { + failure_code = Some(FailureCode::JobDeadline); + if compilation.status != CompilationStatus::Succeeded { + compilation.status = CompilationStatus::SystemError; + compilation.program = None; + } + } + let mut cases = Vec::with_capacity(m.cases.len()); + let mut run_ids = Vec::with_capacity(m.cases.len()); + if let Some(program) = &compilation.program { + for (index, c) in m.cases.iter().enumerate() { + match gate(p, m) { + Ok(true) => break, + Err(e) => { + failure_code = Some(e.code); + break; + } + Ok(false) => {} + } + let run_id = format!("r-{}-{index}", binding.attempt_id.as_uuid()); + let observation = p + .stage(Stage::Running, Some(index as u16)) + .and_then(|()| p.run(program, c, &m.run_limits, &run_id)); + run_ids.push(run_id); + let result = match observation { + Err(e) => { + cleanup_complete = e.cleanup_complete; + failure_code = Some(if cleanup_complete { + e.code + } else { + FailureCode::CleanupFailed + }); + CaseReport { + case_id: c.case_id, + verdict: Verdict::SystemError, + cpu_time_us: None, + memory_peak_bytes: None, + } + } + Ok(o) => { + let verdict = if o.kind == ExecutionKind::InfrastructureFailure { + Verdict::SystemError + } else { + match gate(p, m) { + Err(e) => { + failure_code = Some(e.code); + Verdict::SystemError + } + Ok(true) => Verdict::Cancelled, + Ok(false) if !matches!(o.kind, ExecutionKind::Exited { code: 0 }) => { + classify(&o, &[]) + } + Ok(false) => match p.expected(c) { + Ok(bytes) + if bytes.len() as u64 == c.expected.bytes + && bytes.len() as u64 <= m.run_limits.stdout_bytes => + { + classify(&o, &bytes) + } + Ok(_) => { + failure_code = Some(FailureCode::ArtifactUnavailable); + Verdict::SystemError + } + Err(e) => { + failure_code = Some(e.code); + Verdict::SystemError + } + }, + } + }; + if verdict == Verdict::SystemError && failure_code.is_none() { + failure_code = Some(FailureCode::SandboxUnavailable); + } + CaseReport { + case_id: c.case_id, + verdict, + cpu_time_us: o.cpu_time_us, + memory_peak_bytes: o.memory_peak_bytes, + } + } + }; + let stop = matches!(result.verdict, Verdict::SystemError | Verdict::Cancelled); + cases.push(result); + if stop { + break; + } + } + } + let verdict = if failure_code.is_some() { + Verdict::SystemError + } else { + match compilation.status { + CompilationStatus::CompilationError | CompilationStatus::CompilationLimit => { + Verdict::CompilationError + } + CompilationStatus::Cancelled => Verdict::Cancelled, + CompilationStatus::SystemError => { + failure_code = Some(FailureCode::SandboxUnavailable); + Verdict::SystemError + } + CompilationStatus::Succeeded => { + if cases.len() != m.cases.len() + || cases.iter().any(|c| c.verdict == Verdict::Cancelled) + { + Verdict::Cancelled + } else { + cases + .iter() + .find(|c| c.verdict != Verdict::Accepted) + .map_or(Verdict::Accepted, |c| c.verdict) + } + } + } + }; + let cpu = if !cases.is_empty() && cases.iter().all(|c| c.cpu_time_us.is_some()) { + Some( + cases + .iter() + .try_fold(0_u64, |n, c| n.checked_add(c.cpu_time_us.unwrap_or(0))) + .ok_or("CPU sum overflow")?, + ) + } else { + None + }; + let memory = if !cases.is_empty() && cases.iter().all(|c| c.memory_peak_bytes.is_some()) { + cases.iter().filter_map(|c| c.memory_peak_bytes).max() + } else { + None + }; + let reason = if verdict == Verdict::Cancelled { + SkipReason::Cancelled + } else if verdict == Verdict::CompilationError { + SkipReason::CompilationFailed + } else { + SkipReason::InfrastructureFailure + }; + let skipped = m.cases[cases.len()..] + .iter() + .map(|c| SkippedCase { + case_id: c.case_id, + reason, + }) + .collect(); + let report = JudgementReport { + schema_version: SchemaVersion, + kind: ResultKind::Value, + binding, + manifest_digest: digest, + compilation, + cases, + verdict, + cleanup_complete, + failure_code, + run_cpu_time_us: cpu, + run_memory_peak_bytes: memory, + }; + report.validate_for_manifest(m, &report.manifest_digest)?; + Ok(SubmissionResult { + report, + compile_run_id, + case_run_ids: run_ids, + skipped, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + struct Port { + prepared: CompiledProgram, + observations: Vec, + prepare_status: CompilationStatus, + prepares: usize, + runs: usize, + expected_reads: usize, + cancel_after: usize, + clock: u64, + advance_us: u64, + fail_cleanup: bool, + } + impl PreparationPort for Port { + fn prepare( + &mut self, + _: &JobManifest, + _: &LeaseBinding, + _: &str, + ) -> Result { + self.prepares += 1; + Ok(Compiled { + report: CompilationReport { + status: self.prepare_status, + program: (self.prepare_status == CompilationStatus::Succeeded) + .then(|| self.prepared.clone()), + diagnostics: String::new().try_into().unwrap(), + diagnostics_truncated: false, + cpu_time_us: None, + memory_peak_bytes: Some(4), + }, + cleanup_complete: !self.fail_cleanup, + failure: None, + }) + } + } + impl RunPort for Port { + fn run( + &mut self, + program: &CompiledProgram, + _: &TestCase, + _: &RunLimits, + _: &str, + ) -> Result { + assert_eq!(program, &self.prepared); + self.runs += 1; + Ok(self.observations.remove(0)) + } + } + impl ExpectedPort for Port { + fn expected(&mut self, case: &TestCase) -> Result, StageFailure> { + self.expected_reads += 1; + Ok(vec![0; case.expected.bytes as usize]) + } + } + impl ControlPort for Port { + fn elapsed_us(&self) -> u64 { + self.clock + .saturating_add(self.runs as u64 * self.advance_us) + } + fn cancelled(&self) -> bool { + self.runs >= self.cancel_after + } + fn stage(&mut self, _: Stage, _: Option) -> Result<(), StageFailure> { + Ok(()) + } + } + fn fixture(language: &str) -> (JobManifest, LeaseBinding, Digest, Port) { + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../../tests/fixtures/contracts-v2"); + let m: JobManifest = kf_contracts::judging::decode_wire( + &std::fs::read(root.join(format!("judge-manifest.valid-{language}.json"))).unwrap(), + ) + .unwrap(); + let r: JudgementReport = kf_contracts::judging::decode_wire( + &std::fs::read(root.join(format!("judge-result.valid-{language}.json"))).unwrap(), + ) + .unwrap(); + let p = Port { + prepared: r.compilation.program.unwrap(), + observations: m + .cases + .iter() + .map(|case| Observation { + kind: ExecutionKind::Exited { code: 0 }, + stdout: vec![0; case.expected.bytes as usize], + cpu_time_us: Some(3), + memory_peak_bytes: None, + }) + .collect(), + prepare_status: CompilationStatus::Succeeded, + prepares: 0, + runs: 0, + expected_reads: 0, + cancel_after: usize::MAX, + clock: 0, + advance_us: 0, + fail_cleanup: false, + }; + (m, r.binding, r.manifest_digest, p) + } + #[test] + fn typed_prepare_once_same_program_every_point_and_unknown_measurements() { + for language in ["c17", "cpp20", "python313", "java21"] { + let (m, b, d, mut p) = fixture(language); + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(p.prepares, 1); + assert_eq!(p.runs, m.cases.len()); + assert_eq!(r.report.verdict, Verdict::Accepted); + assert_eq!(r.report.run_memory_peak_bytes, None); + assert_eq!(r.report.compilation.cpu_time_us, None); + assert_eq!(r.report.run_cpu_time_us, Some(3 * m.cases.len() as u64)); + assert!(r.skipped.is_empty()); + } + } + #[test] + fn bad_program_is_refused_before_any_execution_or_expected_read() { + let (m, b, d, mut p) = fixture("java21"); + p.prepared.source.digest = "f".repeat(64).try_into().unwrap(); + assert!(judge_submission(&mut p, &m, b, d).is_err()); + assert_eq!(p.runs, 0); + assert_eq!(p.expected_reads, 0); + let (m, b, d, mut p) = fixture("python313"); + p.prepared.binding.fencing_version += 1; + assert!(judge_submission(&mut p, &m, b, d).is_err()); + assert_eq!(p.runs, 0); + } + #[test] + fn preparation_failure_and_cleanup_precedence_never_fake_points() { + for status in [ + CompilationStatus::CompilationError, + CompilationStatus::CompilationLimit, + CompilationStatus::Cancelled, + ] { + let (m, b, d, mut p) = fixture("java21"); + p.prepare_status = status; + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(p.runs, 0); + assert_eq!(r.skipped.len(), m.cases.len()); + assert_eq!( + r.report.verdict, + if status == CompilationStatus::Cancelled { + Verdict::Cancelled + } else { + Verdict::CompilationError + } + ); + } + let (m, b, d, mut p) = fixture("python313"); + p.fail_cleanup = true; + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.failure_code, Some(FailureCode::CleanupFailed)); + assert!(!r.report.cleanup_complete); + assert_eq!(p.runs, 0); + } + #[test] + fn trusted_outcome_precedes_comparison_and_text_cannot_fake_memory_limit() { + for (kind, verdict) in [ + (ExecutionKind::Exited { code: 1 }, Verdict::RuntimeError), + (ExecutionKind::MemoryLimit, Verdict::MemoryLimitExceeded), + (ExecutionKind::TimeLimit, Verdict::TimeLimitExceeded), + (ExecutionKind::OutputLimit, Verdict::OutputLimitExceeded), + ] { + let (m, b, d, mut p) = fixture("java21"); + for o in &mut p.observations { + o.kind = kind; + o.stdout = b"OutOfMemoryError".to_vec(); + } + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.verdict, verdict); + assert_eq!(p.expected_reads, 0); + } + } + #[test] + fn job_deadline_and_cancel_skip_tail_without_fabricated_grades() { + let (m, b, d, mut p) = fixture("python313"); + p.advance_us = m.job_wall_time_us; + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.failure_code, Some(FailureCode::JobDeadline)); + assert_eq!(p.runs, 1); + assert_eq!(p.expected_reads, 0); + let (m, b, d, mut p) = fixture("java21"); + p.cancel_after = 1; + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.verdict, Verdict::Cancelled); + assert_eq!(r.skipped.len(), m.cases.len() - 1); + assert_eq!(p.runs, 1); + } +} diff --git a/judge/crates/sandbox/src/artifacts.rs b/judge/crates/sandbox/src/artifacts.rs index e621956..e4ec5d4 100644 --- a/judge/crates/sandbox/src/artifacts.rs +++ b/judge/crates/sandbox/src/artifacts.rs @@ -236,6 +236,7 @@ impl OutputPlan { let paths = if self.java_classes { let mut paths = Vec::new(); let mut count = 0; + let mut unsupported = false; // 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()))? { @@ -251,10 +252,36 @@ impl OutputPlan { continue; } if !kf_contracts::judging_v2::class_name(&name) { - return Err(refused("class export path/type")); + // javac accepts packages and Unicode class names outside + // this protocol's flat ASCII class set. Suppress the whole + // batch so Worker reports unsupported entry as user CE. + // Links/special files remain infrastructure refusals and + // no unrecognized output is ever opened for reading. + let fd = rustix::fs::openat( + work, + name.as_str(), + OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let st = rustix::fs::fstat(&fd)?; + if st.st_dev != rustix::fs::fstat(work)?.st_dev + || !matches!( + FileType::from_raw_mode(st.st_mode), + FileType::RegularFile | FileType::Directory + ) + || (FileType::from_raw_mode(st.st_mode) == FileType::RegularFile + && st.st_nlink != 1) + { + return Err(refused("class export path/type")); + } + unsupported = true; + continue; } paths.push(name); } + if unsupported { + return Ok(Vec::new()); + } paths.sort(); paths } else { @@ -406,8 +433,33 @@ pub struct StagePublication { dev: u64, ino: u64, files: Vec, + staged_bytes: u64, removed: bool, } +const MAX_STAGE_FILES: usize = kf_contracts::judging_v2::MAX_CLASS_FILES + 1; +const MAX_STAGE_JOURNAL_BYTES: u64 = 65_536; +fn indexed_program(name: &str) -> bool { + name.strip_prefix("program-").is_some_and(|index| { + index.len() == 3 + && index.bytes().all(|b| b.is_ascii_digit()) + && index + .parse::() + .is_ok_and(|i| i < kf_contracts::judging_v2::MAX_CLASS_FILES) + }) +} +fn stage_entries_valid(names: &[&str]) -> bool { + let indexed = names.iter().any(|n| indexed_program(n)); + let mut unique = std::collections::HashSet::new(); + names.len() <= if indexed { MAX_STAGE_FILES } else { 2 } + && names.iter().all(|n| { + unique.insert(*n) + && if indexed { + *n == "stdin" || indexed_program(n) + } else { + matches!(*n, "source" | "program" | "stdin") + } + }) +} #[derive(serde::Serialize, serde::Deserialize)] #[serde(deny_unknown_fields)] struct StageEntry { @@ -462,6 +514,7 @@ impl StagePublication { dev: st.st_dev, ino: st.st_ino, files: vec![], + staged_bytes: 0, removed: false, }; let header = StageOwner { @@ -537,16 +590,16 @@ impl StagePublication { || j.st_mode & 0o7777 != 0o600 || j.st_nlink != 1 || j.st_size < 0 - || j.st_size > 2048 + || j.st_size as u64 > MAX_STAGE_JOURNAL_BYTES || FileType::from_raw_mode(j.st_mode) != FileType::RegularFile { return Err(refused("untrusted stage journal")); } let mut bytes = String::new(); readonly_handle(&journal)? - .take(2049) + .take(MAX_STAGE_JOURNAL_BYTES + 1) .read_to_string(&mut bytes)?; - if bytes.len() > 2048 || !bytes.ends_with('\n') { + if bytes.len() as u64 > MAX_STAGE_JOURNAL_BYTES || !bytes.ends_with('\n') { return Err(refused("stage journal framing")); } let mut lines = bytes.lines(); @@ -572,13 +625,7 @@ impl StagePublication { .map_err(|_| refused("stage entry journal")) }) .collect::>>()?; - let mut unique = std::collections::HashSet::new(); - if entries.len() > 2 - || entries.iter().any(|e| { - !matches!(e.name.as_str(), "source" | "program" | "stdin") - || !unique.insert(e.name.clone()) - }) - { + if !stage_entries_valid(&entries.iter().map(|e| e.name.as_str()).collect::>()) { return Err(refused("stage entry set")); } let mut value = Self { @@ -588,6 +635,7 @@ impl StagePublication { dev: st.st_dev, ino: st.st_ino, files: entries, + staged_bytes: 0, removed: false, }; value.cleanup()?; @@ -599,9 +647,12 @@ impl StagePublication { &self.root } pub fn put(&mut self, name: &str, reader: &mut impl Read, bytes: u64) -> io::Result<()> { - if !matches!(name, "source" | "program" | "stdin") - || self.files.len() >= 2 + let mut names: Vec<_> = self.files.iter().map(|e| e.name.as_str()).collect(); + names.push(name); + if !stage_entries_valid(&names) || bytes > MAX_SNAPSHOT_BYTES + || (indexed_program(name) && bytes == 0) + || self.staged_bytes.saturating_add(bytes) > MAX_SNAPSHOT_BYTES { return Err(refused("fixed staging entry/budget")); } @@ -628,6 +679,7 @@ impl StagePublication { file.write_all(b"\n")?; file.sync_all()?; self.files.push(entry); + self.staged_bytes += bytes; let mut file = File::from(fd); let copied = io::copy(&mut reader.take(bytes + 1), &mut file)?; if copied != bytes { @@ -650,7 +702,7 @@ impl StagePublication { use rustix::fd::AsRawFd; let actual: Vec<_> = std::fs::read_dir(format!("/proc/self/fd/{}", self.root.fd.as_raw_fd()))? - .take(5) + .take(MAX_STAGE_FILES + 2) .map(|e| Ok(e?.file_name())) .collect::>()?; if actual.len() != self.files.len() + 1 @@ -710,3 +762,33 @@ impl Drop for StagePublication { let _ = self.cleanup(); } } + +#[cfg(test)] +mod stage_policy_tests { + use super::*; + #[test] + fn bounded_indexed_batch_and_legacy_names_are_separate() { + assert!(stage_entries_valid(&["source"])); + assert!(stage_entries_valid(&["program", "stdin"])); + assert!(!stage_entries_valid(&["source", "program", "stdin"])); + let mut names: Vec<_> = (0..256).map(|i| format!("program-{i:03}")).collect(); + names.push("stdin".into()); + assert!(stage_entries_valid( + &names.iter().map(String::as_str).collect::>() + )); + for name in [ + "program-256", + "program-00", + "program-0000", + "program-../a", + "Main.class", + ".owner", + "expected", + ] { + assert!(!stage_entries_valid(&[name])); + } + assert!(!stage_entries_valid(&["program-000", "source"])); + assert!(!stage_entries_valid(&["program-000", "program"])); + assert!(!stage_entries_valid(&["program-000", "program-000"])); + } +} diff --git a/judge/crates/sandbox/src/image.rs b/judge/crates/sandbox/src/image.rs index 293eb0a..312c7e3 100644 --- a/judge/crates/sandbox/src/image.rs +++ b/judge/crates/sandbox/src/image.rs @@ -46,6 +46,23 @@ fn refused() -> io::Error { "image catalog/integrity refused", ) } +fn catalog_path(kind: ImageKind, name: &str) -> bool { + if kind != ImageKind::GnuCxx { + return crate::spec::path(name, "image").is_ok(); + } + // GNU's immutable standard-library directories include c++ and libstdc++. + // This operator catalog rule never changes caller artifact path validation. + !name.is_empty() + && name.len() <= 4096 + && name.split('/').all(|p| { + !p.is_empty() + && p != "." + && p != ".." + && p.len() <= 255 + && p.bytes() + .all(|b| b.is_ascii_alphanumeric() || b"_.-+".contains(&b)) + }) +} pub fn catalog_digest(data: &[u8]) -> Digest { format!("{:x}", Sha256::digest(data)) .try_into() @@ -81,7 +98,9 @@ impl ImageCatalog { .iter() .chain(self.files.iter().map(|v| &v.path)) { - crate::spec::path(name, "image").map_err(|_| refused())?; + if !catalog_path(self.image_kind, name) { + return Err(refused()); + } if name == "manifest.json" || name .split('/') @@ -290,3 +309,25 @@ fn walk<'a>( } Ok(()) } + +#[cfg(test)] +mod path_tests { + use super::*; + #[test] + fn trusted_gnu_library_paths_do_not_widen_request_or_managed_image_paths() { + for p in ["usr/include/c++/16.2.1/vector", "usr/lib/libstdc++.so.6"] { + assert!(catalog_path(ImageKind::GnuCxx, p)); + assert!(!catalog_path(ImageKind::CPython313, p)); + assert!(crate::spec::path(p, "artifact").is_err()); + } + for p in [ + "/usr/lib", + "usr/../lib", + "usr//lib", + "usr/./lib", + "usr/lib\0", + ] { + assert!(!catalog_path(ImageKind::GnuCxx, p)); + } + } +} diff --git a/judge/crates/sandbox/src/launcher/protocol.rs b/judge/crates/sandbox/src/launcher/protocol.rs index 1dda022..80db733 100644 --- a/judge/crates/sandbox/src/launcher/protocol.rs +++ b/judge/crates/sandbox/src/launcher/protocol.rs @@ -244,10 +244,22 @@ impl ControlChannel { } pub(crate) fn execution_deadline(&mut self, wall_us: u64) -> io::Result<()> { + self.runtime_deadline(wall_us, Duration::ZERO) + } + + /// The ordinary RPC starts before catalog verification and isolation setup. + /// Its fixed preparation allowance is separate from the student's measured + /// execution wall limit. The factory's job/cancellation watcher stays active. + pub(crate) fn request_deadline(&mut self, wall_us: u64) -> io::Result<()> { + self.runtime_deadline(wall_us, Duration::from_secs(30)) + } + + fn runtime_deadline(&mut self, wall_us: u64, preparation: Duration) -> io::Result<()> { if wall_us == 0 || wall_us > 120_000_000 || self.poisoned || !self.pending.is_empty() { return Err(invalid("execution deadline")); } - self.deadline = Instant::now() + Duration::from_micros(wall_us) + Duration::from_secs(5); + self.deadline = + Instant::now() + preparation + Duration::from_micros(wall_us) + Duration::from_secs(5); Ok(()) } @@ -579,6 +591,24 @@ mod tests { ) } + #[test] + fn request_preparation_allowance_is_fixed_and_execution_limit_is_separate() { + let (mut request, mut execution) = pair(); + let start = Instant::now(); + request.request_deadline(3_000_000).unwrap(); + execution.execution_deadline(3_000_000).unwrap(); + assert!(request.deadline.duration_since(start) >= Duration::from_secs(38)); + assert!(request.deadline.duration_since(start) < Duration::from_secs(39)); + assert!(execution.deadline.duration_since(start) >= Duration::from_secs(8)); + assert!(execution.deadline.duration_since(start) < Duration::from_secs(9)); + request.request_deadline(120_000_000).unwrap(); + assert!(request.deadline.duration_since(start) < Duration::from_secs(156)); + assert!(request.request_deadline(0).is_err()); + assert!(request.request_deadline(120_000_001).is_err()); + request.poisoned = true; + assert!(request.request_deadline(3_000_000).is_err()); + } + #[test] fn descriptor_transport_preserves_identity_and_cloexec() { let (mut a, mut b) = pair(); diff --git a/judge/crates/sandbox/src/rpc.rs b/judge/crates/sandbox/src/rpc.rs index 271f5d7..8f71e92 100644 --- a/judge/crates/sandbox/src/rpc.rs +++ b/judge/crates/sandbox/src/rpc.rs @@ -117,7 +117,7 @@ pub(crate) fn execute_channel( ) -> io::Result { c.send(&Request { spec })?; c.send_fd(publication)?; - c.execution_deadline(validated.spec().limits.wall_time_us)?; + c.request_deadline(validated.spec().limits.wall_time_us)?; let report = match c.receive::()? { Reply::Completed { report } => *report, Reply::Rejected { @@ -229,7 +229,7 @@ pub(crate) fn serve_stream( } let artifacts = ArtifactRoot::from_worker_directory(c.receive_fd()?, ids.worker_uid)?; let peer = c.duplicate_stream()?; - c.execution_deadline(request.spec.limits.wall_time_us)?; + c.request_deadline(request.spec.limits.wall_time_us)?; match execution::execute( config, request.spec, diff --git a/judge/crates/sandbox/tests/indexed_staging_vm.rs b/judge/crates/sandbox/tests/indexed_staging_vm.rs new file mode 100644 index 0000000..eeb559f --- /dev/null +++ b/judge/crates/sandbox/tests/indexed_staging_vm.rs @@ -0,0 +1,142 @@ +#![cfg(all(target_os = "linux", feature = "vm-acceptance"))] +//! Indexed Worker publications, actual seals/recovery/foreign preservation in VM only. +#[allow(dead_code)] +mod support; +use kf_sandbox::artifacts::StagePublication; +use std::{ + fs, + io::Read, + os::unix::{ + fs::{MetadataExt, PermissionsExt, chown}, + process::CommandExt, + }, + path::PathBuf, + process::Command, +}; + +fn child(name: &str, path: &std::path::Path) -> std::process::Output { + Command::new(std::env::current_exe().unwrap()) + .env_clear() + .env("KF_SANDBOX_INDEPENDENT_VM", "1") + .env("KF_INDEXED_STAGE_DIR", path) + .uid(1000) + .gid(1000) + .args([ + "--ignored", + "--exact", + name, + "--test-threads=1", + "--nocapture", + ]) + .output() + .unwrap() +} + +#[test] +#[ignore = "ordinary peer subprocess in capped independent VM"] +fn leave_indexed_publication_for_recovery() { + support::require_vm(); + assert_eq!(rustix::process::geteuid().as_raw(), 1000); + let path = PathBuf::from(std::env::var_os("KF_INDEXED_STAGE_DIR").unwrap()); + let mut p = StagePublication::create(&path, "dead-owner").unwrap(); + for i in 0..256 { + p.put(&format!("program-{i:03}"), &mut b"class".as_slice(), 5) + .unwrap(); + } + p.put("stdin", &mut b"".as_slice(), 0).unwrap(); + assert!(p.put("program-256", &mut b"class".as_slice(), 5).is_err()); + assert!(p.put("program", &mut b"class".as_slice(), 5).is_err()); + assert!(p.put("program-001", &mut b"class".as_slice(), 5).is_err()); + assert!(fs::metadata(path.join("dead-owner/.owner")).unwrap().len() < 65536); + std::mem::forget(p); + println!("KF_INDEXED_DEAD_OWNER=1"); +} + +#[test] +#[ignore = "ordinary peer subprocess in capped independent VM"] +fn ordinary_indexed_staging_client() { + support::require_vm(); + assert_eq!(rustix::process::geteuid().as_raw(), 1000); + let path = PathBuf::from(std::env::var_os("KF_INDEXED_STAGE_DIR").unwrap()); + let mut p = StagePublication::create(&path, "sealed").unwrap(); + p.put("program-000", &mut b"\0checked\xff".as_slice(), 9) + .unwrap(); + assert!(p.put("program-001", &mut b"".as_slice(), 0).is_err()); + for name in [ + "../foreign", + "program-1", + "program-256", + "source", + "program", + ] { + assert!(p.put(name, &mut b"x".as_slice(), 1).is_err()); + } + assert!( + p.put("program-001", &mut b"x".as_slice(), 64 * 1024 * 1024 + 1) + .is_err() + ); + let snapshot = p.root().snapshot_artifact("program-000", 9).unwrap(); + fs::write( + path.join("sealed/program-000"), + b"overwritten backing bytes", + ) + .unwrap(); + let mut bytes = vec![]; + snapshot.reader().unwrap().read_to_end(&mut bytes).unwrap(); + assert_eq!(bytes, b"\0checked\xff"); + let seals = rustix::fs::fcntl_get_seals(&snapshot).unwrap(); + assert!(seals.contains( + rustix::fs::SealFlags::WRITE + | rustix::fs::SealFlags::GROW + | rustix::fs::SealFlags::SHRINK + | rustix::fs::SealFlags::SEAL + )); + assert_eq!(StagePublication::recover(&path).unwrap(), 0); + p.cleanup().unwrap(); + let owner = child("leave_indexed_publication_for_recovery", &path); + assert!(owner.status.success(), "{owner:?}"); + assert!(String::from_utf8_lossy(&owner.stdout).contains("KF_INDEXED_DEAD_OWNER=1")); + let foreign = path.join("dead-owner/foreign"); + fs::write(&foreign, b"foreign canary").unwrap(); + let inode = fs::metadata(&foreign).unwrap().ino(); + assert!(StagePublication::recover(&path).is_err()); + assert_eq!(fs::metadata(&foreign).unwrap().ino(), inode); + assert_eq!(fs::read(&foreign).unwrap(), b"foreign canary"); + fs::remove_file(foreign).unwrap(); + assert_eq!(StagePublication::recover(&path).unwrap(), 1); + assert_eq!(fs::read_dir(&path).unwrap().count(), 0); + let mut p = StagePublication::create(&path, "replaced").unwrap(); + p.put("program-000", &mut b"original".as_slice(), 8) + .unwrap(); + let file = path.join("replaced/program-000"); + fs::rename(&file, path.join("original-retained")).unwrap(); + fs::write(&file, b"different inode").unwrap(); + let inode = fs::metadata(&file).unwrap().ino(); + assert!(p.cleanup().is_err()); + assert_eq!(fs::metadata(&file).unwrap().ino(), inode); + assert_eq!(fs::read(&file).unwrap(), b"different inode"); + fs::remove_file(&file).unwrap(); + fs::rename(path.join("original-retained"), &file).unwrap(); + p.cleanup().unwrap(); + assert_eq!(fs::read_dir(&path).unwrap().count(), 0); + println!( + "KF_INDEXED_STAGING_PROOF files=256 stdin=1 bounds=1 seals=1 live_owner=1 dead_owner_recovered=1 foreign_inode_preserved=1 replaced_inode_preserved=1 leftovers=0" + ); +} + +#[test] +#[ignore = "actual ordinary indexed staging only inside independent capped no-NIC/no-share VM"] +fn indexed_staging_recovers_only_owned_inodes_and_preserves_sealed_bytes() { + support::require_vm(); + let dir = PathBuf::from(format!("/kf/indexed-staging-{}", std::process::id())); + fs::create_dir(&dir).unwrap(); + chown(&dir, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(&dir, fs::Permissions::from_mode(0o700)).unwrap(); + let output = child("ordinary_indexed_staging_client", &dir); + assert!(output.status.success(), "{output:?}"); + assert!(String::from_utf8_lossy(&output.stdout).contains("KF_INDEXED_STAGING_PROOF")); + assert_eq!(fs::read_dir(&dir).unwrap().count(), 0); + fs::remove_dir(dir).unwrap(); + print!("{}", String::from_utf8_lossy(&output.stdout)); + println!("KF_INDEXED_STAGING_STATUS=0"); +} diff --git a/judge/crates/sandbox/tests/managed_pipeline_vm.rs b/judge/crates/sandbox/tests/managed_pipeline_vm.rs new file mode 100644 index 0000000..db03265 --- /dev/null +++ b/judge/crates/sandbox/tests/managed_pipeline_vm.rs @@ -0,0 +1,509 @@ +#![cfg(all(target_os = "linux", feature = "vm-acceptance"))] +//! Real ordinary Worker/Engine and default Root, only in the independent VM. +#[allow(dead_code)] +mod support; +use kf_contracts::{judging as v1, judging_v2 as v2}; +use kf_sandbox::{ + factory::FactoryClient, + image::{ImageCatalog, catalog_digest}, + runtime::{ImageKind, RegisteredImage}, + spec::{ExecutionProfile, ResourceLimits}, +}; +use sha2::{Digest as _, Sha256}; +use std::{ + fs, + os::unix::{ + fs::{PermissionsExt, chown}, + process::CommandExt, + }, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + time::{Duration, Instant}, +}; +fn store(path: &Path, data: &[u8]) { + fs::write(path, data).unwrap(); + chown(path, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(path, fs::Permissions::from_mode(0o600)).unwrap(); +} +fn id(n: u32) -> String { + format!("aaaaaaaa-aaaa-4aaa-8aaa-{n:012}") +} +fn budget(profile: ExecutionProfile) -> v1::RunLimits { + let mut b = ResourceLimits::for_profile(profile); + if profile.prepares_program() { + b.cpu_time_us = 30_000_000; + b.wall_time_us = 60_000_000; + } else if matches!( + profile, + ExecutionProfile::JavaRun | ExecutionProfile::PythonRun + ) { + b.cpu_time_us = 5_000_000; + b.wall_time_us = 10_000_000; + } + serde_json::from_value(serde_json::to_value(b).unwrap()).unwrap() +} +struct Fixture { + root: support::Root, + dir: PathBuf, + process: Child, + next: u32, +} +impl Fixture { + fn new() -> 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 = 100000; + root.config.per_task_ceiling.cpu_time_us = 30_000_000; + root.config.per_task_ceiling.wall_time_us = 60_000_000; + let mut capabilities = Vec::new(); + for (name, kind, languages) in [ + ( + "python313", + ImageKind::LinuxCPython313X86_64V1, + vec![v2::Language::Python313], + ), + ( + "java21", + ImageKind::LinuxOpenJdk21X86_64V1, + vec![v2::Language::Java21], + ), + ( + "gnu-v2", + ImageKind::LinuxGnuCxxX86_64V1, + vec![v2::Language::C17, v2::Language::Cpp20], + ), + ] { + let image_root = format!("/var/lib/kf-judge/images/{name}"); + let raw = fs::read(format!("{image_root}/manifest.json")).unwrap(); + let c: ImageCatalog = serde_json::from_slice(&raw).unwrap(); + let toolchain = v2::RuntimeIdentity { + toolchain_id: c.toolchain_id, + image_kind: c.image_kind, + image_revision: c.image_revision, + image_digest: catalog_digest(&raw), + runtime_version: c.runtime_version, + }; + for language in languages { + let (prepare, run) = match language { + v2::Language::C17 => (ExecutionProfile::CompileC, ExecutionProfile::Native), + v2::Language::Cpp20 => (ExecutionProfile::CompileCpp, ExecutionProfile::Native), + v2::Language::Python313 => { + (ExecutionProfile::PythonCheck, ExecutionProfile::PythonRun) + } + v2::Language::Java21 => { + (ExecutionProfile::JavaCompile, ExecutionProfile::JavaRun) + } + }; + capabilities.push(v2::RuntimeCapability { + language, + toolchain: toolchain.clone(), + compile_ceiling: budget(prepare), + run_ceiling: budget(run), + }); + } + root.config.runtime_images.push(RegisteredImage { + image_root, + image_kind: kind, + runtime_identity: toolchain, + per_task_ceiling: root.config.per_task_ceiling.clone(), + }); + } + root.config.clone().validate().unwrap(); + let dir = PathBuf::from(format!("/kf/managed-pipeline-{}", std::process::id())); + fs::create_dir(&dir).unwrap(); + fs::set_permissions(&dir, fs::Permissions::from_mode(0o711)).unwrap(); + for name in ["artifacts", "staging"] { + let p = dir.join(name); + fs::create_dir(&p).unwrap(); + chown(&p, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(&p, fs::Permissions::from_mode(0o700)).unwrap(); + } + let admission = v2::AdmissionPolicy { + schema_version: v2::SchemaVersion, + max_concurrent_jobs: 1, + prefetch: 1, + max_retries: 2, + max_job_wall_time_us: 180_000_000, + max_package_bytes: v1::MAX_PACKAGE_BYTES, + capabilities, + }; + admission.validate().unwrap(); + store( + &dir.join("admission.json"), + &serde_json::to_vec(&admission).unwrap(), + ); + fs::write( + dir.join("root.json"), + serde_json::to_vec(&root.config).unwrap(), + ) + .unwrap(); + fs::set_permissions(dir.join("root.json"), fs::Permissions::from_mode(0o600)).unwrap(); + let mut process = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap()) + .env_clear() + .args(["factory", "--config"]) + .arg(dir.join("root.json")) + .args(["--lifetime-seconds", "1200"]) + .stdout(Stdio::null()) + .stderr(fs::File::create(dir.join("root-diagnostics.log")).unwrap()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(90); + while !fs::metadata("/var/lib/kf-judge/factory/control.sock") + .is_ok_and(|m| m.permissions().mode() & 0o777 == 0o660) + { + assert!( + process.try_wait().unwrap().is_none(), + "factory startup refused" + ); + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(10)); + } + Self { + root, + dir, + process, + next: 1000, + } + } + fn artifact(&self, data: &[u8]) -> v1::ArtifactReference { + let digest: v1::Digest = format!("{:x}", Sha256::digest(data)).try_into().unwrap(); + store(&self.dir.join("artifacts").join(digest.as_str()), data); + v1::ArtifactReference { + digest, + bytes: data.len() as u64, + } + } + fn execute( + &mut self, + language: v2::Language, + source: &[u8], + points: &[(&[u8], &[u8])], + verdict: v1::Verdict, + legacy: bool, + ) -> serde_json::Value { + self.next += 10; + let mut m: serde_json::Value = serde_json::from_str(include_str!( + "../../../../tests/fixtures/contracts-v2/judge-manifest.valid-python313.json" + )) + .unwrap(); + let n = self.next; + let binding = v1::LeaseBinding { + evaluation_id: id(n).try_into().unwrap(), + attempt_id: id(n + 1).try_into().unwrap(), + worker_id: id(8).try_into().unwrap(), + fencing_version: 1, + }; + let admission: v2::AdmissionPolicy = + serde_json::from_slice(&fs::read(self.dir.join("admission.json")).unwrap()).unwrap(); + let cap = admission + .capabilities + .iter() + .find(|c| c.language == language) + .unwrap(); + m["evaluation_id"] = serde_json::json!(id(n)); + m["submission_id"] = serde_json::json!(id(n + 2)); + m["source"] = serde_json::to_value(self.artifact(source)).unwrap(); + m["language"] = serde_json::to_value(language).unwrap(); + m["toolchain"] = serde_json::to_value(&cap.toolchain).unwrap(); + m["compile_limits"] = serde_json::to_value(&cap.compile_ceiling).unwrap(); + m["run_limits"] = serde_json::to_value(&cap.run_ceiling).unwrap(); + m["job_wall_time_us"] = serde_json::json!(180_000_000); + m["cases"] = serde_json::json!( + points + .iter() + .enumerate() + .map(|(i, (input, expected))| v1::TestCase { + case_id: id(n + 3 + i as u32).try_into().unwrap(), + input: self.artifact(input), + expected: self.artifact(expected) + }) + .collect::>() + ); + 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}); + v1::manifest_bytes(&serde_json::from_value(m.clone()).unwrap()).unwrap() + } else { + v2::manifest_bytes(&serde_json::from_value(m.clone()).unwrap()).unwrap() + }; + let digest = format!("{:x}", Sha256::digest(canonical)); + let mut job = serde_json::json!({"manifest":m,"binding":binding,"manifest_digest":digest}); + if !legacy { + job["attempt_number"] = serde_json::json!(1); + } + store( + &self.dir.join("job.json"), + &serde_json::to_vec(&job).unwrap(), + ); + let mut command = Command::new(std::env::var_os("KF_WORKER_TEST_BIN").unwrap()); + command + .env_clear() + .uid(1000) + .gid(1000) + .arg(if legacy { + "judge-submission" + } else { + "judge-submission-v2" + }) + .args(["--socket", "/var/lib/kf-judge/factory/control.sock"]) + .arg("--job") + .arg(self.dir.join("job.json")) + .arg("--artifacts") + .arg(self.dir.join("artifacts")) + .arg("--staging") + .arg(self.dir.join("staging")); + if !legacy { + command + .arg("--admission") + .arg(self.dir.join("admission.json")); + } + let output = command.output().unwrap(); + if !output.stderr.is_empty() { + println!( + "KF_PIPELINE_WORKER_DIAGNOSTIC {}", + String::from_utf8_lossy(&output.stderr) + ); + } + assert!( + output.status.success(), + "worker failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + let actual: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + if actual["result"]["report"]["verdict"] != serde_json::to_value(verdict).unwrap() { + use std::io::Read; + let mut diagnostic = Vec::new(); + fs::File::open(self.dir.join("root-diagnostics.log")) + .unwrap() + .take(16_384) + .read_to_end(&mut diagnostic) + .unwrap(); + println!( + "KF_PIPELINE_ROOT_DIAGNOSTIC {}", + String::from_utf8_lossy(&diagnostic) + ); + } + assert_eq!( + actual["result"]["report"]["verdict"], + serde_json::to_value(verdict).unwrap(), + "{actual}" + ); + assert_eq!( + actual["result"]["report"]["cleanup_complete"], true, + "{actual}" + ); + assert_eq!( + actual["result"]["report"]["schema_version"], + if legacy { 1 } else { 2 } + ); + let stages = actual["stages"].as_array().unwrap(); + assert!(!stages.is_empty()); + let mut ids = std::collections::HashSet::new(); + for s in stages { + assert!(s["helper_pid"].as_u64().unwrap() > 1); + assert!(ids.insert(s["helper_pid"].as_u64().unwrap())); + 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"] { + assert_eq!( + fs::read_dir(format!("/var/lib/kf-judge/{name}")) + .unwrap() + .count(), + 0 + ); + } + 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}) + ); + actual + } +} +impl Drop for Fixture { + fn drop(&mut self) { + let _ = self.process.kill(); + let _ = self.process.wait(); + for name in ["requests", "socket.json", "control.sock", "lock"] { + let _ = fs::remove_file(Path::new("/var/lib/kf-judge/factory").join(name)); + } + let _ = fs::remove_dir("/var/lib/kf-judge/factory"); + let _ = fs::remove_dir_all(&self.dir); + } +} +#[test] +#[ignore = "ordinary peer subprocess inside the isolated pipeline VM"] +fn ordinary_pipeline_stopped_client() { + assert_eq!(rustix::process::geteuid().as_raw(), 1000); + assert_eq!(rustix::process::getegid().as_raw(), 1000); + assert!( + FactoryClient::new(PathBuf::from("/var/lib/kf-judge/factory/control.sock")) + .stopped() + .unwrap() + ); + println!("KF_PIPELINE_STOPPED=1"); +} +#[test] +#[ignore = "actual ordinary Worker typed Python/Java and old/new GNU only inside capped no-NIC/no-share VM"] +fn ordinary_worker_prepares_once_and_runs_all_four_runtimes() { + let mut f = Fixture::new(); + let points: &[(&[u8], &[u8])] = &[ + (b"", b""), + (b"\0\xff\n", b"\0\xff\n"), + ("你好\n".as_bytes(), "你好\n".as_bytes()), + ]; + let py = br#"import os,sys +assert not os.path.exists('previous') +assert not os.path.exists('expected') +assert not os.path.exists('/kf') +open('previous','w').close() +sys.stdout.buffer.write(sys.stdin.buffer.read()) +"#; + f.execute( + v2::Language::Python313, + py, + points, + v1::Verdict::Accepted, + false, + ); + f.execute( + v2::Language::Python313, + py, + &[(b"x", b"y"), (b"", b"")], + v1::Verdict::WrongAnswer, + false, + ); + f.execute( + v2::Language::Python313, + b"def bad(:\n", + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + f.execute( + v2::Language::Python313, + b"raise MemoryError('fake kernel OOM')\n", + &[(b"", b"")], + v1::Verdict::RuntimeError, + false, + ); + let pyloop = f.execute( + v2::Language::Python313, + b"while True: pass\n", + &[(b"", b"")], + v1::Verdict::TimeLimitExceeded, + false, + ); + assert_eq!( + pyloop["result"]["report"]["compilation"]["status"], + "succeeded" + ); + let java=br#"import java.io.*; public class Main {static int counter=0;static class Inner {} public static void main(String[] args) throws Exception {new Runnable(){public void run(){}}.run();if(++counter!=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();Helper.copy();}} class Helper {static void copy() throws Exception {System.in.transferTo(System.out);}}"#; + let j = f.execute( + v2::Language::Java21, + java, + points, + v1::Verdict::Accepted, + false, + ); + assert_eq!( + j["result"]["report"]["compilation"]["program"]["files"] + .as_array() + .unwrap() + .len(), + 4 + ); + f.execute( + v2::Language::Java21, + java, + &[(b"x", b"y")], + v1::Verdict::WrongAnswer, + false, + ); + for source in [ + &b"public class Main { syntax error }"[..], + &b"public class Main {static {if(System.nanoTime()>0){while(true){}}}}"[..], + &b"package unwanted; public class Main {public static void main(String[] a){}}"[..], + &b"public class Wrong {public static void main(String[] a){}}"[..], + ] { + f.execute( + v2::Language::Java21, + source, + &[(b"", b"")], + v1::Verdict::CompilationError, + false, + ); + } + f.execute(v2::Language::Java21,b"public class Main {public static void main(String[] a){throw new OutOfMemoryError(\"fake kernel OOM\");}}",&[(b"",b"")],v1::Verdict::RuntimeError,false); + gnu_points(&mut f, points); + assert!( + !fs::read_dir(&f.root.path) + .unwrap() + .any(|e| e.unwrap().file_type().unwrap().is_dir()) + ); + println!("KF_MANAGED_PIPELINE_STATUS=0"); +} + +fn gnu_points(f: &mut Fixture, points: &[(&[u8], &[u8])]) { + let c=b"#include \nint main(void){int c;while((c=getchar())!=EOF)putchar(c);return 0;}\n"; + let cpp = b"#include \nint main(){char c;while(std::cin.get(c))std::cout.put(c);}\n"; + f.execute(v2::Language::C17, c, points, v1::Verdict::Accepted, false); + f.execute( + v2::Language::Cpp20, + cpp, + points, + v1::Verdict::Accepted, + false, + ); + f.execute(v2::Language::C17, c, points, v1::Verdict::Accepted, true); + f.execute( + v2::Language::Cpp20, + cpp, + points, + v1::Verdict::Accepted, + true, + ); +} +#[test] +#[ignore = "targeted actual ordinary Worker new/legacy GNU after typed staging changes inside capped VM"] +fn ordinary_worker_runs_versioned_and_legacy_gnu() { + let mut f = Fixture::new(); + let points: &[(&[u8], &[u8])] = &[ + (b"", b""), + (b"\0\xff\n", b"\0\xff\n"), + ("你好\n".as_bytes(), "你好\n".as_bytes()), + ]; + gnu_points(&mut f, points); + assert!( + !fs::read_dir(&f.root.path) + .unwrap() + .any(|e| e.unwrap().file_type().unwrap().is_dir()) + ); + println!("KF_MANAGED_GNU_PIPELINE_STATUS=0"); +} diff --git a/judge/crates/worker/src/consumer.rs b/judge/crates/worker/src/consumer.rs index e17f2fd..cc151f3 100644 --- a/judge/crates/worker/src/consumer.rs +++ b/judge/crates/worker/src/consumer.rs @@ -1,9 +1,16 @@ //! Ordinary transport adapter. AMQP is a bounded wake-up; claim is sole authority. use crate::pipeline::{ArtifactPort, Attempt}; +use crate::{pipeline_v2::RuntimeAttempt, runtime_specs::AuthorizedJob}; use futures_lite::{StreamExt, future::block_on}; -use kf_contracts::judging_versions::{JobManifest as VersionedManifest, NodeAdmission}; +use kf_contracts::judging_v2 as v2; +use kf_contracts::judging_versions::{ + self as versions, JobManifest as VersionedManifest, NodeAdmission, +}; use kf_contracts::{JudgeTaskNotification, SchemaVersion, judging::*}; -use kf_engine::submission::{StageFailure, judge_submission}; +use kf_engine::{ + submission::{StageFailure, judge_submission}, + submission_v2, +}; use kf_sandbox::{artifacts::StagePublication, factory::FactoryClient}; use lapin::{Connection, ConnectionProperties, options::*, types::FieldTable}; use reqwest::blocking::Client; @@ -20,6 +27,37 @@ use std::{ time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; const QUEUE: &str = "kf.judge.evaluations.v1"; +/// Transport projection after strict versioned wire decoding. It never changes +/// the authorized manifest or its canonical bytes. +#[derive(Clone)] +struct Grant { + version: u16, + binding: LeaseBinding, + lease_token: LeaseToken, + attempt_number: u32, + expires_at_unix_ms: u64, + manifest_digest: Digest, +} +impl From for Grant { + fn from(wire: versions::LeaseGrant) -> Self { + macro_rules! fields { + ($g:ident, $version:literal) => { + Self { + version: $version, + binding: $g.binding, + lease_token: $g.lease_token, + attempt_number: $g.attempt_number, + expires_at_unix_ms: $g.expires_at_unix_ms, + manifest_digest: $g.manifest_digest, + } + }; + } + match wire { + versions::LeaseGrant::V1(g) => fields!(g, 1), + versions::LeaseGrant::V2(g) => fields!(g, 2), + } + } +} #[derive(Debug)] struct TransportError { operation: &'static str, @@ -139,7 +177,7 @@ impl Remote { &self, route: &str, body: &T, - grant: Option<&LeaseGrant>, + grant: Option<&Grant>, ) -> io::Result { let mut r = self .client @@ -173,8 +211,9 @@ impl Remote { let v: Vec = serde_json::from_slice(&b) .map_err(|_| transport_error("coordinator_pending", "decode"))?; if v.len() > 16 - || v.iter() - .any(|p| p.case_count == 0 || p.case_count > MAX_CASES) + || v.iter().any(|p| { + p.case_count == 0 || p.case_count > MAX_CASES || !matches!(p.schema_version, 1 | 2) + }) { return Err(error()); } @@ -193,12 +232,7 @@ impl Remote { if p.binding.worker_id != c.worker_id { return Err(error()); } - let r = StoppedReport { - schema_version: SchemaVersion, - kind: StoppedKind::Value, - binding: p.binding, - cleanup_complete: clean, - }; + let r = p.stopped(clean)?; if self.post("/internal/stopped", &r, None)?.status().as_u16() != 204 { return Err(error()); } @@ -209,13 +243,37 @@ impl Remote { #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct Pending { + #[serde(default = "legacy_version")] + schema_version: u16, binding: LeaseBinding, case_count: usize, } +fn legacy_version() -> u16 { + 1 +} +impl Pending { + fn stopped(self, clean: bool) -> io::Result { + match self.schema_version { + 1 => Ok(versions::StoppedReport::V1(StoppedReport { + schema_version: SchemaVersion, + kind: StoppedKind::Value, + binding: self.binding, + cleanup_complete: clean, + })), + 2 => Ok(versions::StoppedReport::V2(v2::StoppedReport { + schema_version: v2::SchemaVersion, + kind: StoppedKind::Value, + binding: self.binding, + cleanup_complete: clean, + })), + _ => Err(error()), + } + } +} struct RemoteArtifacts { client: Client, base: String, - grant: LeaseGrant, + grant: Grant, } impl RemoteArtifacts { fn headers(&self, r: reqwest::blocking::RequestBuilder) -> reqwest::blocking::RequestBuilder { @@ -234,25 +292,28 @@ impl RemoteArtifacts { self.grant.binding.attempt_id.as_uuid() ) } - fn manifest(&self) -> io::Result { + fn manifest(&self) -> io::Result { let response = self .headers(self.client.get(self.url("/manifest"))) .send() .map_err(|_| error())?; - let m: JobManifest = - decode_wire(&Remote::bytes(response, MAX_WIRE_BYTES)?).map_err(|_| error())?; - let digest = format!( - "{:x}", - Sha256::digest(manifest_bytes(&m).map_err(|_| error())?) - ); - if digest != self.grant.manifest_digest.as_str() - || m.evaluation_id != self.grant.binding.evaluation_id - { - return Err(error()); - } - Ok(m) + authorized_manifest(&Remote::bytes(response, MAX_WIRE_BYTES)?, &self.grant) } } +fn authorized_manifest(raw: &[u8], grant: &Grant) -> io::Result { + let m: VersionedManifest = decode_wire(raw).map_err(|_| error())?; + let digest = format!( + "{:x}", + Sha256::digest(m.canonical_bytes().map_err(|_| error())?) + ); + if digest != grant.manifest_digest.as_str() + || m.evaluation_id() != grant.binding.evaluation_id + || m.version() != grant.version + { + return Err(error()); + } + Ok(m) +} impl ArtifactPort for RemoteArtifacts { fn read(&mut self, reference: &ArtifactReference) -> io::Result> { let r = self @@ -282,7 +343,7 @@ fn lease_deadline(ms: u64) -> io::Result { .ok_or_else(error)?; Ok(Instant::now() + Duration::from_millis(remaining)) } -fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Result<()> { +fn execute(c: &Config, r: &Remote, g: Grant, stop: &AtomicBool) -> io::Result<()> { if g.binding.worker_id != c.worker_id { return Err(error()); } @@ -293,7 +354,7 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu }; let m = artifacts.manifest()?; c.admission - .admit(&VersionedManifest::V1(m.clone()), g.attempt_number) + .admit(&m, g.attempt_number) .map_err(|_| error())?; let cancel = AtomicBool::new(false); let done = AtomicBool::new(false); @@ -329,23 +390,9 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu }); let result = (|| { let mut sequence = 0u64; - let mut adapter = Attempt::new( - artifacts, - c.socket.clone(), - c.staging.clone(), - m.job_wall_time_us, - &cancel, - )? - .with_control(|stage, index| { + let mut control = |stage, index| { sequence += 1; - let report = StageReport { - schema_version: SchemaVersion, - kind: StageKind::Value, - binding: g.binding.clone(), - sequence, - stage, - case_index: index, - }; + let report = stage_report(&m, g.binding.clone(), sequence, stage, index); for _ in 0..2 { match r.post("/internal/stage", &report, Some(&g)) { Ok(response) if response.status().as_u16() == 204 => return Ok(()), @@ -359,22 +406,64 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu code: FailureCode::LeaseLost, cleanup_complete: true, }) - }); - let mut result = judge_submission( - &mut adapter, - &m, - g.binding.clone(), - g.manifest_digest.clone(), - ) - .map_err(|_| error())?; - transport_cancellation(&mut result.report); - result - .report + }; + let mut report = match &m { + VersionedManifest::V1(manifest) => { + let mut adapter = Attempt::new( + artifacts, + c.socket.clone(), + c.staging.clone(), + manifest.job_wall_time_us, + &cancel, + )? + .with_control(&mut control); + versions::JudgementReport::V1( + judge_submission( + &mut adapter, + manifest, + g.binding.clone(), + g.manifest_digest.clone(), + ) + .map_err(|_| error())? + .report, + ) + } + VersionedManifest::V2(manifest) => { + let authorized = AuthorizedJob::admit( + manifest.clone(), + g.binding.clone(), + g.manifest_digest.clone(), + g.attempt_number, + &c.admission, + ) + .map_err(|_| error())?; + let mut adapter = RuntimeAttempt::new( + authorized, + artifacts, + c.socket.clone(), + c.staging.clone(), + &cancel, + )? + .with_control(&mut control); + versions::JudgementReport::V2( + submission_v2::judge_submission( + &mut adapter, + manifest, + g.binding.clone(), + g.manifest_digest.clone(), + ) + .map_err(|_| error())? + .report, + ) + } + }; + transport_cancellation_versioned(&mut report); + report .validate_for_manifest(&m, &g.manifest_digest) .map_err(|_| error())?; // Same immutable report only; a transport ambiguity never becomes success. for _ in 0..3 { - match r.post("/internal/report", &result.report, Some(&g)) { + match r.post("/internal/report", &report, Some(&g)) { Ok(response) if response.status().as_u16() == 204 => return Ok(()), Ok(response) if response.status().as_u16() == 409 => break, _ => {} @@ -388,6 +477,47 @@ fn execute(c: &Config, r: &Remote, g: LeaseGrant, stop: &AtomicBool) -> io::Resu result }) } +fn stage_report( + m: &VersionedManifest, + binding: LeaseBinding, + sequence: u64, + stage: Stage, + case_index: Option, +) -> versions::StageReport { + match m { + VersionedManifest::V1(_) => versions::StageReport::V1(StageReport { + schema_version: SchemaVersion, + kind: StageKind::Value, + binding, + sequence, + stage, + case_index, + }), + VersionedManifest::V2(manifest) => versions::StageReport::V2(v2::StageReport { + schema_version: v2::SchemaVersion, + kind: StageKind::Value, + binding, + sequence, + stage, + case_index, + preparation_kind: manifest.language.contract().preparation_kind, + }), + } +} +fn transport_cancellation_versioned(report: &mut versions::JudgementReport) { + match report { + versions::JudgementReport::V1(r) => transport_cancellation(r), + versions::JudgementReport::V2(r) => { + if r.verdict == Verdict::Cancelled { + r.verdict = Verdict::SystemError; + r.failure_code = Some(FailureCode::LeaseLost); + if r.compilation.status == CompilationStatus::Cancelled { + r.compilation.status = CompilationStatus::SystemError; + } + } + } + } +} // A task stopped by service shutdown/transport loss is an infrastructure failure. // User cancellation is authoritative in Coordinator and completes via actual stop proof. fn transport_cancellation(report: &mut JudgementReport) { @@ -486,9 +616,9 @@ async fn consume(c: &Config, r: &Remote, stop: &AtomicBool) -> io::Result<()> { let grant = r.post("/internal/claim", ¬ice, None)?; match grant.status().as_u16() { 200 => { - let g: LeaseGrant = + let g: versions::LeaseGrant = decode_wire(&Remote::bytes(grant, MAX_WIRE_BYTES)?).map_err(|_| error())?; - let result = execute(c, r, g, stop); + let result = execute(c, r, g.into(), stop); if result.is_err() && !r.reconcile(c)? { return Err(error()); } @@ -539,6 +669,86 @@ pub fn run(c: Config, stop: Arc) -> io::Result<()> { mod tests { use super::*; #[test] + fn versioned_manifest_is_bound_to_lease_generation_digest_and_identity() { + for (raw, version) in [ + ( + &include_bytes!("../../../../tests/fixtures/contracts/judge-manifest.valid.json")[..], + 1, + ), + ( + &include_bytes!( + "../../../../tests/fixtures/contracts-v2/judge-manifest.valid-python313.json" + )[..], + 2, + ), + ] { + let m: VersionedManifest = decode_wire(raw).unwrap(); + let base: versions::LeaseGrant = decode_wire(include_bytes!( + "../../../../tests/fixtures/contracts-v2/judge-lease.valid.json" + )) + .unwrap(); + let mut g: Grant = base.into(); + g.version = version; + g.binding.evaluation_id = m.evaluation_id(); + g.manifest_digest = format!("{:x}", Sha256::digest(m.canonical_bytes().unwrap())) + .try_into() + .unwrap(); + assert_eq!(authorized_manifest(raw, &g).unwrap(), m); + g.version = 3 - version; + assert!(authorized_manifest(raw, &g).is_err()); + g.version = version; + g.manifest_digest = "f".repeat(64).try_into().unwrap(); + assert!(authorized_manifest(raw, &g).is_err()); + } + } + #[test] + fn preparation_stage_and_pending_stop_retain_exact_protocol_generation() { + let raw = include_bytes!( + "../../../../tests/fixtures/contracts-v2/judge-manifest.valid-python313.json" + ); + let m: VersionedManifest = decode_wire(raw).unwrap(); + let g: versions::LeaseGrant = decode_wire(include_bytes!( + "../../../../tests/fixtures/contracts-v2/judge-lease.valid.json" + )) + .unwrap(); + let mut g: Grant = g.into(); + g.binding.evaluation_id = m.evaluation_id(); + let report = stage_report(&m, g.binding.clone(), 1, Stage::Compiling, None); + report.validate_for_manifest(&m).unwrap(); + let value = serde_json::to_value(report).unwrap(); + assert_eq!(value["schema_version"], 2); + assert_eq!(value["preparation_kind"], "syntax_check"); + for version in [1, 2, 3] { + let mut pending = serde_json::json!({"binding":g.binding,"case_count":2}); + if version != 1 { + pending["schema_version"] = serde_json::json!(version); + } + let p: Pending = serde_json::from_value(pending).unwrap(); + if version == 3 { + assert!(p.stopped(true).is_err()); + continue; + } + let stopped = serde_json::to_value(p.stopped(true).unwrap()).unwrap(); + assert_eq!(stopped["schema_version"], version); + assert_eq!( + stopped, + serde_json::json!({"schema_version":version,"kind":"judge.stopped","binding":g.binding,"cleanup_complete":true}) + ); + } + let mut legacy: VersionedManifest = decode_wire(include_bytes!( + "../../../../tests/fixtures/contracts/judge-manifest.valid.json" + )) + .unwrap(); + if let VersionedManifest::V1(m) = &mut legacy { + m.evaluation_id = g.binding.evaluation_id; + } + let report = stage_report(&legacy, g.binding, 1, Stage::Compiling, None); + let value = serde_json::to_value(&report).unwrap(); + assert_eq!(value["schema_version"], 1); + assert!(value.get("preparation_kind").is_none()); + report.validate_for_manifest(&legacy).unwrap(); + } + #[test] fn amqp_numeric_authority_and_default_vhost_are_preserved() { for (raw, host, vhost) in [ ("amqp://fixture:temporary@127.0.0.2:5672/", "127.0.0.2", "/"), diff --git a/judge/crates/worker/src/java_entry.rs b/judge/crates/worker/src/java_entry.rs new file mode 100644 index 0000000..0b65945 --- /dev/null +++ b/judge/crates/worker/src/java_entry.rs @@ -0,0 +1,212 @@ +//! Bounded metadata inspection of sealed javac output, never class loading. +//! Format: https://docs.oracle.com/javase/specs/jvms/se21/html/jvms-4.html +//! Bytecode verification and execution remain inside the isolated JVM. +use std::collections::{BTreeMap, HashSet}; +struct Reader<'a>(&'a [u8]); +impl<'a> Reader<'a> { + fn take(&mut self, n: usize) -> Result<&'a [u8], &'static str> { + let value = self.0.get(..n).ok_or("truncated class metadata")?; + self.0 = &self.0[n..]; + Ok(value) + } + fn u8(&mut self) -> Result { + Ok(self.take(1)?[0]) + } + fn u16(&mut self) -> Result { + Ok(u16::from_be_bytes( + self.take(2)?.try_into().expect("two bytes"), + )) + } + fn u32(&mut self) -> Result { + Ok(u32::from_be_bytes( + self.take(4)?.try_into().expect("four bytes"), + )) + } + fn attributes(&mut self, pool: &[Constant<'a>]) -> Result<(), &'static str> { + for _ in 0..self.u16()? { + utf8(pool, self.u16()?)?; + let length = self.u32()? as usize; + self.take(length)?; + } + Ok(()) + } +} +enum Constant<'a> { + Other, + Utf8(&'a [u8]), + Class(u16), +} +fn utf8<'a>(pool: &[Constant<'a>], index: u16) -> Result<&'a [u8], &'static str> { + match pool.get(usize::from(index)) { + Some(Constant::Utf8(bytes)) if index != 0 => Ok(bytes), + _ => Err("invalid class UTF8 reference"), + } +} +fn class<'a>(pool: &[Constant<'a>], index: u16) -> Result<&'a [u8], &'static str> { + match pool.get(usize::from(index)) { + Some(Constant::Class(name)) if index != 0 => utf8(pool, *name), + _ => Err("invalid class name reference"), + } +} +struct Info<'a> { + parent: &'a [u8], + main: bool, +} +fn parse<'a>(path: &str, data: &'a [u8]) -> Result, &'static str> { + if data.len() as u64 > kf_contracts::judging::MAX_PACKAGE_BYTES { + return Err("class metadata byte budget"); + } + let mut r = Reader(data); + if r.u32()? != 0xcafebabe || r.u16()? != 0 || r.u16()? != 65 { + return Err("class differs from fixed javac release"); + } + let count = usize::from(r.u16()?); + if count < 2 { + return Err("empty class constant pool"); + } + let mut pool = vec![Constant::Other]; + while pool.len() < count { + let c = match r.u8()? { + 1 => { + let n = usize::from(r.u16()?); + Constant::Utf8(r.take(n)?) + } + 7 => Constant::Class(r.u16()?), + 3 | 4 | 9 | 10 | 11 | 12 | 17 | 18 => { + r.take(4)?; + Constant::Other + } + 8 | 16 | 19 | 20 => { + r.take(2)?; + Constant::Other + } + 15 => { + r.take(3)?; + Constant::Other + } + 5 | 6 => { + r.take(8)?; + if pool.len() + 1 >= count { + return Err("wide constant slot budget"); + } + pool.push(Constant::Other); + Constant::Other + } + _ => return Err("unknown class constant"), + }; + pool.push(c); + } + r.u16()?; + if class(&pool, r.u16()?)? + != path + .strip_suffix(".class") + .ok_or("class file path")? + .as_bytes() + { + return Err("class identity differs from sealed file"); + } + let parent = class(&pool, r.u16()?)?; + for _ in 0..r.u16()? { + class(&pool, r.u16()?)?; + } + for _ in 0..r.u16()? { + r.u16()?; + utf8(&pool, r.u16()?)?; + utf8(&pool, r.u16()?)?; + r.attributes(&pool)?; + } + let mut main = false; + for _ in 0..r.u16()? { + let flags = r.u16()?; + let name = utf8(&pool, r.u16()?)?; + let descriptor = utf8(&pool, r.u16()?)?; + main |= + flags & 0x0009 == 0x0009 && name == b"main" && descriptor == b"([Ljava/lang/String;)V"; + r.attributes(&pool)?; + } + r.attributes(&pool)?; + if !r.0.is_empty() { + return Err("trailing class metadata"); + } + Ok(Info { parent, main }) +} +pub(crate) fn has_main(files: &[(String, Vec)]) -> Result { + if files.is_empty() + || files.len() > kf_contracts::judging_v2::MAX_CLASS_FILES + || files.iter().map(|(_, d)| d.len() as u64).sum::() + > kf_contracts::judging::MAX_PACKAGE_BYTES + { + return Err("class collection metadata budget"); + } + let mut classes = BTreeMap::new(); + for (path, bytes) in files { + if !kf_contracts::judging_v2::class_name(path) { + return Err("unsupported class path"); + } + if classes + .insert( + path.strip_suffix(".class").expect("class path").as_bytes(), + parse(path, bytes)?, + ) + .is_some() + { + return Err("duplicate class identity"); + } + } + let mut name = &b"Main"[..]; + let mut seen = HashSet::new(); + while let Some(info) = classes.get(name) { + if !seen.insert(name) { + return Err("class superclass cycle"); + } + if info.main { + return Ok(true); + } + name = info.parent; + } + Ok(false) +} + +#[cfg(test)] +mod tests { + use super::*; + fn minimal(flags: u16, descriptor: &[u8]) -> Vec { + let mut b = vec![0xca, 0xfe, 0xba, 0xbe, 0, 0, 0, 65, 0, 7]; + for name in [ + &b"Main"[..], + &b"java/lang/Object"[..], + &b"main"[..], + descriptor, + ] { + b.push(1); + b.extend_from_slice(&(name.len() as u16).to_be_bytes()); + b.extend_from_slice(name); + } + b.extend_from_slice(&[7, 0, 1, 7, 0, 2, 0, 0x21, 0, 5, 0, 6, 0, 0, 0, 0, 0, 1]); + b.extend_from_slice(&flags.to_be_bytes()); + b.extend_from_slice(&[0, 3, 0, 4, 0, 0, 0, 0]); + b + } + #[test] + fn entry_signature_and_truncation_are_checked_without_loading_classes() { + let yes = minimal(9, b"([Ljava/lang/String;)V"); + assert!(has_main(&[("Main.class".into(), yes.clone())]).unwrap()); + for flags in [0, 1, 8] { + assert!( + !has_main(&[( + "Main.class".into(), + minimal(flags, b"([Ljava/lang/String;)V") + )]) + .unwrap() + ); + } + assert!(!has_main(&[("Main.class".into(), minimal(9, b"()V"))]).unwrap()); + for n in 0..yes.len() { + assert!(parse("Main.class", &yes[..n]).is_err()); + } + assert!(parse("Other.class", &yes).is_err()); + let mut extra = yes; + extra.push(0); + assert!(parse("Main.class", &extra).is_err()); + } +} diff --git a/judge/crates/worker/src/lib.rs b/judge/crates/worker/src/lib.rs index 340a9a8..90b4671 100644 --- a/judge/crates/worker/src/lib.rs +++ b/judge/crates/worker/src/lib.rs @@ -116,5 +116,14 @@ pub mod native { #[cfg(target_os = "linux")] pub mod pipeline; +#[cfg(target_os = "linux")] +pub mod runtime_specs; + +#[cfg(target_os = "linux")] +pub mod pipeline_v2; + +#[cfg(target_os = "linux")] +mod java_entry; + #[cfg(target_os = "linux")] pub mod consumer; diff --git a/judge/crates/worker/src/main.rs b/judge/crates/worker/src/main.rs index 2951798..edc0976 100644 --- a/judge/crates/worker/src/main.rs +++ b/judge/crates/worker/src/main.rs @@ -32,6 +32,18 @@ enum Command { #[arg(long)] staging: PathBuf, }, + JudgeSubmissionV2 { + #[arg(long)] + socket: PathBuf, + #[arg(long)] + job: PathBuf, + #[arg(long)] + artifacts: PathBuf, + #[arg(long)] + staging: PathBuf, + #[arg(long)] + admission: PathBuf, + }, JudgeOnce { #[arg(long)] socket: PathBuf, @@ -114,6 +126,36 @@ async fn main() -> ExitCode { ExitCode::from(2) } } + Some(Command::JudgeSubmissionV2 { + socket, + job, + artifacts, + staging, + admission, + }) => { + let result = (|| { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + let m = std::fs::symlink_metadata(&admission)?; + if !m.is_file() || m.nlink() != 1 || m.permissions().mode() & 0o7777 != 0o600 { + return Err(std::io::Error::other("operator admission file refused")); + } + let policy = kf_contracts::judging::decode_wire( + &kf_sandbox::execution::read_bounded_file(&admission, 32768)?, + ) + .map_err(std::io::Error::other)?; + kf_worker::pipeline_v2::judge_local(&socket, &job, &artifacts, &staging, &policy) + })(); + match result { + Ok(result) => { + println!("{}", serde_json::json!(result)); + ExitCode::SUCCESS + } + Err(_) => { + eprintln!("submission_failed"); + ExitCode::from(2) + } + } + } Some(Command::JudgeOnce { socket, job, diff --git a/judge/crates/worker/src/pipeline.rs b/judge/crates/worker/src/pipeline.rs index 4356b06..f3fa2c9 100644 --- a/judge/crates/worker/src/pipeline.rs +++ b/judge/crates/worker/src/pipeline.rs @@ -61,12 +61,12 @@ impl ArtifactPort for LocalArtifacts { Ok(data) } } -fn digest(data: &[u8]) -> Digest { +pub(crate) fn digest(data: &[u8]) -> Digest { format!("{:x}", Sha256::digest(data)) .try_into() .expect("SHA256 representation") } -fn verify(r: &ArtifactReference, data: &[u8]) -> io::Result<()> { +pub(crate) fn verify(r: &ArtifactReference, data: &[u8]) -> io::Result<()> { if data.len() as u64 != r.bytes || digest(data) != r.digest { return Err(io::Error::other("artifact digest mismatch")); } @@ -78,7 +78,7 @@ fn stage_error(code: FailureCode, cleanup_complete: bool) -> StageFailure { cleanup_complete, } } -fn limits(r: &RunLimits) -> ResourceLimits { +pub(crate) fn limits(r: &RunLimits) -> ResourceLimits { ResourceLimits { cpu_time_us: r.cpu_time_us, wall_time_us: r.wall_time_us, @@ -92,13 +92,13 @@ fn limits(r: &RunLimits) -> ResourceLimits { workdir_bytes: r.workdir_bytes, } } -fn metric(m: &Measurement) -> Option { +pub(crate) fn metric(m: &Measurement) -> Option { match m { Measurement::Measured { value } => Some(*value), Measurement::Unavailable { .. } => None, } } -fn kind(o: ExecutionOutcome) -> ExecutionKind { +pub(crate) fn kind(o: ExecutionOutcome) -> ExecutionKind { match o { ExecutionOutcome::Exited { code } => ExecutionKind::Exited { code }, ExecutionOutcome::Signaled { .. } => ExecutionKind::Signaled, @@ -210,7 +210,10 @@ impl<'a, A: ArtifactPort> Attempt<'a, A> { self.control = Some(Box::new(control)); self } - fn execute( + pub(crate) fn staging_path(&self) -> &Path { + &self.staging + } + pub(crate) fn execute( &mut self, spec: ExecutionSpec, publication: &mut StagePublication, diff --git a/judge/crates/worker/src/pipeline_v2.rs b/judge/crates/worker/src/pipeline_v2.rs new file mode 100644 index 0000000..873fb97 --- /dev/null +++ b/judge/crates/worker/src/pipeline_v2.rs @@ -0,0 +1,414 @@ +//! Ordinary multi-runtime adapter. User source is only checked/compiled/run by +//! isolated Root helpers; Worker handles bounded bytes and sealed capabilities. +use crate::{ + pipeline::{ArtifactPort, Attempt, LocalArtifacts, StageEvidence, kind, metric, verify}, + runtime_specs::AuthorizedJob, +}; +use kf_contracts::{judging as v1, judging_v2 as v2, judging_versions::NodeAdmission}; +use kf_engine::{Observation, submission as old, submission_v2 as engine}; +use kf_sandbox::{ + artifacts::{SnapshotFile, StagePublication}, + outcome::ExecutionOutcome, +}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use std::{ + io::{self, Read}, + path::{Path, PathBuf}, + sync::atomic::AtomicBool, +}; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Job { + pub manifest: v2::JobManifest, + pub binding: v1::LeaseBinding, + pub manifest_digest: v1::Digest, + pub attempt_number: u32, +} +#[derive(Serialize)] +pub struct Judged { + pub result: engine::SubmissionResult, + pub stages: Vec, +} +struct SealedProgram { + metadata: v2::CompiledProgram, + files: Vec, +} +pub struct RuntimeAttempt<'a, A: ArtifactPort> { + authorized: AuthorizedJob, + io: Attempt<'a, A>, + program: Option, +} +fn failure(code: v1::FailureCode, cleanup_complete: bool) -> old::StageFailure { + old::StageFailure { + code, + cleanup_complete, + } +} +fn artifact_error(_: io::Error) -> old::StageFailure { + failure(v1::FailureCode::ArtifactUnavailable, true) +} +fn snapshot_reference(file: &SnapshotFile) -> Result { + let mut h = Sha256::new(); + let mut reader = file.reader().map_err(artifact_error)?; + let mut buffer = [0u8; 16384]; + let mut total = 0u64; + loop { + let n = reader.read(&mut buffer).map_err(artifact_error)?; + if n == 0 { + break; + } + total += n as u64; + if total > file.bytes() { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + h.update(&buffer[..n]); + } + if total != file.bytes() { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + Ok(v1::ArtifactReference { + digest: format!("{:x}", h.finalize()).try_into().expect("SHA256"), + bytes: total, + }) +} +fn diagnostics(result: &kf_sandbox::rpc::Executed) -> Result<(String, bool), old::StageFailure> { + let mut raw = Vec::new(); + result + .stderr + .reader() + .map_err(artifact_error)? + .take(65537) + .read_to_end(&mut raw) + .map_err(artifact_error)?; + let mut text = String::from_utf8_lossy(&raw).into_owned(); + let mut truncated = result.report.stderr.truncated || text.len() > 65536; + if text.len() > 65536 { + let mut end = 65536; + while !text.is_char_boundary(end) { + end -= 1; + } + text.truncate(end); + truncated = true; + } + Ok((text, truncated)) +} +impl<'a, A: ArtifactPort> RuntimeAttempt<'a, A> { + pub fn new( + authorized: AuthorizedJob, + artifacts: A, + socket: PathBuf, + staging: PathBuf, + cancellation: &'a AtomicBool, + ) -> io::Result { + let io = Attempt::new( + artifacts, + socket, + staging, + authorized.manifest().job_wall_time_us, + cancellation, + )?; + Ok(Self { + authorized, + io, + program: None, + }) + } + pub fn with_control( + mut self, + control: impl FnMut(v1::Stage, Option) -> Result<(), old::StageFailure> + 'a, + ) -> Self { + self.io = self.io.with_control(control); + self + } + pub fn into_stages(self) -> Vec { + self.io.stages + } +} +impl engine::PreparationPort for RuntimeAttempt<'_, A> { + fn prepare( + &mut self, + m: &v2::JobManifest, + binding: &v1::LeaseBinding, + run_id: &str, + ) -> Result { + if m != self.authorized.manifest() + || binding != self.authorized.binding() + || self.program.is_some() + { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + let source = self.io.artifacts.read(&m.source).map_err(artifact_error)?; + verify(&m.source, &source).map_err(artifact_error)?; + let mut publication = + StagePublication::create(self.io.staging_path(), run_id).map_err(artifact_error)?; + publication + .put("source", &mut source.as_slice(), m.source.bytes) + .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, false))?; + let checked_source = if m.language == v2::Language::Python313 { + let file = publication + .root() + .snapshot_artifact("source", m.source.bytes) + .map_err(artifact_error)?; + if snapshot_reference(&file)? != m.source { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + Some(file) + } else { + None + }; + drop(source); + let spec = self + .authorized + .preparation_spec(run_id.into()) + .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, true))?; + let validated = spec + .clone() + .validate(&kf_sandbox::spec::SpecPolicy::default()) + .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, true))?; + let result = self.io.execute(spec, &mut publication)?; + let mut status = match result + .report + .classify(&validated) + .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, false))? + { + ExecutionOutcome::Exited { code: 0 } => v1::CompilationStatus::Succeeded, + ExecutionOutcome::Exited { .. } | ExecutionOutcome::Signaled { .. } => { + v1::CompilationStatus::CompilationError + } + ExecutionOutcome::LimitExceeded { .. } => v1::CompilationStatus::CompilationLimit, + ExecutionOutcome::Cancelled => v1::CompilationStatus::Cancelled, + _ => v1::CompilationStatus::SystemError, + }; + let (mut text, truncated) = diagnostics(&result)?; + let cpu_time_us = metric(&result.report.usage.cpu_time_us); + let memory_peak_bytes = metric(&result.report.usage.memory_peak_bytes); + let mut files = Vec::new(); + let mut sealed = Vec::new(); + if status == v1::CompilationStatus::Succeeded { + if let Some(source) = checked_source { + if !result.outputs.is_empty() { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + files.push(v2::ProgramFile { + path: "main.py".into(), + artifact: m.source.clone(), + }); + sealed.push(source); + } else { + for output in result.outputs { + files.push(v2::ProgramFile { + path: output.path, + artifact: snapshot_reference(&output.file)?, + }); + sealed.push(output.file); + } + } + if m.language == v2::Language::Java21 { + // No reflective loading or initialization outside the JVM. A + // successfully compiled unsupported entry is a user CE. + let mut metadata = Vec::with_capacity(files.len()); + for (info, file) in files.iter().zip(&sealed) { + let mut bytes = Vec::new(); + file.reader() + .map_err(artifact_error)? + .take(info.artifact.bytes + 1) + .read_to_end(&mut bytes) + .map_err(artifact_error)?; + verify(&info.artifact, &bytes).map_err(artifact_error)?; + metadata.push((info.path.clone(), bytes)); + } + let has_main = if metadata.is_empty() { + false + } else { + crate::java_entry::has_main(&metadata) + .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, true))? + }; + if !has_main { + status = v1::CompilationStatus::CompilationError; + text="Java submission requires default-package Main with public static void main(String[]).".into(); + } + } + } + let program = if status == v1::CompilationStatus::Succeeded { + let metadata = v2::CompiledProgram { + kind: m.language.contract().program_kind, + binding: binding.clone(), + source: m.source.clone(), + toolchain: m.toolchain.clone(), + files, + }; + self.authorized + .validate_program(&metadata) + .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, true))?; + self.program = Some(SealedProgram { + metadata: metadata.clone(), + files: sealed, + }); + Some(metadata) + } else { + None + }; + Ok(engine::Compiled { + report: v2::CompilationReport { + status, + program, + diagnostics: text.try_into().expect("bounded diagnostics"), + diagnostics_truncated: truncated, + cpu_time_us, + memory_peak_bytes, + }, + cleanup_complete: true, + failure: (status == v1::CompilationStatus::SystemError) + .then_some(v1::FailureCode::SandboxUnavailable), + }) + } +} +impl engine::RunPort for RuntimeAttempt<'_, A> { + fn run( + &mut self, + program: &v2::CompiledProgram, + case: &v1::TestCase, + budget: &v1::RunLimits, + run_id: &str, + ) -> Result { + let index = self + .authorized + .manifest() + .cases + .iter() + .position(|c| c == case) + .ok_or_else(|| failure(v1::FailureCode::ArtifactUnavailable, true))?; + if budget != &self.authorized.manifest().run_limits { + return Err(failure(v1::FailureCode::SandboxUnavailable, true)); + } + let spec = self + .authorized + .run_spec(program, index, run_id.into()) + .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, true))?; + let stored = self + .program + .as_ref() + .filter(|s| &s.metadata == program) + .ok_or_else(|| failure(v1::FailureCode::ArtifactUnavailable, true))?; + if stored.files.len() != program.files.len() { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + let input = self + .io + .artifacts + .read(&case.input) + .map_err(artifact_error)?; + verify(&case.input, &input).map_err(artifact_error)?; + let mut publication = + StagePublication::create(self.io.staging_path(), run_id).map_err(artifact_error)?; + for ((info, file), artifact) in program.files.iter().zip(&stored.files).zip(&spec.inputs) { + if snapshot_reference(file)? != info.artifact { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + publication + .put( + &artifact.artifact_id, + &mut file.reader().map_err(artifact_error)?, + info.artifact.bytes, + ) + .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, false))?; + } + publication + .put("stdin", &mut input.as_slice(), case.input.bytes) + .map_err(|_| failure(v1::FailureCode::ArtifactUnavailable, false))?; + drop(input); + let validated = spec + .clone() + .validate(&kf_sandbox::spec::SpecPolicy::default()) + .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, true))?; + let result = self.io.execute(spec, &mut publication)?; + let outcome = result + .report + .classify(&validated) + .map_err(|_| failure(v1::FailureCode::SandboxUnavailable, false))?; + let mut stdout = Vec::new(); + result + .stdout + .reader() + .map_err(artifact_error)? + .take(budget.stdout_bytes + 1) + .read_to_end(&mut stdout) + .map_err(artifact_error)?; + if stdout.len() as u64 > budget.stdout_bytes { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + Ok(Observation { + kind: kind(outcome), + stdout, + cpu_time_us: metric(&result.report.usage.cpu_time_us), + memory_peak_bytes: metric(&result.report.usage.memory_peak_bytes), + }) + } +} +impl old::ExpectedPort for RuntimeAttempt<'_, A> { + fn expected(&mut self, case: &v1::TestCase) -> Result, old::StageFailure> { + if !self.authorized.manifest().cases.contains(case) { + return Err(failure(v1::FailureCode::ArtifactUnavailable, true)); + } + let bytes = self + .io + .artifacts + .read(&case.expected) + .map_err(artifact_error)?; + verify(&case.expected, &bytes).map_err(artifact_error)?; + Ok(bytes) + } +} +impl old::ControlPort for RuntimeAttempt<'_, A> { + fn elapsed_us(&self) -> u64 { + old::ControlPort::elapsed_us(&self.io) + } + fn cancelled(&self) -> bool { + old::ControlPort::cancelled(&self.io) + } + fn stage(&mut self, stage: v1::Stage, index: Option) -> Result<(), old::StageFailure> { + old::ControlPort::stage(&mut self.io, stage, index) + } +} +pub fn judge_local( + socket: &Path, + job: &Path, + artifacts: &Path, + staging: &Path, + admission: &NodeAdmission, +) -> io::Result { + let job: Job = v1::decode_wire(&kf_sandbox::execution::read_bounded_file( + job, + v1::MAX_WIRE_BYTES as u64, + )?) + .map_err(io::Error::other)?; + let authorized = AuthorizedJob::admit( + job.manifest.clone(), + job.binding.clone(), + job.manifest_digest.clone(), + job.attempt_number, + admission, + ) + .map_err(io::Error::other)?; + let cancellation = AtomicBool::new(false); + let mut attempt = RuntimeAttempt::new( + authorized, + LocalArtifacts::open(artifacts)?, + socket.into(), + staging.into(), + &cancellation, + )?; + let result = engine::judge_submission( + &mut attempt, + &job.manifest, + job.binding, + job.manifest_digest, + ) + .map_err(io::Error::other)?; + Ok(Judged { + result, + stages: attempt.into_stages(), + }) +} diff --git a/judge/crates/worker/src/runtime_specs.rs b/judge/crates/worker/src/runtime_specs.rs new file mode 100644 index 0000000..3c4de44 --- /dev/null +++ b/judge/crates/worker/src/runtime_specs.rs @@ -0,0 +1,310 @@ +//! Ordinary v2 admission and closed specifications. This module neither reads +//! user files nor executes source; every stage still goes through Root gates. +use crate::pipeline::{compiler_spec, limits}; +use kf_contracts::{judging as v1, judging_v2 as v2, judging_versions as versions}; +use kf_sandbox::spec::{ExecutionProfile as P, ExecutionSpec, InputArtifact, SpecPolicy}; +use sha2::{Digest as _, Sha256}; + +/// A canonical manifest and lease checked against the operator's capability +/// registry. Private fields prevent widening budgets after admission. +pub struct AuthorizedJob { + manifest: v2::JobManifest, + binding: v1::LeaseBinding, + manifest_digest: v1::Digest, +} +impl AuthorizedJob { + pub fn admit( + manifest: v2::JobManifest, + binding: v1::LeaseBinding, + manifest_digest: v1::Digest, + attempt_number: u32, + policy: &versions::NodeAdmission, + ) -> Result { + binding.validate()?; + if binding.evaluation_id != manifest.evaluation_id { + return Err("lease differs from authorized manifest"); + } + policy.admit(&versions::JobManifest::V2(manifest.clone()), attempt_number)?; + let canonical = v2::manifest_bytes(&manifest)?; + if format!("{:x}", Sha256::digest(canonical)) != manifest_digest.as_str() { + return Err("canonical manifest digest differs from lease"); + } + Ok(Self { + manifest, + binding, + manifest_digest, + }) + } + pub fn manifest(&self) -> &v2::JobManifest { + &self.manifest + } + pub fn binding(&self) -> &v1::LeaseBinding { + &self.binding + } + pub fn manifest_digest(&self) -> &v1::Digest { + &self.manifest_digest + } + pub fn preparation_spec(&self, run_id: String) -> Result { + let m = &self.manifest; + let spec = match m.language { + v2::Language::C17 | v2::Language::Cpp20 => { + let mut spec = compiler_spec( + if m.language == v2::Language::C17 { + v1::Language::C17 + } else { + v1::Language::Cpp20 + }, + "source".into(), + run_id, + &m.compile_limits, + ); + spec.runtime_identity = Some(m.toolchain.clone()); + spec + } + v2::Language::Python313 | v2::Language::Java21 => { + let profile = if m.language == v2::Language::Python313 { + P::PythonCheck + } else { + P::JavaCompile + }; + let budget = limits(&m.compile_limits); + ExecutionSpec { + schema_version: kf_contracts::SchemaVersion, + runtime_identity: Some(m.toolchain.clone()), + run_id, + profile, + argv: kf_sandbox::profiles::argv(profile, &budget)?, + env: Default::default(), + cwd: ".".into(), + inputs: vec![InputArtifact { + artifact_id: "source".into(), + path: m.language.contract().source_name.into(), + }], + stdin_artifact: None, + outputs: vec![], + limits: budget, + } + } + }; + checked(spec) + } + /// Metadata is only accepted for this source/runtime/attempt. The adapter + /// must separately verify every sealed file's byte count and digest. + pub fn validate_program(&self, program: &v2::CompiledProgram) -> v1::Validation { + program.validate()?; + let m = &self.manifest; + if program.binding != self.binding + || program.source != m.source + || program.toolchain != m.toolchain + || program.kind != m.language.contract().program_kind + { + return Err("program differs from authorized source, runtime or lease"); + } + if program.kind != v2::ProgramKind::PythonSource + && (program + .files + .iter() + .any(|f| f.artifact.bytes > m.compile_limits.file_bytes) + || program.bytes() > m.compile_limits.workdir_bytes) + { + return Err("program exceeds preparation output budget"); + } + Ok(()) + } + pub fn run_spec( + &self, + program: &v2::CompiledProgram, + case_index: usize, + run_id: String, + ) -> Result { + self.validate_program(program)?; + if case_index >= self.manifest.cases.len() { + return Err("test point is outside authorized manifest"); + } + let profile = match self.manifest.language { + v2::Language::C17 | v2::Language::Cpp20 => P::Native, + v2::Language::Python313 => P::PythonRun, + v2::Language::Java21 => P::JavaRun, + }; + let budget = limits(&self.manifest.run_limits); + checked(ExecutionSpec { + schema_version: kf_contracts::SchemaVersion, + runtime_identity: Some(self.manifest.toolchain.clone()), + run_id, + profile, + argv: if profile == P::Native { + vec!["/work/main".into()] + } else { + kf_sandbox::profiles::argv(profile, &budget)? + }, + env: Default::default(), + cwd: ".".into(), + inputs: program + .files + .iter() + .enumerate() + .map(|(i, f)| InputArtifact { + artifact_id: program_artifact_id(i), + path: f.path.clone(), + }) + .collect(), + stdin_artifact: Some("stdin".into()), + outputs: vec![], + limits: budget, + }) + } +} +pub(crate) fn program_artifact_id(index: usize) -> String { + format!("program-{index:03}") +} +fn checked(spec: ExecutionSpec) -> Result { + spec.clone() + .validate(&SpecPolicy::default()) + .map_err(|_| "invalid fixed runtime specification")?; + Ok(spec) +} + +#[cfg(test)] +mod tests { + use super::*; + fn fixture(name: &str) -> T { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../../tests/fixtures/contracts-v2") + .join(format!("{name}.json")); + v1::decode_wire(&std::fs::read(path).unwrap()).unwrap() + } + fn authorized(language: &str) -> (AuthorizedJob, v2::CompiledProgram) { + let m: v2::JobManifest = fixture(&format!("judge-manifest.valid-{language}")); + let report: v2::JudgementReport = fixture(&format!("judge-result.valid-{language}")); + let policy: versions::NodeAdmission = fixture("judge-registry.valid"); + let digest = format!("{:x}", Sha256::digest(v2::manifest_bytes(&m).unwrap())) + .try_into() + .unwrap(); + let job = AuthorizedJob::admit(m, report.binding, digest, 1, &policy).unwrap(); + (job, report.compilation.program.unwrap()) + } + #[test] + fn all_four_languages_use_exact_pins_fixed_commands_and_private_points() { + for (language, prepare, run) in [ + ("c17", P::CompileC, P::Native), + ("cpp20", P::CompileCpp, P::Native), + ("python313", P::PythonCheck, P::PythonRun), + ("java21", P::JavaCompile, P::JavaRun), + ] { + let (j, program) = authorized(language); + let p = j.preparation_spec("prepare".into()).unwrap(); + assert_eq!(p.profile, prepare); + assert_eq!(p.runtime_identity.as_ref(), Some(&j.manifest.toolchain)); + assert!(p.env.is_empty()); + assert!(p.stdin_artifact.is_none()); + assert_eq!(p.inputs[0].path, j.manifest.language.contract().source_name); + for index in 0..j.manifest.cases.len() { + let r = j + .run_spec(&program, index, format!("point-{index}")) + .unwrap(); + assert_eq!(r.profile, run); + assert_eq!(r.runtime_identity, p.runtime_identity); + assert_eq!(r.limits, limits(&j.manifest.run_limits)); + assert!(r.env.is_empty() && r.outputs.is_empty()); + assert_eq!(r.stdin_artifact.as_deref(), Some("stdin")); + assert_eq!(r.inputs.len(), program.files.len()); + assert!( + r.inputs + .iter() + .all(|f| f.artifact_id.starts_with("program-")) + ); + // Expected answers, source paths and manifest host paths are not inputs. + assert_eq!( + r.inputs.iter().map(|f| &f.path).collect::>(), + program.files.iter().map(|f| &f.path).collect::>() + ); + } + assert!( + j.run_spec(&program, j.manifest.cases.len(), "point".into()) + .is_err() + ); + } + } + #[test] + fn lease_manifest_digest_runtime_and_operator_capabilities_are_fenced() { + let (j, _) = authorized("python313"); + let policy: versions::NodeAdmission = fixture("judge-registry.valid"); + let admit = |m, b, d, n| AuthorizedJob::admit(m, b, d, n, &policy); + let mut wrong = j.manifest.clone(); + wrong.toolchain.image_digest = "f".repeat(64).try_into().unwrap(); + assert!(admit(wrong, j.binding.clone(), j.manifest_digest.clone(), 1).is_err()); + let mut b = j.binding.clone(); + b.evaluation_id = "aaaaaaaa-aaaa-4aaa-8aaa-000000001000" + .to_string() + .try_into() + .unwrap(); + assert!(admit(j.manifest.clone(), b, j.manifest_digest.clone(), 1).is_err()); + assert!( + admit( + j.manifest.clone(), + j.binding.clone(), + "f".repeat(64).try_into().unwrap(), + 1 + ) + .is_err() + ); + assert!( + admit( + j.manifest.clone(), + j.binding.clone(), + j.manifest_digest.clone(), + 4 + ) + .is_err() + ); + let mut wrong = j.manifest.clone(); + wrong.run_limits.memory_bytes += 1; + assert!(admit(wrong, j.binding.clone(), j.manifest_digest.clone(), 1).is_err()); + let mut legacy: v2::AdmissionRegistry = fixture("judge-registry.valid"); + legacy.multi_runtime = None; + assert!( + AuthorizedJob::admit( + j.manifest.clone(), + j.binding.clone(), + j.manifest_digest.clone(), + 1, + &versions::NodeAdmission::Registry(legacy) + ) + .is_err() + ); + } + #[test] + fn sealed_program_metadata_cannot_change_source_attempt_type_or_budget() { + for language in ["c17", "python313", "java21"] { + let (j, p) = authorized(language); + let mut bad = p.clone(); + bad.binding.attempt_id = "aaaaaaaa-aaaa-4aaa-8aaa-000000001000" + .to_string() + .try_into() + .unwrap(); + assert!(j.validate_program(&bad).is_err()); + bad = p.clone(); + bad.source.digest = "f".repeat(64).try_into().unwrap(); + assert!(j.validate_program(&bad).is_err()); + bad = p.clone(); + bad.toolchain.image_digest = "f".repeat(64).try_into().unwrap(); + assert!(j.validate_program(&bad).is_err()); + bad = p.clone(); + bad.files[0].path = "../Main.class".into(); + assert!(j.validate_program(&bad).is_err()); + bad = p.clone(); + bad.files[0].artifact.bytes = 0; + assert!(j.validate_program(&bad).is_err()); + if language != "python313" { + bad = p; + bad.files[0].artifact.bytes = j.manifest.compile_limits.file_bytes + 1; + assert!(j.validate_program(&bad).is_err()); + } + } + let (j, mut p) = authorized("java21"); + p.files[0].path = "A$1.class".into(); + j.validate_program(&p).unwrap(); + p.files.push(p.files[0].clone()); + assert!(j.validate_program(&p).is_err()); + } +} diff --git a/scripts/build_gnu_image.py b/scripts/build_gnu_image.py new file mode 100644 index 0000000..16724db --- /dev/null +++ b/scripts/build_gnu_image.py @@ -0,0 +1,181 @@ +"""Offline GNU catalog from an explicitly pinned, verified public package closure. + +No target executable, package hook, submitted source or network installer runs. +Root publication and real execution acceptance remain separate steps. +""" + +import argparse +import hashlib +import json +import re +from pathlib import Path + + +def sha(path): + with path.open("rb") as source: + return hashlib.file_digest(source, "sha256").hexdigest() + + +def canonical(name): + return len(name) <= 4096 and all( + re.fullmatch(r"[A-Za-z0-9_.+\-]{1,255}", p) and p not in (".", "..") + for p in name.split("/") + ) + + +def resolve_verified(root, records, name): + parts = name.split("/") + for _ in range(9): + changed = False + for i in range(len(parts)): + prefix = "/".join(parts[: i + 1]) + record = records.get(prefix) + path = root / prefix + if record and record["kind"] == "symlink": + if not path.is_symlink() or str(path.readlink()) != record["target"]: + raise ValueError("GNU package link differs from verified manifest") + target = record["target"] + expanded = ( + ([] if target.startswith("/") else parts[:i]) + + target.split("/") + + parts[i + 1 :] + ) + parts = [] + for p in expanded: + if p in ("", "."): + continue + if p == "..": + if not parts: + raise ValueError("GNU image-relative link escapes closure") + parts.pop() + else: + parts.append(p) + changed = True + break + if path.is_symlink(): + raise ValueError("unverified GNU ancestor link") + if not changed: + relative = "/".join(parts) + record = records.get(relative) + if record is None or record["kind"] != "file": + raise ValueError("GNU link outside verified regular-file closure") + source = root / relative + source.resolve(strict=True).relative_to(root) + return source, record + raise ValueError("GNU link cycle or depth budget") + + +def build(args): + if sha(args.support_manifest) != args.support_manifest_sha256: + raise ValueError("verified GNU support manifest fingerprint differs") + manifest = json.loads(args.support_manifest.read_bytes()) + packages = {p["name"]: p for p in manifest["packages"]} + expected = { + "gcc": ( + "16.2.1+r23+gd564253eb6c8-1", + "ebac12ceef494c616a36fc6f0f792b365dc59b6ed46ca1a69f138aecc810b6ab", + ), + "glibc": ( + "2.44+r50+g1848099f063e-3", + "071d807f4d3bf6b921f86b9a91e83a47ecc026c5821634efa46b09d890e06785", + ), + "binutils": ("2.47-2", "62efd213e87590f025cf861b646793d3cacb03e7e26187b73d33b00adaf7a977"), + } + for name, (version, digest) in expected.items(): + if (packages[name]["version"], packages[name]["sha256"]) != (version, digest): + raise ValueError("GNU image requires the pinned compiler/libc/linker closure") + records = {f["path"]: f for f in manifest["files"]} + source_root = args.support_dir.resolve(strict=True) + if args.output.exists(): + raise ValueError("refuse to replace an image") + args.output.mkdir(mode=0o700) + directories = {"dev", "proc", "tmp", "work", "usr", "lib64"} + files = [] + + def put(name, source, expected_file): + if not canonical(name) or expected_file["kind"] != "file": + raise ValueError("noncanonical verified GNU file") + if ( + source.stat().st_size != expected_file["bytes"] + or sha(source) != expected_file["sha256"] + ): + raise ValueError("GNU file differs from verified public package bytes") + if len(files) >= 4096 or sum(f["bytes"] for f in files) + source.stat().st_size > 1024**3: + raise ValueError("GNU image exceeds finite catalog budget") + target = args.output / name + target.parent.mkdir(parents=True, exist_ok=True) + # Copy to a fresh regular inode; never publish source links/hard links. + with source.open("rb") as reader, target.open("xb") as writer: + while data := reader.read(1024 * 1024): + writer.write(data) + executable = bool(source.stat().st_mode & 0o111) + target.chmod(0o555 if executable else 0o444) + if sha(target) != expected_file["sha256"]: + raise ValueError("GNU copy fingerprint differs") + files.append( + { + "path": name, + "bytes": target.stat().st_size, + "sha256": sha(target), + "executable": executable, + } + ) + directories.update(p.as_posix() for p in Path(name).parents if p != Path(".")) + + for name, record in sorted(records.items()): + if not ( + name.startswith(("usr/include/", "usr/lib/")) + or name in ("usr/bin/gcc", "usr/bin/g++", "usr/bin/as", "usr/bin/ld") + ): + continue + if record["kind"] == "directory": + continue + source, real = resolve_verified(source_root, records, name) + put(name, source, real) + loader, loader_record = resolve_verified(source_root, records, "usr/lib/ld-linux-x86-64.so.2") + put("lib64/ld-linux-x86-64.so.2", loader, loader_record) + for name in directories: + (args.output / name).mkdir(parents=True, exist_ok=True) + catalog = dict( + format_version=2, + architecture="x86_64_v3", + image_kind="linux_gnu_cxx_x86_64_v1", + image_revision=1, + toolchain_id="gnu-cxx-v1", + runtime_version="16.2.1+r23+gd564253eb6c8", + directories=sorted(directories), + files=sorted(files, key=lambda f: f["path"]), + ) + raw = (json.dumps(catalog, sort_keys=True, separators=(",", ":")) + "\n").encode() + if len(raw) > 1024 * 1024: + raise ValueError("GNU catalog transport budget") + (args.output / "manifest.json").write_bytes(raw) + (args.output / "manifest.json").chmod(0o444) + for path in sorted((p for p in args.output.rglob("*") if p.is_dir()), reverse=True): + path.chmod(0o555) + args.output.chmod(0o555) + print( + json.dumps( + dict( + image_digest=hashlib.sha256(raw).hexdigest(), + files=len(files), + bytes=sum(f["bytes"] for f in files), + support_manifest_sha256=args.support_manifest_sha256, + source_execution=False, + root_publication_and_VM_acceptance=False, + ) + ) + ) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + 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()