From 8a01081cc95482378035425a17935a352c48701f Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Fri, 9 Oct 2026 04:12:22 +0800 Subject: [PATCH] feat(judge): compile once and run isolated points through helper factory --- README.md | 4 +- deploy/README.md | 23 + deploy/kf-sandbox-factory.service | 28 + judge/Cargo.lock | 73 +++ judge/README.md | 31 +- judge/crates/engine/src/submission.rs | 33 +- judge/crates/sandbox/Cargo.toml | 3 + judge/crates/sandbox/src/artifacts.rs | 314 ++++++++++ judge/crates/sandbox/src/execution.rs | 2 +- judge/crates/sandbox/src/factory.rs | 448 ++++++++++++++ judge/crates/sandbox/src/launcher/run.rs | 61 +- judge/crates/sandbox/src/lib.rs | 3 + judge/crates/sandbox/src/main.rs | 42 ++ judge/crates/sandbox/src/rpc.rs | 41 +- judge/crates/sandbox/tests/pipeline_vm.rs | 714 ++++++++++++++++++++++ judge/crates/worker/Cargo.toml | 1 + judge/crates/worker/src/lib.rs | 3 + judge/crates/worker/src/main.rs | 35 ++ judge/crates/worker/src/pipeline.rs | 504 +++++++++++++++ 19 files changed, 2336 insertions(+), 27 deletions(-) create mode 100644 deploy/kf-sandbox-factory.service create mode 100644 judge/crates/sandbox/src/factory.rs create mode 100644 judge/crates/sandbox/tests/pipeline_vm.rs create mode 100644 judge/crates/worker/src/pipeline.rs diff --git a/README.md b/README.md index 6cae75c..6b36692 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,8 @@ 前端使用 Vue 3、TypeScript、Element Plus 和 Vite。 当前版本为 **0.0.1**：沙箱已有受控隔离、C/C++ 编译及 native 执行；平台已实现真实身份、 -不可变题目、工件及有效租约读取基础。完整提交投递、协调器调度、多测试点评测与前端业务 -接入仍在开发，不能将前端预览成绩当作实际结果。 +不可变题目、工件及有效租约读取基础。受控 helper 工厂与本地 C/C++ 编译一次、多测试点评测 +已经接通；完整提交投递、协调器调度与前端业务接入仍在开发，不能将前端预览成绩当作实际结果。 ## 工程划分 diff --git a/deploy/README.md b/deploy/README.md index c91f14e..417025a 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -19,3 +19,26 @@ API 存活检查和前端预览不依赖这些容器，可以先独立启动。 官方镜像说明：[PostgreSQL](https://hub.docker.com/_/postgres)、 [RabbitMQ](https://hub.docker.com/_/rabbitmq)。 + +### 单机 helper 工厂 + +`kf-sandbox factory --config /etc/kf-judge/runtime.json` 是 Root 服务入口； +普通 Worker 使用配置中 `state_root/factory/control.sock` 的普通用户客户端。 +`deploy/kf-sandbox-factory.service` 给出有限服务预算示例，安装前由可信管理员 +核对固定二进制、不可写 GNU 镜像/配置、Worker UID/GID，以及**与服务自身 cgroup +分离**的 Root 所有任务委派根。工厂实际读回服务 cgroup 限额，应用/读回任务总预算； +每次编译和测试点仍须通过已有完整隔离门控。服务文件本身不证明这些条件成立。 + +工厂只接收有界 run ID，无排队：槽位满、重复 ID 或未知字段立即拒绝。 +每次申请使用全新单线程 helper，固定继承已认证的连接与工厂生命线； +额外控制帧、断开或生命线消失会终止整棵任务树。ID 先写入有限持久记录 +（最多 4096 个，跨重启拒绝重放）；容量用尽需要管理员在所有任务停止、已核对 +回收后安排新记录周期，不自动删除记录。清理失败保留身份租约并停止新分配； +重启只根据已记录归属和 inode 回收，未知/外来内容保留并拒绝启动。 + +`kf-worker judge-submission` 是受控本地适配入口，仍不消费队列。 +它验证不可变清单摘要，C17/C++20 仅映射到固定 GNU 命令，编译一次； +成功产物经完整清理后封印并计算 SHA-256，每点分别申请 helper、stdin 与工作区。 +期望值逐点读取，留在比较端，不传给工厂/用户文件系统。 +新增锁定 `sha2 0.10.9` 仅供 Worker 验证输入及编译产物摘要； +Sandbox 同版本仅为独立 VM 测试数据生成的开发依赖，Engine 不依赖加密或 OS 库。 diff --git a/deploy/kf-sandbox-factory.service b/deploy/kf-sandbox-factory.service new file mode 100644 index 0000000..880fd0d --- /dev/null +++ b/deploy/kf-sandbox-factory.service @@ -0,0 +1,28 @@ +[Unit] +Description=KF bounded Linux helper factory +After=local-fs.target + +[Service] +Type=exec +User=root +Group=root +# Trusted operator installs this exact binary/config and separate delegated +# task root. No caller-controlled executable, image or environment is accepted. +ExecStart=/usr/bin/env -i PATH=/usr/bin:/bin LC_ALL=C /usr/lib/kf-judge/kf-sandbox factory --config /etc/kf-judge/runtime.json --lifetime-seconds 3600 +MemoryMax=256M +MemorySwapMax=0 +TasksMax=32 +CPUQuota=50% +LimitFSIZE=64M +RuntimeMaxSec=3700 +TimeoutStopSec=30 +# First stop the factory; guardians cancel live helpers through their normal +# cleanup path. Remaining processes are terminated at the finite stop deadline. +KillMode=mixed +Restart=no +UMask=0077 +# Requires root namespace/mount/cgroup privileges. Do not set NoNewPrivileges or +# an inherited seccomp filter: the existing execution gate intentionally refuses. + +[Install] +WantedBy=multi-user.target diff --git a/judge/Cargo.lock b/judge/Cargo.lock index cf8799e..59f5513 100644 --- a/judge/Cargo.lock +++ b/judge/Cargo.lock @@ -125,6 +125,15 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -189,6 +198,35 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "errno" version = "0.3.14" @@ -241,6 +279,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "heck" version = "0.5.0" @@ -392,6 +440,7 @@ dependencies = [ "seccompiler", "serde", "serde_json", + "sha2", ] [[package]] @@ -404,6 +453,7 @@ dependencies = [ "kf-sandbox", "serde", "serde_json", + "sha2", ] [[package]] @@ -636,6 +686,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -849,6 +910,12 @@ dependencies = [ "tracing-serde", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.26" @@ -878,6 +945,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" diff --git a/judge/README.md b/judge/README.md index d96bcfb..cf7b310 100644 --- a/judge/README.md +++ b/judge/README.md @@ -52,11 +52,38 @@ kf-worker judge-once --socket /run/kf-judge/task.sock \ Worker 选取业务授权的自身发布目录，准备有界封印的期望输出，并向 Engine 注入 `ExecutionPort`。适配器构造 native 规格，仅绑定程序与 stdin，不将期望输出放入用户运行根。helper 根据 `SO_PEERCRED` 核验配置中的 Worker UID/GID；客户端核验 helper 的 root 凭据。有界私有帧与 `SCM_RIGHTS` 只传持有的目录能力和只读封印输出，收到的 FD 重新检查类型、长度、访问模式、封印和 CLOEXEC，不传任意宿主挂载路径。 -helper 不接管 root 或其他身份的输入工件；每任务保留相同的隔离和预算门控。执行中客户端断开或出现额外控制数据会取消全树，清理后退出。socket 清理核对自身 inode，不删除外来替换。业务发布目录的选择与业务授权仍属于可信 Worker；文件所有权不是业务授权证明。 +helper 不接管 root 或其他身份的输入工件；每任务保留相同的隔离和预算门控。执行中客户端断开会取消全树，额外控制数据归为协议错误并拒绝执行，清理后退出。socket 清理核对自身 inode，不删除外来替换。业务发布目录的选择与业务授权仍属于可信 Worker；文件所有权不是业务授权证明。 Engine 仅在真实、无基础设施错误的 exit 0 后逐字节比较输出。非零退出/信号为 RuntimeError，CPU/墙钟、内存、输出限制分别映射 TLE/MLE/OLE，取消与 SystemError 单独表达；不可用计量保持 null。汇总在 Engine 内按输入顺序确定，基础设施错误或取消停止后续执行；空用例和重复 run ID 在执行前拒绝。`judge-once` 返回结构化 verdict、用例与实际计量。新增 `serde` 直接依赖用于本地作业/领域结果类型，Engine 不引入 sandbox 或 OS 库。 -首轮提供 native 单用例和编译/运行沙箱能力；平台队列消费、完整提交编排、交互题、SPJ 和部署仍未实现。`kf-worker --check` 只检查入口，不消费或执行任务。权限 helper 和镜像必须由部署方配置，Worker 不调用 sudo、不修改账号或权限。 +本地已提供 native 单用例及编译一次、多测试点编排；平台队列消费、持久化协调、完整提交投递、交互题、SPJ 和部署仍未实现。`kf-worker --check` 只检查入口，不消费或执行任务。权限 helper 和镜像必须由部署方配置，Worker 不调用 sudo、不修改账号或权限。 + +## 编译一次、多测试点 + +Root 服务 `kf-sandbox factory --config /etc/kf-judge/runtime.json` 先核对有限 cgroup +预留、配置/镜像权限和已确认归属的恢复记录，逐次产生独立单线程 helper。 +普通用户通过 `FactoryClient` 申请固定 run ID 的连接能力；配置、镜像、二进制、 +UID 池和宿主目录由服务决定。槽位满直接拒绝；有限持久 ID 记录跨重启阻止重放。 +工厂退出通过生命线取消 helper，清理失败保留租约并停止分配；服务配置见 +[部署说明](../deploy/README.md)。 + +```sh +kf-worker judge-submission --socket /var/lib/kf-judge/factory/control.sock \ + --job submission.json --artifacts /srv/kf-worker/authorized \ + --staging /srv/kf-worker/staging +``` + +本地作业包含版本化 `manifest`、`binding` 和 `manifest_digest`，清单摘要采用共享 +规范序列化。工件发布按已授权的 SHA-256 名称读取，暂存父目录由普通 Worker +持有且为 0700。暂存日志记录进程启动时间、内核启动标识和目录/文件 inode； +进程重启只回收已确认归属的旧文件，未知内容保留并拒绝继续。 + +Engine 的编译、运行、期望值和时钟/取消端口分离。C17/C++20 映射固定 GNU 参数， +无 shell 或用户 flags；每 attempt 编译一次，封印产物、计算摘要并完成清理后， +按清单顺序逐点申请新 helper。每点只传程序和 stdin，期望值留在 Worker 比较端。 +编译失败/超限为 CE，准备、导出或清理失败为 SystemError；取消/基础设施失败停止 +后续点并列出跳过原因。计量分开保存编译消耗与运行 CPU 总量、内存峰值，缺失值 +保留 null；完整清单缺点不能 Accepted。`judge-submission` 返回真实报告与阶段 PID。 ## 监督和恢复 diff --git a/judge/crates/engine/src/submission.rs b/judge/crates/engine/src/submission.rs index a3e277b..58d3392 100644 --- a/judge/crates/engine/src/submission.rs +++ b/judge/crates/engine/src/submission.rs @@ -118,7 +118,7 @@ pub fn judge_submission { c.report.validate()?; cleanup_complete = c.cleanup_complete; @@ -142,6 +142,13 @@ pub fn judge_submission= 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 { @@ -175,13 +182,18 @@ pub fn judge_submission { - let verdict = if matches!(o.kind, ExecutionKind::Exited { code: 0 }) { + 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 @@ -199,8 +211,6 @@ pub fn judge_submission u64 { self.clock + .saturating_add(self.calls as u64 * self.advance_us) } fn cancelled(&self) -> bool { self.calls >= self.cancel_after @@ -389,6 +401,7 @@ mod tests { expected_calls: 0, cancel_after: usize::MAX, clock: 0, + advance_us: 0, fail_compile: false, } } @@ -527,4 +540,16 @@ mod tests { assert!(judge_submission(&mut p, &m, b, d).is_err()); assert_eq!(p.compile_calls, 1); } + #[test] + fn deadline_interrupt_is_infrastructure_and_overrides_cancel_kind() { + let (m, b, d) = fixture(); + let mut p = port(vec![ExecutionKind::Cancelled]); + p.advance_us = m.job_wall_time_us; + let r = judge_submission(&mut p, &m, b, d).unwrap(); + assert_eq!(r.report.verdict, Verdict::SystemError); + assert_eq!(r.report.failure_code, Some(FailureCode::JobDeadline)); + assert_eq!(p.calls, 1); + assert_eq!(p.expected_calls, 0); + assert_eq!(r.skipped.len(), 2); + } } diff --git a/judge/crates/sandbox/Cargo.toml b/judge/crates/sandbox/Cargo.toml index 3eec2df..0ca7ebc 100644 --- a/judge/crates/sandbox/Cargo.toml +++ b/judge/crates/sandbox/Cargo.toml @@ -29,3 +29,6 @@ workspace = true [[example]] name = "identity_probe" test = false + +[dev-dependencies] +sha2 = "=0.10.9" diff --git a/judge/crates/sandbox/src/artifacts.rs b/judge/crates/sandbox/src/artifacts.rs index b3bfad6..7e4f5b7 100644 --- a/judge/crates/sandbox/src/artifacts.rs +++ b/judge/crates/sandbox/src/artifacts.rs @@ -363,3 +363,317 @@ fn snapshot( budget.remaining_bytes -= bytes; Ok(SnapshotFile { file, bytes }) } + +/// Ordinary Worker staging, distinct from immutable hidden-data publication. +/// Fixed entries only; no recursive cleanup or adoption of existing directories. +pub struct StagePublication { + root: ArtifactRoot, + parent: OwnedFd, + name: String, + dev: u64, + ino: u64, + files: Vec, + removed: bool, +} +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct StageEntry { + name: String, + dev: u64, + ino: u64, +} +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct StageOwner { + pid: u32, + ticks: u64, + boot: String, + dev: u64, + ino: u64, +} +fn stage_ticks(value: &str) -> io::Result { + value + .rsplit_once(") ") + .and_then(|(_, s)| s.split_whitespace().nth(19)) + .and_then(|s| s.parse::().ok()) + .filter(|v| *v > 0) + .ok_or_else(|| refused("stage owner time")) +} +impl StagePublication { + pub fn create(parent_path: &std::path::Path, name: &str) -> io::Result { + crate::spec::identifier(name, "stage").map_err(|_| refused("stage identifier"))?; + let parent = rustix::fs::open( + parent_path, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let st = rustix::fs::fstat(&parent)?; + if st.st_uid != rustix::process::geteuid().as_raw() + || st.st_uid < 1000 + || st.st_mode & 0o7777 != 0o700 + { + return Err(refused("private ordinary staging parent")); + } + rustix::fs::mkdirat(&parent, name, Mode::from_raw_mode(0o700))?; + let fd = rustix::fs::openat( + &parent, + name, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let st = rustix::fs::fstat(&fd)?; + let value = Self { + root: ArtifactRoot::from_directory(fd)?, + parent, + name: name.into(), + dev: st.st_dev, + ino: st.st_ino, + files: vec![], + removed: false, + }; + let header = StageOwner { + pid: std::process::id(), + ticks: stage_ticks(&std::fs::read_to_string("/proc/self/stat")?)?, + boot: std::fs::read_to_string("/proc/sys/kernel/random/boot_id")? + .trim() + .into(), + dev: st.st_dev, + ino: st.st_ino, + }; + let journal = rustix::fs::openat( + &value.root.fd, + ".owner", + OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::from_raw_mode(0o600), + )?; + let mut file = File::from(journal); + serde_json::to_writer(&mut file, &header)?; + file.write_all(b"\n")?; + file.sync_all()?; + rustix::fs::fsync(&value.root.fd)?; + Ok(value) + } + /// Startup recovery authenticates both journal and inode. A reused PID or + /// foreign object cannot confer authority; unknown content is preserved. + pub fn recover(parent_path: &std::path::Path) -> io::Result { + use rustix::fd::AsRawFd; + let parent = rustix::fs::open( + parent_path, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let st = rustix::fs::fstat(&parent)?; + if st.st_uid != rustix::process::geteuid().as_raw() + || st.st_uid < 1000 + || st.st_mode & 0o7777 != 0o700 + { + return Err(refused("ordinary recovery parent")); + } + let names: Vec<_> = std::fs::read_dir(format!("/proc/self/fd/{}", parent.as_raw_fd()))? + .take(513) + .map(|e| { + e?.file_name() + .into_string() + .map_err(|_| refused("stage name")) + }) + .collect::>()?; + if names.len() > 512 { + return Err(refused("stage recovery count")); + } + let boot = std::fs::read_to_string("/proc/sys/kernel/random/boot_id")?; + let mut recovered = 0; + for name in names { + crate::spec::identifier(&name, "stage") + .map_err(|_| refused("unrecognized staging entry retained"))?; + let fd = rustix::fs::openat( + &parent, + name.as_str(), + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let root = ArtifactRoot::from_directory(fd)?; + let st = rustix::fs::fstat(&root.fd)?; + let journal = rustix::fs::openat( + &root.fd, + ".owner", + OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let j = rustix::fs::fstat(&journal)?; + if j.st_uid != root.owner + || j.st_mode & 0o7777 != 0o600 + || j.st_nlink != 1 + || j.st_size < 0 + || j.st_size > 2048 + || 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) + .read_to_string(&mut bytes)?; + if bytes.len() > 2048 || !bytes.ends_with('\n') { + return Err(refused("stage journal framing")); + } + let mut lines = bytes.lines(); + let owner: StageOwner = + serde_json::from_str(lines.next().ok_or_else(|| refused("stage journal"))?) + .map_err(|_| refused("stage journal"))?; + if owner.boot != boot.trim() + || owner.pid < 2 + || owner.dev != st.st_dev + || owner.ino != st.st_ino + { + return Err(refused("stage ownership binding")); + } + match std::fs::read_to_string(format!("/proc/{}/stat", owner.pid)) { + Ok(stat) if stage_ticks(&stat)? == owner.ticks => continue, + Ok(_) => {} + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => return Err(e), + } + let entries = lines + .map(|l| { + serde_json::from_str::(l) + .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()) + }) + { + return Err(refused("stage entry set")); + } + let mut value = Self { + root, + parent: rustix::io::fcntl_dupfd_cloexec(&parent, 3)?, + name, + dev: st.st_dev, + ino: st.st_ino, + files: entries, + removed: false, + }; + value.cleanup()?; + recovered += 1; + } + Ok(recovered) + } + pub fn root(&self) -> &ArtifactRoot { + &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 + || bytes > MAX_SNAPSHOT_BYTES + { + return Err(refused("fixed staging entry/budget")); + } + let fd = rustix::fs::openat( + &self.root.fd, + name, + OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::from_raw_mode(0o600), + )?; + let s = rustix::fs::fstat(&fd)?; + let entry = StageEntry { + name: name.into(), + dev: s.st_dev, + ino: s.st_ino, + }; + let journal = rustix::fs::openat( + &self.root.fd, + ".owner", + OFlags::WRONLY | OFlags::APPEND | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let mut file = File::from(journal); + serde_json::to_writer(&mut file, &entry)?; + file.write_all(b"\n")?; + file.sync_all()?; + self.files.push(entry); + let mut file = File::from(fd); + let copied = io::copy(&mut reader.take(bytes + 1), &mut file)?; + if copied != bytes { + return Err(refused("staged artifact byte count")); + } + file.sync_all() + } + pub fn cleanup(&mut self) -> io::Result<()> { + if self.removed { + return Ok(()); + } + let named = rustix::fs::statat( + &self.parent, + self.name.as_str(), + rustix::fs::AtFlags::SYMLINK_NOFOLLOW, + )?; + if named.st_dev != self.dev || named.st_ino != self.ino { + return Err(refused("foreign staging replacement")); + } + use rustix::fd::AsRawFd; + let actual: Vec<_> = + std::fs::read_dir(format!("/proc/self/fd/{}", self.root.fd.as_raw_fd()))? + .take(5) + .map(|e| Ok(e?.file_name())) + .collect::>()?; + if actual.len() != self.files.len() + 1 + || actual + .iter() + .any(|n| n != ".owner" && !self.files.iter().any(|f| n == f.name.as_str())) + { + return Err(refused("foreign staging entry retained")); + } + for entry in &self.files { + let st = rustix::fs::statat( + &self.root.fd, + entry.name.as_str(), + rustix::fs::AtFlags::SYMLINK_NOFOLLOW, + )?; + if st.st_dev != entry.dev + || st.st_ino != entry.ino + || st.st_uid != self.root.owner + || st.st_nlink != 1 + || FileType::from_raw_mode(st.st_mode) != FileType::RegularFile + { + return Err(refused("foreign staging content")); + } + } + let journal = rustix::fs::statat( + &self.root.fd, + ".owner", + rustix::fs::AtFlags::SYMLINK_NOFOLLOW, + )?; + if journal.st_uid != self.root.owner + || journal.st_nlink != 1 + || journal.st_mode & 0o7777 != 0o600 + || FileType::from_raw_mode(journal.st_mode) != FileType::RegularFile + { + return Err(refused("foreign staging journal")); + } + for entry in &self.files { + rustix::fs::unlinkat( + &self.root.fd, + entry.name.as_str(), + rustix::fs::AtFlags::empty(), + )?; + } + rustix::fs::unlinkat(&self.root.fd, ".owner", rustix::fs::AtFlags::empty())?; + self.files.clear(); + rustix::fs::unlinkat( + &self.parent, + self.name.as_str(), + rustix::fs::AtFlags::REMOVEDIR, + )?; + self.removed = true; + Ok(()) + } +} +impl Drop for StagePublication { + fn drop(&mut self) { + let _ = self.cleanup(); + } +} diff --git a/judge/crates/sandbox/src/execution.rs b/judge/crates/sandbox/src/execution.rs index 2c90ffe..05e9524 100644 --- a/judge/crates/sandbox/src/execution.rs +++ b/judge/crates/sandbox/src/execution.rs @@ -77,7 +77,7 @@ fn read_pinned(fd: OwnedFd, limit: u64) -> io::Result> { Ok(bytes) } pub fn read_bounded_file(path: &Path, limit: u64) -> io::Result> { - if limit == 0 || limit > crate::spec::MAX_SPEC_BYTES { + if limit == 0 || limit > 262_144 { return Err(invalid("bounded JSON ceiling")); } let fd = rustix::fs::open( diff --git a/judge/crates/sandbox/src/factory.rs b/judge/crates/sandbox/src/factory.rs new file mode 100644 index 0000000..7337643 --- /dev/null +++ b/judge/crates/sandbox/src/factory.rs @@ -0,0 +1,448 @@ +//! Root service -> fresh single-threaded helper. Only a run identifier crosses +//! the allocation protocol. The accepted kernel-authenticated connection is the +//! capability, so the client never selects a helper path, executable or config. +use crate::{ + artifacts::ArtifactRoot, execution, launcher::protocol::WireMessage, rpc, spec::ExecutionSpec, +}; +use rustix::{ + fd::OwnedFd, + fs::{Mode, OFlags}, +}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashSet, + fs::File, + io::{self, Read, Write}, + os::unix::{ + fs::{FileTypeExt, MetadataExt, PermissionsExt}, + net::{UnixListener, UnixStream}, + }, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + time::{Duration, Instant}, +}; +const MAX_IDS: usize = 4096; +fn denied() -> io::Error { + io::Error::new(io::ErrorKind::PermissionDenied, "helper factory refused") +} +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Acquire { + run_id: String, +} +#[derive(Serialize, Deserialize)] +#[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)] +enum Allocation { + Ready { run_id: String, helper_pid: u32 }, + Refused {}, +} +impl crate::launcher::protocol::sealed::Sealed for Acquire {} +impl crate::launcher::protocol::sealed::Sealed for Allocation {} +impl WireMessage for Acquire { + fn valid(&self) -> bool { + crate::spec::identifier(&self.run_id, "run_id").is_ok() + } +} +impl WireMessage for Allocation { + fn valid(&self) -> bool { + match self { + Self::Ready { run_id, helper_pid } => { + *helper_pid > 1 && crate::spec::identifier(run_id, "run_id").is_ok() + } + Self::Refused {} => true, + } + } +} +pub struct FactoryClient { + socket: PathBuf, +} +impl FactoryClient { + pub fn new(socket: PathBuf) -> Self { + Self { socket } + } + /// The caller holds only ordinary artifact and Unix socket capabilities. + pub fn execute( + &self, + spec: ExecutionSpec, + publication: &ArtifactRoot, + ) -> io::Result<(rpc::Executed, u32)> { + self.execute_controlled(spec, publication, None, None) + } + pub fn execute_controlled( + &self, + spec: ExecutionSpec, + publication: &ArtifactRoot, + deadline: Option, + cancellation: Option<&std::sync::atomic::AtomicBool>, + ) -> io::Result<(rpc::Executed, u32)> { + if rustix::process::geteuid().as_raw() < 1000 || rustix::process::getegid().as_raw() < 1000 + { + return Err(denied()); + } + let validated = spec + .clone() + .validate(&crate::spec::SpecPolicy::default()) + .map_err(|_| denied())?; + let stream = UnixStream::connect(&self.socket)?; + let interrupter = stream.try_clone()?; + let done = std::sync::atomic::AtomicBool::new(false); + std::thread::scope(|scope| { + let watch = scope.spawn(|| { + while !done.load(std::sync::atomic::Ordering::Acquire) { + if deadline.is_some_and(|d| Instant::now() >= d) + || cancellation + .is_some_and(|c| c.load(std::sync::atomic::Ordering::Acquire)) + { + // Half close is an authenticated cancellation. Preserve the + // read side for the final report and sealed cleanup proof. + let _ = interrupter.shutdown(std::net::Shutdown::Write); + break; + } + std::thread::sleep(Duration::from_millis(2)); + } + }); + let result = (|| { + let mut channel = rpc::channel(stream, 0, 0)?; + channel.send(&Acquire { + run_id: spec.run_id.clone(), + })?; + let pid = match channel.receive::()? { + Allocation::Ready { run_id, helper_pid } if run_id == spec.run_id => helper_pid, + _ => return Err(denied()), + }; + Ok(( + rpc::execute_channel(channel, spec, publication, validated)?, + pid, + )) + })(); + done.store(true, std::sync::atomic::Ordering::Release); + watch.join().map_err(|_| denied())?; + result + }) + } +} +struct SocketLease { + path: PathBuf, + dev: u64, + ino: u64, +} +impl Drop for SocketLease { + fn drop(&mut self) { + if let Ok(m) = std::fs::symlink_metadata(&self.path) + && m.dev() == self.dev + && m.ino() == self.ino + { + let _ = std::fs::remove_file(&self.path); + } + } +} +struct Active { + child: Child, + _guardian: UnixStream, +} +fn private_file(path: &Path) -> io::Result { + let fd = match rustix::fs::open( + path, + OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + ) { + Ok(pin) => { + let st = rustix::fs::fstat(&pin)?; + if rustix::fs::FileType::from_raw_mode(st.st_mode) != rustix::fs::FileType::RegularFile + { + return Err(denied()); + } + use rustix::fd::AsRawFd; + rustix::fs::open( + format!("/proc/self/fd/{}", pin.as_raw_fd()), + OFlags::RDWR | OFlags::CLOEXEC, + Mode::empty(), + )? + } + Err(rustix::io::Errno::NOENT) => rustix::fs::open( + path, + OFlags::RDWR | OFlags::CREATE | OFlags::EXCL | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::from_raw_mode(0o600), + )?, + Err(e) => return Err(e.into()), + }; + let s = rustix::fs::fstat(&fd)?; + if s.st_uid != 0 + || s.st_gid != 0 + || s.st_mode & 0o7777 != 0o600 + || s.st_nlink != 1 + || rustix::fs::FileType::from_raw_mode(s.st_mode) != rustix::fs::FileType::RegularFile + { + return Err(denied()); + } + Ok(File::from(fd)) +} +fn idle_recovery(config: &crate::runtime::ValidatedRuntimeConfig) -> io::Result<()> { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let report = crate::cleanup::recover(config)?; + if report + .items + .iter() + .any(|r| !matches!(r.status, "active" | "reclaimed")) + { + return Err(denied()); + } + if !report.items.iter().any(|r| r.status == "active") { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(denied()); + } + std::thread::sleep(Duration::from_millis(10)); + } +} +/// Fixed executable/config come solely from the trusted service invocation. +/// Runtime is finite even when the service manager is misconfigured; cgroup +/// reservations are read back by authorize, and tasks retain their own gates. +pub fn serve(config_path: &Path, lifetime_seconds: u64) -> io::Result<()> { + if !(1..=86400).contains(&lifetime_seconds) { + return Err(denied()); + } + // Root management receives no ambient business/credential environment. + if std::env::vars_os().any(|(k, v)| { + !matches!( + (k.to_str(), v.to_str()), + (Some("PATH"), Some("/usr/bin:/bin")) + | (Some("LC_ALL" | "LANG"), Some("C" | "C.UTF-8")) + | (Some("TZ"), Some("UTC")) + ) + }) { + return Err(denied()); + } + let config = execution::load_config(config_path)?; + execution::authorize(&config)?; + let policy = config.config(); + if policy.aggregate.supervisor_processes < 4 + 8 * policy.max_concurrent_runs { + return Err(denied()); + } + let binary = std::env::current_exe()?; + // Pin and validate the service binary; Worker cannot replace its ancestors. + let parent = crate::isolation::root::open_directory(binary.parent().ok_or_else(denied)?)?; + let fd = rustix::fs::openat( + &parent, + binary.file_name().ok_or_else(denied)?, + OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let st = rustix::fs::fstat(&fd)?; + if st.st_uid != 0 + || st.st_gid != 0 + || st.st_mode & 0o7022 != 0 + || rustix::fs::FileType::from_raw_mode(st.st_mode) != rustix::fs::FileType::RegularFile + { + return Err(denied()); + } + let root = Path::new(&policy.state_root).join("factory"); + if !root.exists() { + std::fs::create_dir(&root)?; + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o711))?; + } + let root_fd = crate::isolation::root::open_directory(&root)?; + let lock = private_file(&root.join("lock"))?; + rustix::fs::flock(&lock, rustix::fs::FlockOperation::NonBlockingLockExclusive)?; + // Previous live children cancel through their guardian connection. Durable + // task journals alone authorize recovery; foreign entries fail closed. + idle_recovery(&config)?; + let socket = root.join("control.sock"); + let record_path = root.join("socket.json"); + if let Ok(m) = std::fs::symlink_metadata(&socket) { + let bytes = execution::read_bounded_file(&record_path, 1024)?; + let record: serde_json::Value = serde_json::from_slice(&bytes).map_err(|_| denied())?; + let trusted = std::fs::symlink_metadata(&record_path)?; + if trusted.uid() != 0 + || trusted.gid() != 0 + || trusted.mode() & 0o7777 != 0o600 + || trusted.nlink() != 1 + || !trusted.is_file() + || !m.file_type().is_socket() + || m.uid() != 0 + || record != serde_json::json!({"dev":m.dev(),"ino":m.ino()}) + { + return Err(denied()); + } + std::fs::remove_file(&socket)?; + } + let mut ledger = private_file(&root.join("requests"))?; + if ledger.metadata()?.len() > (MAX_IDS * 65) as u64 { + return Err(denied()); + } + let mut old = String::new(); + (&mut ledger) + .take((MAX_IDS * 65 + 1) as u64) + .read_to_string(&mut old)?; + let mut ids = HashSet::new(); + for id in old.lines() { + if crate::spec::identifier(id, "run_id").is_err() || !ids.insert(id.to_owned()) { + return Err(denied()); + } + } + if ids.len() > MAX_IDS || (!old.is_empty() && !old.ends_with('\n')) { + return Err(denied()); + } + let listener = UnixListener::bind(&socket)?; + let m = std::fs::symlink_metadata(&socket)?; + let _lease = SocketLease { + path: socket.clone(), + dev: m.dev(), + ino: m.ino(), + }; + let record = serde_json::to_vec(&serde_json::json!({"dev":m.dev(),"ino":m.ino()}))?; + let mut f = private_file(&record_path)?; + f.set_len(0)?; + f.write_all(&record)?; + f.sync_all()?; + rustix::fs::fsync(&root_fd)?; + std::os::unix::fs::chown(&socket, Some(0), Some(policy.identities.worker_gid))?; + std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o660))?; + listener.set_nonblocking(true)?; + // Actual overall task delegation limits, including concurrency and no swap. + let _delegated = crate::cgroup::DelegatedCgroup::open(&config).map_err(|_| denied())?; + let mut active: Vec = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(lifetime_seconds); + let result = (|| { + while Instant::now() < deadline { + let mut i = 0; + while i < active.len() { + if let Some(status) = active[i].child.try_wait()? { + active.swap_remove(i); + // A failed helper may have durable residuals. Never start + // another task until recovery proves safe release. + if !status.success() { + idle_recovery(&config)?; + } + } else { + i += 1; + } + } + let stream = match listener.accept() { + Ok((s, _)) => s, + Err(e) if e.kind() == io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(5)); + continue; + } + Err(e) => return Err(e), + }; + let peer = stream.try_clone()?; + let Ok(mut channel) = rpc::channel( + stream, + policy.identities.worker_uid, + policy.identities.worker_gid, + ) else { + continue; + }; + let Ok(request) = channel.receive::() else { + continue; + }; + if active.len() >= policy.max_concurrent_runs as usize + || ids.len() >= MAX_IDS + || ids.contains(&request.run_id) + { + let _ = channel.send(&Allocation::Refused {}); + continue; + } + // Persist before allocating; IDs remain consumed across restarts. + writeln!(ledger, "{}", request.run_id)?; + ledger.sync_all()?; + ids.insert(request.run_id.clone()); + let (guardian, child_guardian) = UnixStream::pair()?; + let child = Command::new(&binary) + .env_clear() + .args(["factory-child", "--config"]) + .arg(config_path) + .arg("--run-id") + .arg(&request.run_id) + .stdin(Stdio::from(OwnedFd::from(peer))) + .stdout(Stdio::from(OwnedFd::from(child_guardian))) + .stderr(Stdio::null()) + .spawn()?; + let helper_pid = child.id(); + active.push(Active { + child, + _guardian: guardian, + }); + if channel + .send(&Allocation::Ready { + run_id: request.run_id, + helper_pid, + }) + .is_err() + { + // Helper receives EOF, or times out within its fixed deadline. + } + } + Ok(()) + })(); + drop(listener); + // Guardian shutdown initiates the same bounded task-tree termination. + for a in &active { + let _ = a._guardian.shutdown(std::net::Shutdown::Both); + } + let stop_deadline = Instant::now() + Duration::from_secs(20); + while !active.is_empty() && Instant::now() < stop_deadline { + let mut i = 0; + while i < active.len() { + if active[i].child.try_wait()?.is_some() { + active.swap_remove(i); + } else { + i += 1; + } + } + std::thread::sleep(Duration::from_millis(5)); + } + if !active.is_empty() { + return Err(denied()); + } + idle_recovery(&config)?; + result +} +/// Hidden child role: fixed accepted socket on stdin, factory lifeline on stdout. +/// No credentials, path selection, ambient stdio or previous task state survive. +pub fn child(config_path: &Path, run_id: &str) -> io::Result<()> { + // Duplicate only the fixed descriptors installed by the root factory. + let stream: UnixStream = rustix::io::fcntl_dupfd_cloexec(rustix::stdio::stdin(), 3)?.into(); + let guardian = rustix::io::fcntl_dupfd_cloexec(rustix::stdio::stdout(), 3)?; + let peer = rustix::net::sockopt::socket_peercred(&guardian)?; + if peer.uid.as_raw() != 0 + || peer.gid.as_raw() != 0 + || peer.pid != rustix::process::getppid().ok_or_else(denied)? + { + return Err(denied()); + } + let config = execution::load_config(config_path)?; + execution::authorize(&config)?; + if crate::spec::identifier(run_id, "run_id").is_err() { + return Err(denied()); + } + let result = rpc::serve_stream(&config, stream, Some(run_id), Some(&guardian))?; + if result["report"]["cleanup"] != "complete" { + return Err(denied()); + } + Ok(()) +} +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn protocol_never_accepts_worker_control_of_root_command_or_path() { + for data in [ + r#"{"run_id":"a","argv":["/bin/sh"]}"#, + r#"{"run_id":"a","config":"/tmp/x"}"#, + r#"{"run_id":"../x"}"#, + r#"{"run_id":""}"#, + ] { + let decoded = serde_json::from_str::(data); + assert!(decoded.is_err() || !decoded.unwrap().valid()); + } + assert!( + serde_json::from_str::(r#"{"run_id":"a-1"}"#) + .unwrap() + .valid() + ); + } +} diff --git a/judge/crates/sandbox/src/launcher/run.rs b/judge/crates/sandbox/src/launcher/run.rs index fafda42..d183c9d 100644 --- a/judge/crates/sandbox/src/launcher/run.rs +++ b/judge/crates/sandbox/src/launcher/run.rs @@ -41,6 +41,7 @@ pub(crate) struct RunProof { #[derive(Clone, Copy)] pub(crate) enum MonitorMode<'a> { Peer(&'a rustix::fd::OwnedFd), + FactoryPeers(&'a rustix::fd::OwnedFd, &'a rustix::fd::OwnedFd), Normal, #[cfg(feature = "vm-acceptance")] CancelAfter(u64), @@ -79,7 +80,7 @@ fn consume(event: BudgetEvent) -> (ResourceUsage, Vec) { } } -fn peer_closed(fd: &rustix::fd::OwnedFd) -> io::Result { +pub(crate) fn peer_closed(fd: &rustix::fd::OwnedFd) -> io::Result { use rustix::event::{PollFd, PollFlags, poll}; let mut f = [PollFd::new( fd, @@ -92,7 +93,23 @@ fn peer_closed(fd: &rustix::fd::OwnedFd) -> io::Result { tv_nsec: 0, }), )?; - Ok(!f[0].revents().is_empty()) + if f[0].revents().is_empty() { + return Ok(false); + } + let mut byte = [0u8; 1]; + match rustix::net::recv( + fd, + &mut byte, + rustix::net::RecvFlags::PEEK | rustix::net::RecvFlags::DONTWAIT, + ) { + Ok((0, _)) => Ok(true), + Ok(_) => Err(io::Error::new( + io::ErrorKind::InvalidData, + "unexpected control bytes during execution", + )), + Err(rustix::io::Errno::AGAIN) => Ok(false), + Err(e) => Err(e.into()), + } } pub(crate) fn execute( config: &ValidatedRuntimeConfig, @@ -157,9 +174,11 @@ pub(crate) fn execute( }); } let released = (|| { - if let MonitorMode::Peer(fd) = mode - && peer_closed(fd)? - { + if match mode { + MonitorMode::Peer(fd) => peer_closed(fd)?, + MonitorMode::FactoryPeers(a, b) => peer_closed(a)? || peer_closed(b)?, + _ => false, + } { return Err(io::Error::new( io::ErrorKind::Interrupted, "client disconnected before exec", @@ -174,6 +193,10 @@ pub(crate) fn execute( })(); let main_fd = match released { Ok(fd) => Some(fd), + Err(e) if e.kind() == io::ErrorKind::Interrupted => { + cancelled = Some(0); + None + } Err(e) => { errors.push(error(ExecutionStage::Exec, ErrorCode::ProtocolFailed, &e)); None @@ -194,7 +217,7 @@ pub(crate) fn execute( None }; let mut monitoring_failed = !errors.is_empty(); - if !monitoring_failed { + if !monitoring_failed && cancelled.is_none() { loop { let event = match budget.poll() { Ok(e) => e, @@ -275,10 +298,21 @@ pub(crate) fn execute( monitoring_failed = true; } } - match mode { - MonitorMode::Peer(fd) if peer_closed(fd).unwrap_or(true) => { - cancelled = Some(wall(&usage)); + let connection = match mode { + MonitorMode::Peer(fd) => peer_closed(fd), + MonitorMode::FactoryPeers(a, b) => peer_closed(a) + .and_then(|closed| if closed { Ok(true) } else { peer_closed(b) }), + _ => Ok(false), + }; + match connection { + Ok(true) => cancelled = Some(wall(&usage)), + Err(e) => { + errors.push(error(ExecutionStage::Wait, ErrorCode::ProtocolFailed, &e)); + monitoring_failed = true; } + Ok(false) => {} + } + match mode { #[cfg(feature = "vm-acceptance")] MonitorMode::CancelAfter(t) if wall(&usage) >= t => { cancelled = Some(wall(&usage)) @@ -476,7 +510,14 @@ pub(crate) fn execute( } let stdout = SnapshotFile::from_capture(pump.stdout()); let stderr = SnapshotFile::from_capture(pump.stderr()); - let outputs = if errors.iter().any(|e| e.stage == ExecutionStage::Cleanup) { + let compile_failed = matches!( + spec.spec().profile, + crate::spec::ExecutionProfile::CompileC | crate::spec::ExecutionProfile::CompileCpp + ) && (main != Some(Termination::Exited { code: 0 }) + || cancelled.is_some() + || !limits.is_empty() + || !errors.is_empty()); + let outputs = if compile_failed || errors.iter().any(|e| e.stage == ExecutionStage::Cleanup) { Vec::new() } else { match target.root().final_outputs(&work) { diff --git a/judge/crates/sandbox/src/lib.rs b/judge/crates/sandbox/src/lib.rs index eb5eaa5..d510120 100644 --- a/judge/crates/sandbox/src/lib.rs +++ b/judge/crates/sandbox/src/lib.rs @@ -33,3 +33,6 @@ pub mod execution; #[cfg(target_os = "linux")] pub mod rpc; + +#[cfg(target_os = "linux")] +pub mod factory; diff --git a/judge/crates/sandbox/src/main.rs b/judge/crates/sandbox/src/main.rs index 744e7b0..2c537db 100644 --- a/judge/crates/sandbox/src/main.rs +++ b/judge/crates/sandbox/src/main.rs @@ -47,6 +47,21 @@ enum Command { #[arg(long)] socket: PathBuf, }, + /// Root factory: finite service lifetime and fresh authenticated helpers. + Factory { + #[arg(long)] + config: PathBuf, + #[arg(long, default_value_t = 3600)] + lifetime_seconds: u64, + }, + #[cfg(target_os = "linux")] + #[command(hide = true)] + FactoryChild { + #[arg(long)] + config: PathBuf, + #[arg(long)] + run_id: String, + }, /// Internal trusted launcher control entry; never runs a user program. #[cfg(target_os = "linux")] #[command(hide = true)] @@ -142,6 +157,33 @@ fn main() -> ExitCode { spec_error("unsupported_platform", None) } } + Command::Factory { + config, + lifetime_seconds, + } => { + #[cfg(target_os = "linux")] + { + if kf_sandbox::factory::serve(&config, lifetime_seconds).is_ok() { + ExitCode::SUCCESS + } else { + spec_error("factory_refused", None) + } + } + #[cfg(not(target_os = "linux"))] + { + let _ = (config, lifetime_seconds); + spec_error("unsupported_platform", None) + } + } + #[cfg(target_os = "linux")] + Command::FactoryChild { config, run_id } => { + // stdout is the root factory lifeline, never a log stream. + if kf_sandbox::factory::child(&config, &run_id).is_ok() { + ExitCode::SUCCESS + } else { + ExitCode::from(2) + } + } Command::Validate { spec } => validate_spec(spec), Command::CheckRuntime { config } => check_runtime(config), #[cfg(target_os = "linux")] diff --git a/judge/crates/sandbox/src/rpc.rs b/judge/crates/sandbox/src/rpc.rs index 81be740..04277ae 100644 --- a/judge/crates/sandbox/src/rpc.rs +++ b/judge/crates/sandbox/src/rpc.rs @@ -64,7 +64,7 @@ impl WireMessage for Reply { } } } -fn channel(stream: UnixStream, uid: u32, gid: u32) -> io::Result { +pub(crate) fn channel(stream: UnixStream, uid: u32, gid: u32) -> io::Result { let c = rustix::net::sockopt::socket_peercred(&stream)?; ControlChannel::new( stream, @@ -106,7 +106,15 @@ pub fn execute( .clone() .validate(&SpecPolicy::default()) .map_err(|_| denied())?; - let mut c = channel(UnixStream::connect(socket)?, 0, 0)?; + let c = channel(UnixStream::connect(socket)?, 0, 0)?; + execute_channel(c, spec, publication, validated) +} +pub(crate) fn execute_channel( + mut c: ControlChannel, + spec: ExecutionSpec, + publication: &ArtifactRoot, + validated: crate::spec::ValidatedSpec, +) -> io::Result { c.send(&Request { spec })?; c.send_fd(publication)?; c.execution_deadline(validated.spec().limits.wall_time_us)?; @@ -205,27 +213,44 @@ pub fn serve_one(config_path: &Path, socket: &Path) -> io::Result, + guardian: Option<&OwnedFd>, +) -> io::Result { let ids = &config.config().identities; let mut c = channel(stream, ids.worker_uid, ids.worker_gid)?; let request = c.receive::()?; + if binding.is_some_and(|id| id != request.spec.run_id) { + return Err(denied()); + } 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)?; match execution::execute( - &config, + config, request.spec, &artifacts, - run::MonitorMode::Peer(&peer), + guardian.map_or(run::MonitorMode::Peer(&peer), |g| { + run::MonitorMode::FactoryPeers(&peer, g) + }), ) { Ok((result, _)) => { let report = result.report; // Cleanup failures remain in the report; caller may never claim a successful verdict. - if report.cancelled_at_wall_time_us.is_some() { + if report.cleanup != crate::outcome::CleanupState::Complete { + return Err(denied()); + } + let sent = c.send(&Reply::Completed { + report: Box::new(report.clone()), + }); + if sent.is_err() && report.cancelled_at_wall_time_us.is_some() { return Ok(serde_json::json!({"report":report,"proof":result.proof})); } - c.send(&Reply::Completed { - report: Box::new(report.clone()), - })?; + sent?; c.send_fd(&result.stdout)?; c.send_fd(&result.stderr)?; for o in result.outputs { diff --git a/judge/crates/sandbox/tests/pipeline_vm.rs b/judge/crates/sandbox/tests/pipeline_vm.rs new file mode 100644 index 0000000..cff60ac --- /dev/null +++ b/judge/crates/sandbox/tests/pipeline_vm.rs @@ -0,0 +1,714 @@ +#![cfg(all(target_os = "linux", feature = "vm-acceptance"))] +#[allow(dead_code)] +mod support; +use kf_sandbox::{ + artifacts::ArtifactRoot, + factory::FactoryClient, + outcome::{CleanupState, ExecutionOutcome}, + spec::ExecutionSpec, +}; +use sha2::{Digest as _, Sha256}; +use std::{ + fs, + io::{Read, Write}, + os::unix::{ + fs::{PermissionsExt, chown}, + net::UnixStream, + process::CommandExt, + }, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + sync::atomic::{AtomicBool, AtomicU32, Ordering}, + time::{Duration, Instant}, +}; +static COUNT: AtomicU32 = AtomicU32::new(0); +const SOURCE:&[u8]=br#"#include +#include +#include +int main(void){ unsigned char c; FILE *f=fopen("previous","rb"); if(f) return 91; f=fopen("previous","wb"); if(!f)return 92; fclose(f); if(access("expected",F_OK)==0)return 93; while(fread(&c,1,1,stdin)){if(c=='!'){fputs("live",stdout);fflush(stdout);for(;;){}} if(c=='#')return 7; fwrite(&c,1,1,stdout);} return 0; } +"#; +fn worker_file(p: &Path) { + chown(p, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(p, fs::Permissions::from_mode(0o600)).unwrap(); +} +fn id(n: u32) -> String { + format!("aaaaaaaa-aaaa-4aaa-8aaa-{n:012}") +} +fn blob(dir: &Path, data: &[u8]) -> serde_json::Value { + let hash = format!("{:x}", Sha256::digest(data)); + let path = dir.join(&hash); + if !path.exists() { + fs::write(&path, data).unwrap(); + worker_file(&path); + } + serde_json::json!({"digest":hash,"bytes":data.len()}) +} +struct Fixture { + root: support::Root, + dir: PathBuf, + config: PathBuf, + socket: PathBuf, + factory: Option, +} +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; + let dir = PathBuf::from(format!( + "/kf/pipeline-{}-{}", + std::process::id(), + COUNT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&dir).unwrap(); + fs::set_permissions(&dir, fs::Permissions::from_mode(0o711)).unwrap(); + let config = dir.join("runtime.json"); + fs::write(&config, serde_json::to_vec(&root.config).unwrap()).unwrap(); + fs::set_permissions(&config, fs::Permissions::from_mode(0o600)).unwrap(); + for n in ["artifacts", "staging"] { + let p = dir.join(n); + fs::create_dir(&p).unwrap(); + chown(&p, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(&p, fs::Permissions::from_mode(0o700)).unwrap(); + } + let socket = PathBuf::from("/var/lib/kf-judge/factory/control.sock"); + Self { + root, + dir, + config, + socket, + factory: None, + } + } + fn start(&mut self) { + use std::os::unix::fs::MetadataExt; + let previous = fs::symlink_metadata(&self.socket).ok().map(|m| m.ino()); + let mut child = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap()) + .env_clear() + .args(["factory", "--config"]) + .arg(&self.config) + .args(["--lifetime-seconds", "180"]) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(15); + loop { + assert!( + child.try_wait().unwrap().is_none(), + "factory exited before ready" + ); + if self.socket.exists() + && fs::metadata(&self.socket).unwrap().permissions().mode() & 0o777 == 0o660 + && previous != Some(fs::metadata(&self.socket).unwrap().ino()) + { + break; + } + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(5)); + } + let own = fs::read_to_string(format!("/proc/{}/cgroup", child.id())).unwrap(); + let group = PathBuf::from(format!( + "/sys/fs/cgroup{}", + own.trim().strip_prefix("0::").unwrap() + )); + let values: serde_json::Value = ["memory.max", "memory.swap.max", "pids.max", "cpu.max"] + .iter() + .map(|n| { + ( + n.to_string(), + serde_json::Value::String( + fs::read_to_string(group.join(n)).unwrap().trim().into(), + ), + ) + }) + .collect::>() + .into(); + assert_eq!( + values, + serde_json::json!({"memory.max":"268435456","memory.swap.max":"0","pids.max":"32","cpu.max":"50000 100000"}) + ); + println!( + "KF_FACTORY_BUDGET_PROOF {}", + serde_json::json!({"pid":child.id(),"actual_factory_limits":values,"slots":1,"lifetime_seconds":180}) + ); + let deadline = Instant::now() + Duration::from_secs(5); + while fs::read_to_string(self.root.path.join("memory.max")) + .unwrap() + .trim() + != self.root.config.aggregate.memory_bytes.to_string() + { + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(5)); + } + let total: serde_json::Value = [ + "memory.max", + "memory.swap.max", + "pids.max", + "cpu.max", + "cgroup.max.depth", + "cgroup.max.descendants", + ] + .iter() + .map(|n| { + ( + n.to_string(), + serde_json::Value::String( + fs::read_to_string(self.root.path.join(n)) + .unwrap() + .trim() + .into(), + ), + ) + }) + .collect::>() + .into(); + assert_eq!( + total, + serde_json::json!({"memory.max":self.root.config.aggregate.memory_bytes.to_string(),"memory.swap.max":"0","pids.max":self.root.config.aggregate.processes.to_string(),"cpu.max":format!("{} {}",self.root.config.aggregate.cpu_quota_us,self.root.config.aggregate.cpu_period_us),"cgroup.max.depth":"1","cgroup.max.descendants":"1"}) + ); + println!("KF_FACTORY_TOTAL_PROOF {}", total); + self.factory = Some(child); + } + fn job(&self, language: &str, source: &[u8], inputs: &[&[u8]], answers: &[&[u8]]) -> PathBuf { + let n = COUNT.fetch_add(1, Ordering::Relaxed) + 100; + let mut m: serde_json::Value = serde_json::from_str(include_str!( + "../../../../tests/fixtures/contracts/judge-manifest.valid.json" + )) + .unwrap(); + let a = self.dir.join("artifacts"); + m["language"] = language.into(); + m["source"] = blob(&a, source); + m["cases"]=serde_json::Value::Array(inputs.iter().zip(answers).enumerate().map(|(i,(input,expected))|serde_json::json!({"case_id":id(1000+i as u32),"input":blob(&a,input),"expected":blob(&a,expected)})).collect()); + m["job_wall_time_us"] = 300_000_000u64.into(); + let parsed: kf_contracts::judging::JobManifest = serde_json::from_value(m.clone()).unwrap(); + let hash = format!( + "{:x}", + Sha256::digest(kf_contracts::judging::manifest_bytes(&parsed).unwrap()) + ); + let path = self.dir.join(format!("job-{n}.json")); + fs::write(&path,serde_json::to_vec(&serde_json::json!({"manifest":m,"manifest_digest":hash,"binding":{"evaluation_id":parsed.evaluation_id,"attempt_id":id(n),"worker_id":id(999),"fencing_version":1}})).unwrap()).unwrap(); + worker_file(&path); + path + } + fn worker(&self, job: &Path) -> Child { + Command::new(std::env::var_os("KF_WORKER_TEST_BIN").unwrap()) + .env_clear() + .args(["judge-submission", "--socket"]) + .arg(&self.socket) + .arg("--job") + .arg(job) + .arg("--artifacts") + .arg(self.dir.join("artifacts")) + .arg("--staging") + .arg(self.dir.join("staging")) + .uid(1000) + .gid(1000) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap() + } + fn wait_empty(&self) { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let empty = ["tasks", "identities", "captures"].iter().all(|n| { + let p = PathBuf::from(format!("/var/lib/kf-judge/{n}")); + !p.exists() || fs::read_dir(p).unwrap().count() == 0 + }); + let groups = fs::read_dir(&self.root.path) + .unwrap() + .all(|e| !e.unwrap().file_type().unwrap().is_dir()); + if empty && groups { + break; + } + assert!( + Instant::now() < deadline, + "owned task resources did not clear" + ); + std::thread::sleep(Duration::from_millis(5)); + } + } + fn run(&self, job: &Path, verdict: &str) -> serde_json::Value { + let output = self.worker(job).wait_with_output().unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["result"]["report"]["verdict"], verdict, "{value}"); + assert_eq!(value["result"]["report"]["cleanup_complete"], true); + self.wait_empty(); + assert_eq!(fs::read_dir(self.dir.join("staging")).unwrap().count(), 0); + println!( + "KF_PIPELINE_PROOF {}", + serde_json::json!({"worker_uid":1000,"worker_gid":1000,"value":value}) + ); + value + } + fn live(&self) { + let path = PathBuf::from("/var/lib/kf-judge/identities/uid-100000/owner.json"); + let deadline = Instant::now() + Duration::from_secs(30); + loop { + if let Ok(text) = fs::read_to_string(&path) + && text.contains("r-") + { + let parsed: serde_json::Value = serde_json::from_str(&text).unwrap(); + let pid = parsed["owner"]["pid"].as_u64().unwrap(); + let prefix = format!( + "/var/lib/kf-judge/captures/kf-{}", + parsed["run_id"].as_str().unwrap() + ); + for e in fs::read_dir(format!("/proc/{pid}/fd")).unwrap().flatten() { + if fs::read_link(e.path()) + .is_ok_and(|p| p.to_string_lossy().starts_with(&prefix)) + && fs::metadata(e.path()).is_ok_and(|m| m.is_file() && m.len() >= 4) + { + let mut bytes = [0; 4]; + if fs::File::open(e.path()) + .unwrap() + .read_exact(&mut bytes) + .is_ok() + && bytes == *b"live" + { + let status = fs::read_to_string(format!("/proc/{pid}/status")).unwrap(); + assert!(status.contains("Uid:\t0\t0\t0\t0")); + assert!(status.contains("Threads:\t1")); + assert!(fs::read(format!("/proc/{pid}/environ")).unwrap().is_empty()); + println!( + "KF_FACTORY_LIVE_PROOF {}", + serde_json::json!({"pid":pid,"uid":0,"threads":1,"environment_empty":true,"stdout":"live","run_id":parsed["run_id"]}) + ); + return; + } + } + } + } + assert!(Instant::now() < deadline, "runtime task not reached"); + std::thread::sleep(Duration::from_millis(5)); + } + } + fn probe(&self, mode: &str, run_id: &str) -> serde_json::Value { + let o = Command::new(std::env::current_exe().unwrap()) + .env_clear() + .args([ + "--ignored", + "--exact", + "ordinary_control_probe", + "--test-threads=1", + "--nocapture", + ]) + .env("KF_FACTORY_PROBE", mode) + .env("KF_FACTORY_SOCKET", &self.socket) + .env("KF_FACTORY_INPUTS", self.dir.join("control-inputs")) + .env("KF_FACTORY_RUN", run_id) + .uid(1000) + .gid(1000) + .output() + .unwrap(); + assert!( + o.status.success(), + "{} {}", + String::from_utf8_lossy(&o.stdout), + String::from_utf8_lossy(&o.stderr) + ); + let text = String::from_utf8(o.stdout).unwrap(); + let value: serde_json::Value = serde_json::from_str( + text.lines() + .find(|l| l.contains("KF_FACTORY_CONTROL_PROOF ")) + .unwrap() + .split_once("KF_FACTORY_CONTROL_PROOF ") + .unwrap() + .1, + ) + .unwrap(); + println!("KF_FACTORY_CONTROL_PROOF {value}"); + value + } + fn control_inputs(&self) { + let p = self.dir.join("control-inputs"); + fs::create_dir(&p).unwrap(); + chown(&p, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(&p, fs::Permissions::from_mode(0o700)).unwrap(); + fs::copy("/kf/bin/worker-native", p.join("program")).unwrap(); + fs::write(p.join("stdin"), b"!").unwrap(); + for n in ["program", "stdin"] { + worker_file(&p.join(n)); + } + } +} +impl Drop for Fixture { + fn drop(&mut self) { + if let Some(mut child) = self.factory.take() { + let _ = child.kill(); + let _ = child.wait(); + } + let _ = fs::remove_dir_all(&self.dir); + } +} +#[test] +#[ignore = "independent capped VM; ordinary authenticated probe child only"] +fn ordinary_control_probe() { + let Ok(mode) = std::env::var("KF_FACTORY_PROBE") else { + return; + }; + let socket = PathBuf::from(std::env::var_os("KF_FACTORY_SOCKET").unwrap()); + let run_id = std::env::var("KF_FACTORY_RUN").unwrap(); + let publication = + ArtifactRoot::open(Path::new(&std::env::var_os("KF_FACTORY_INPUTS").unwrap())).unwrap(); + let mut spec: ExecutionSpec = + serde_json::from_str(include_str!("fixtures/spec.valid.json")).unwrap(); + spec.run_id = run_id.clone(); + spec.inputs[0].artifact_id = "program".into(); + spec.stdin_artifact = Some("stdin".into()); + let client = FactoryClient::new(socket.clone()); + match mode.as_str() { + "cancel" => { + let cancelled = AtomicBool::new(false); + let (result, pid) = std::thread::scope(|s| { + s.spawn(|| { + std::thread::sleep(Duration::from_millis(800)); + cancelled.store(true, Ordering::Release) + }); + client + .execute_controlled(spec.clone(), &publication, None, Some(&cancelled)) + .unwrap() + }); + assert_eq!(result.report.cleanup, CleanupState::Complete); + assert_eq!( + result + .report + .classify(&spec.validate(&Default::default()).unwrap()) + .unwrap(), + ExecutionOutcome::Cancelled + ); + println!( + "KF_FACTORY_CONTROL_PROOF {}", + serde_json::json!({"kind":"cancel","run_id":run_id,"pid":pid,"report":result.report}) + ); + } + "replay" => { + assert!(client.execute(spec, &publication).is_err()); + println!( + "KF_FACTORY_CONTROL_PROOF {}", + serde_json::json!({"kind":"replay","run_id":run_id,"refused":true}) + ); + } + "capacity" => { + let mut first = UnixStream::connect(&socket).unwrap(); + raw_send(&mut first, &serde_json::json!({"run_id":run_id})); + let ready = raw_receive(&mut first); + assert_eq!(ready["event"], "ready"); + let mut second = UnixStream::connect(&socket).unwrap(); + raw_send(&mut second, &serde_json::json!({"run_id":"capacity-extra"})); + assert_eq!(raw_receive(&mut second)["event"], "refused"); + drop(first); + drop(second); + println!( + "KF_FACTORY_CONTROL_PROOF {}", + serde_json::json!({"kind":"capacity","slot_limit":1,"extra_refused":true,"queued_disconnect":true}) + ); + } + "extra" => { + use rustix::fd::AsFd; + use rustix::net::{SendAncillaryBuffer, SendAncillaryMessage, SendFlags}; + let mut stream = UnixStream::connect(&socket).unwrap(); + raw_send(&mut stream, &serde_json::json!({"run_id":run_id})); + assert_eq!(raw_receive(&mut stream)["event"], "ready"); + raw_send(&mut stream, &serde_json::json!({"spec":spec})); + let mut space = [std::mem::MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(1))]; + let borrowed = [publication.as_fd()]; + let mut ancillary = SendAncillaryBuffer::new(&mut space); + assert!(ancillary.push(SendAncillaryMessage::ScmRights(&borrowed))); + assert_eq!( + rustix::net::sendmsg( + &stream, + &[std::io::IoSlice::new(&[0xa5])], + &mut ancillary, + SendFlags::NOSIGNAL + ) + .unwrap(), + 1 + ); + raw_send( + &mut stream, + &serde_json::json!({"run_id":"unrequested-frame"}), + ); + let reply = raw_receive(&mut stream); + assert_eq!(reply["event"], "completed"); + assert!(!reply["report"]["errors"].as_array().unwrap().is_empty()); + assert_eq!(reply["report"]["stdout"]["retained_bytes"], 0); + assert_eq!(reply["report"]["cleanup"], "complete"); + println!( + "KF_FACTORY_CONTROL_PROOF {}", + serde_json::json!({"kind":"extra_frame","refused_before_user_exec":true,"report":reply["report"]}) + ); + } + "binding" => { + let mut stream = UnixStream::connect(&socket).unwrap(); + raw_send(&mut stream, &serde_json::json!({"run_id":run_id})); + assert_eq!(raw_receive(&mut stream)["event"], "ready"); + spec.run_id = "wrong-bound-run".into(); + raw_send(&mut stream, &serde_json::json!({"spec":spec})); + let mut buf = [0]; + assert_eq!(stream.read(&mut buf).unwrap_or(0), 0); + println!( + "KF_FACTORY_CONTROL_PROOF {}", + serde_json::json!({"kind":"binding","wrong_run_refused":true}) + ); + } + _ => panic!("unknown fixed probe"), + } +} +fn raw_send(s: &mut UnixStream, v: &serde_json::Value) { + s.set_read_timeout(Some(Duration::from_secs(10))).unwrap(); + let b = serde_json::to_vec(v).unwrap(); + s.write_all(&[b'K', b'F', 1, 0]).unwrap(); + s.write_all(&(b.len() as u32).to_be_bytes()).unwrap(); + s.write_all(&b).unwrap(); +} +fn raw_receive(s: &mut UnixStream) -> serde_json::Value { + let mut head = [0; 8]; + s.read_exact(&mut head).unwrap(); + let n = u32::from_be_bytes(head[4..].try_into().unwrap()) as usize; + assert!(n <= 1024); + let mut b = vec![0; n]; + s.read_exact(&mut b).unwrap(); + serde_json::from_slice(&b).unwrap() +} +#[test] +#[ignore = "independent capped VM: real compile-once C17/CPP20, 3 isolated points and structured CE/WA/RE"] +fn compilation_and_three_points() { + let mut f = Fixture::new(); + f.start(); + for language in ["c17", "cpp20"] { + let j = f.job( + language, + SOURCE, + &[b"", &[0, 255, 10], b"abc\n"], + &[b"", &[0, 255, 10], b"abc\n"], + ); + let v = f.run(&j, "accepted"); + let stages = v["stages"].as_array().unwrap(); + assert_eq!(stages.len(), 4); + let mut pids = std::collections::HashSet::new(); + for s in stages { + assert!(pids.insert(s["helper_pid"].as_u64().unwrap())); + assert_eq!(s["report"]["cleanup"], "complete"); + } + assert_eq!(v["result"]["report"]["cases"].as_array().unwrap().len(), 3); + assert_eq!(v["result"]["report"]["compilation"]["status"], "succeeded"); + } + for (source, input, answer, verdict) in [ + ( + b"invalid C source".as_slice(), + b"".as_slice(), + b"".as_slice(), + "compilation_error", + ), + (SOURCE, b"binary\n", b"binary", "wrong_answer"), + (SOURCE, b"#", b"", "runtime_error"), + ] { + let j = f.job( + "c17", + source, + &[input, input, input], + &[answer, answer, answer], + ); + let v = f.run(&j, verdict); + if verdict == "compilation_error" { + assert_eq!(v["stages"].as_array().unwrap().len(), 1); + assert!( + !v["result"]["report"]["compilation"]["diagnostics"] + .as_str() + .unwrap() + .is_empty() + ); + assert_eq!(v["result"]["skipped"].as_array().unwrap().len(), 3); + } + } + let j = f.job("c17", SOURCE, &[b"a", b"b", b"c"], &[b"a", b"b", b"c"]); + let mut job: serde_json::Value = serde_json::from_slice(&fs::read(&j).unwrap()).unwrap(); + job["manifest"]["compile_limits"]["cpu_time_us"] = 10_000_000.into(); + job["manifest"]["compile_limits"]["wall_time_us"] = 1000.into(); + let m: kf_contracts::judging::JobManifest = + serde_json::from_value(job["manifest"].clone()).unwrap(); + job["manifest_digest"] = format!( + "{:x}", + Sha256::digest(kf_contracts::judging::manifest_bytes(&m).unwrap()) + ) + .into(); + fs::write(&j, serde_json::to_vec(&job).unwrap()).unwrap(); + worker_file(&j); + let v = f.run(&j, "compilation_error"); + assert_eq!( + v["result"]["report"]["compilation"]["status"], + "compilation_limit" + ); + assert_eq!(v["stages"].as_array().unwrap().len(), 1); +} +#[test] +#[ignore = "independent capped VM: actual capacity/replay/binding/cancel and factory crash recovery"] +fn factory_lifecycle_and_refusal() { + let mut f = Fixture::new(); + f.start(); + f.control_inputs(); + for (mode, run) in [ + ("capacity", "capacity-one"), + ("binding", "binding-one"), + ("extra", "extra-one"), + ("cancel", "cancel-one"), + ("replay", "cancel-one"), + ] { + f.probe(mode, run); + f.wait_empty(); + std::thread::sleep(Duration::from_millis(30)); + } + let mut root_client = UnixStream::connect(&f.socket).unwrap(); + let body = serde_json::to_vec(&serde_json::json!({"run_id":"root-forged-peer"})).unwrap(); + let mut frame = vec![b'K', b'F', 1, 0]; + frame.extend((body.len() as u32).to_be_bytes()); + frame.extend(body); + root_client + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + if let Err(e) = root_client.write_all(&frame) { + assert!(matches!( + e.kind(), + std::io::ErrorKind::BrokenPipe | std::io::ErrorKind::ConnectionReset + )); + } + let mut byte = [0u8; 1]; + match root_client.read(&mut byte) { + Ok(0) => {} + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::ConnectionReset | std::io::ErrorKind::BrokenPipe + ) => {} + other => panic!("unexpected root peer result {other:?}"), + } + drop(root_client); + println!("KF_FACTORY_AUTH_PROOF wrong_uid_valid_request_refused=1"); + std::thread::sleep(Duration::from_millis(30)); + assert!(f.factory.as_mut().unwrap().try_wait().unwrap().is_none()); + let job = f.job("c17", SOURCE, &[b"!", b"a", b"b"], &[b"live", b"a", b"b"]); + let worker = f.worker(&job); + f.live(); + let mut old = f.factory.take().unwrap(); + old.kill().unwrap(); + old.wait().unwrap(); + let output = worker.wait_with_output().unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let v: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(v["result"]["report"]["verdict"], "cancelled", "{v}"); + assert_eq!(v["result"]["report"]["cleanup_complete"], true); + assert_eq!(v["result"]["report"]["cases"].as_array().unwrap().len(), 1); + assert_eq!(v["result"]["skipped"].as_array().unwrap().len(), 2); + println!("KF_FACTORY_RESTART_PROOF {v}"); + f.wait_empty(); + f.start(); + f.probe("replay", "cancel-one"); + let j = f.job("c17", SOURCE, &[b"a", b"b", b"c"], &[b"a", b"b", b"c"]); + f.run(&j, "accepted"); +} + +#[test] +#[ignore = "independent capped VM: disconnect, initialization failure, retained unknown lease and foreign socket"] +fn faults_preserve_foreign_resources_and_recover_owned() { + let mut f = Fixture::new(); + f.start(); + // Kernel-authenticated Worker disappears while actual user code is live. + let j = f.job("c17", SOURCE, &[b"!", b"a", b"b"], &[b"live", b"a", b"b"]); + let mut worker = f.worker(&j); + f.live(); + worker.kill().unwrap(); + worker.wait().unwrap(); + f.wait_empty(); + println!( + "KF_FACTORY_FAULT_PROOF {}", + serde_json::json!({"kind":"worker_disconnect","owned_resources_zero":true}) + ); + // Root-owned image publication is invalidated before preparation. No exec. + let work = PathBuf::from(&f.root.config.image_root).join("work"); + let moved = work.with_file_name("work-operator-held"); + fs::rename(&work, &moved).unwrap(); + let j = f.job("c17", SOURCE, &[b"a", b"b", b"c"], &[b"a", b"b", b"c"]); + let o = f.worker(&j).wait_with_output().unwrap(); + assert!(o.status.success()); + let v: serde_json::Value = serde_json::from_slice(&o.stdout).unwrap(); + assert_eq!(v["result"]["report"]["verdict"], "system_error"); + assert!( + v["result"]["report"]["cases"] + .as_array() + .unwrap() + .is_empty() + ); + fs::rename(&moved, &work).unwrap(); + f.wait_empty(); + println!( + "KF_FACTORY_FAULT_PROOF {}", + serde_json::json!({"kind":"initialization","no_test_points_started":true,"owned_resources_zero":true}) + ); + // Unknown identity-slot entry prevents UID release/reuse even after tree stop. + let j = f.job("c17", SOURCE, &[b"!", b"a", b"b"], &[b"live", b"a", b"b"]); + let mut worker = f.worker(&j); + f.live(); + let slot = PathBuf::from("/var/lib/kf-judge/identities/uid-100000"); + let foreign = slot.join("foreign-owner-content"); + fs::write(&foreign, b"preserve me").unwrap(); + worker.kill().unwrap(); + worker.wait().unwrap(); + let deadline = Instant::now() + Duration::from_secs(15); + loop { + if let Some(status) = f.factory.as_mut().unwrap().try_wait().unwrap() { + assert!(!status.success()); + break; + } + assert!(Instant::now() < deadline, "fault factory did not stop"); + std::thread::sleep(Duration::from_millis(5)); + } + assert_eq!(fs::read(&foreign).unwrap(), b"preserve me"); + assert!(slot.exists()); + f.factory.take(); + println!( + "KF_FACTORY_FAULT_PROOF {}", + serde_json::json!({"kind":"cleanup_failure","foreign_preserved":true,"uid_lease_retained":true,"factory_stopped":true}) + ); + fs::remove_file(&foreign).unwrap(); + f.start(); + f.wait_empty(); + let j = f.job("c17", SOURCE, &[b"a", b"b", b"c"], &[b"a", b"b", b"c"]); + f.run(&j, "accepted"); + // Stale socket recovery is inode-bound; replacement content is never adopted. + let mut child = f.factory.take().unwrap(); + child.kill().unwrap(); + child.wait().unwrap(); + let saved = f.socket.with_file_name("operator-held.sock"); + fs::rename(&f.socket, &saved).unwrap(); + fs::write(&f.socket, b"foreign socket replacement").unwrap(); + let o = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap()) + .env_clear() + .args(["factory", "--config"]) + .arg(&f.config) + .args(["--lifetime-seconds", "1"]) + .output() + .unwrap(); + assert_eq!(o.status.code(), Some(2)); + assert_eq!(fs::read(&f.socket).unwrap(), b"foreign socket replacement"); + fs::remove_file(&f.socket).unwrap(); + fs::rename(saved, &f.socket).unwrap(); + f.start(); + f.wait_empty(); + println!( + "KF_FACTORY_FAULT_PROOF {}", + serde_json::json!({"kind":"socket_replacement","foreign_preserved":true,"owned_restart_recovered":true}) + ); +} diff --git a/judge/crates/worker/Cargo.toml b/judge/crates/worker/Cargo.toml index daed9a8..79de4df 100644 --- a/judge/crates/worker/Cargo.toml +++ b/judge/crates/worker/Cargo.toml @@ -12,6 +12,7 @@ kf-engine.workspace = true kf-sandbox.workspace = true clap.workspace = true serde_json.workspace = true +sha2 = "=0.10.9" [lints] workspace = true diff --git a/judge/crates/worker/src/lib.rs b/judge/crates/worker/src/lib.rs index ec734d1..5024723 100644 --- a/judge/crates/worker/src/lib.rs +++ b/judge/crates/worker/src/lib.rs @@ -111,3 +111,6 @@ pub mod native { kf_engine::judge(&mut port, &[case]).map_err(|_| io::Error::other("invalid judge request")) } } + +#[cfg(target_os = "linux")] +pub mod pipeline; diff --git a/judge/crates/worker/src/main.rs b/judge/crates/worker/src/main.rs index 98881ca..7828cf6 100644 --- a/judge/crates/worker/src/main.rs +++ b/judge/crates/worker/src/main.rs @@ -14,6 +14,16 @@ struct Cli { } #[derive(Subcommand)] enum Command { + JudgeSubmission { + #[arg(long)] + socket: PathBuf, + #[arg(long)] + job: PathBuf, + #[arg(long)] + artifacts: PathBuf, + #[arg(long)] + staging: PathBuf, + }, JudgeOnce { #[arg(long)] socket: PathBuf, @@ -33,6 +43,31 @@ fn main() -> ExitCode { return ExitCode::SUCCESS; } match cli.command { + Some(Command::JudgeSubmission { + socket, + job, + artifacts, + staging, + }) => { + #[cfg(target_os = "linux")] + { + match kf_worker::pipeline::judge_local(&socket, &job, &artifacts, &staging) { + Ok(result) => { + println!("{}", serde_json::json!(result)); + ExitCode::SUCCESS + } + Err(_) => { + eprintln!("submission_failed"); + ExitCode::from(2) + } + } + } + #[cfg(not(target_os = "linux"))] + { + let _ = (socket, job, artifacts, staging); + ExitCode::from(2) + } + } Some(Command::JudgeOnce { socket, job, diff --git a/judge/crates/worker/src/pipeline.rs b/judge/crates/worker/src/pipeline.rs new file mode 100644 index 0000000..4121b43 --- /dev/null +++ b/judge/crates/worker/src/pipeline.rs @@ -0,0 +1,504 @@ +//! Ordinary adapter: fixed GNU command profiles, immutable digest verification, +//! bounded point-at-a-time data and injected Engine ports. No database or sudo. +use kf_contracts::judging::*; +use kf_engine::{ExecutionKind, Observation, submission::*}; +use kf_sandbox::{ + artifacts::{ArtifactRoot, SnapshotFile, StagePublication}, + factory::FactoryClient, + outcome::{ExecutionOutcome, LimitKind, Measurement}, + spec::{ExecutionProfile, ExecutionSpec, InputArtifact, ResourceLimits}, +}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use std::{ + io::{self, Read}, + path::{Path, PathBuf}, + sync::atomic::{AtomicBool, Ordering}, + time::Instant, +}; +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Job { + pub manifest: JobManifest, + pub binding: LeaseBinding, + pub manifest_digest: Digest, +} +#[derive(Serialize)] +pub struct StageEvidence { + pub run_id: String, + pub helper_pid: u32, + pub report: kf_sandbox::outcome::ExecutionReport, +} +#[derive(Serialize)] +pub struct Judged { + pub result: SubmissionResult, + pub stages: Vec, +} +pub trait ArtifactPort { + fn read(&mut self, reference: &ArtifactReference) -> io::Result>; +} +pub struct LocalArtifacts { + root: ArtifactRoot, +} +impl LocalArtifacts { + pub fn open(path: &Path) -> io::Result { + Ok(Self { + root: ArtifactRoot::open(path)?, + }) + } +} +impl ArtifactPort for LocalArtifacts { + fn read(&mut self, r: &ArtifactReference) -> io::Result> { + let s = self + .root + .snapshot_artifact(r.digest.as_str(), r.bytes.max(1))?; + if s.bytes() != r.bytes { + return Err(io::Error::other("artifact byte count")); + } + let mut data = Vec::new(); + s.reader()?.take(r.bytes + 1).read_to_end(&mut data)?; + verify(r, &data)?; + Ok(data) + } +} +fn digest(data: &[u8]) -> Digest { + format!("{:x}", Sha256::digest(data)) + .try_into() + .expect("SHA256 representation") +} +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")); + } + Ok(()) +} +fn stage_error(code: FailureCode, cleanup_complete: bool) -> StageFailure { + StageFailure { + code, + cleanup_complete, + } +} +fn limits(r: &RunLimits) -> ResourceLimits { + ResourceLimits { + cpu_time_us: r.cpu_time_us, + wall_time_us: r.wall_time_us, + memory_bytes: r.memory_bytes, + stack_bytes: r.stack_bytes, + processes: r.processes, + open_files: r.open_files, + stdout_bytes: r.stdout_bytes, + stderr_bytes: r.stderr_bytes, + file_bytes: r.file_bytes, + workdir_bytes: r.workdir_bytes, + } +} +fn metric(m: &Measurement) -> Option { + match m { + Measurement::Measured { value } => Some(*value), + Measurement::Unavailable { .. } => None, + } +} +fn kind(o: ExecutionOutcome) -> ExecutionKind { + match o { + ExecutionOutcome::Exited { code } => ExecutionKind::Exited { code }, + ExecutionOutcome::Signaled { .. } => ExecutionKind::Signaled, + ExecutionOutcome::LimitExceeded { + primary: LimitKind::Memory, + } => ExecutionKind::MemoryLimit, + ExecutionOutcome::LimitExceeded { + primary: LimitKind::CpuTime | LimitKind::WallTime, + } => ExecutionKind::TimeLimit, + ExecutionOutcome::LimitExceeded { + primary: LimitKind::Stdout | LimitKind::Stderr, + } => ExecutionKind::OutputLimit, + ExecutionOutcome::LimitExceeded { .. } => ExecutionKind::OtherLimit, + ExecutionOutcome::Cancelled => ExecutionKind::Cancelled, + ExecutionOutcome::InfrastructureFailure | ExecutionOutcome::NotStarted => { + ExecutionKind::InfrastructureFailure + } + } +} +pub fn compiler_spec( + language: Language, + source: String, + run_id: String, + budget: &RunLimits, +) -> ExecutionSpec { + let (profile, compiler, standard, name) = match language { + Language::C17 => ( + ExecutionProfile::CompileC, + "/usr/bin/gcc", + "-std=c17", + "main.c", + ), + Language::Cpp20 => ( + ExecutionProfile::CompileCpp, + "/usr/bin/g++", + "-std=c++20", + "main.cpp", + ), + }; + ExecutionSpec { + schema_version: kf_contracts::SchemaVersion, + run_id, + profile, + argv: vec![ + compiler.into(), + standard.into(), + "-O2".into(), + "-pipe".into(), + format!("/work/{name}"), + "-o".into(), + "/work/main".into(), + ], + env: Default::default(), + cwd: ".".into(), + inputs: vec![InputArtifact { + artifact_id: source, + path: name.into(), + }], + stdin_artifact: None, + outputs: vec!["main".into()], + limits: limits(budget), + } +} +pub struct Attempt<'a, A: ArtifactPort> { + pub artifacts: A, + pub stages: Vec, + pub cancellation: &'a AtomicBool, + factory: FactoryClient, + staging: PathBuf, + started: Instant, + deadline: Instant, + program: Option<(ArtifactReference, SnapshotFile)>, +} +impl<'a, A: ArtifactPort> Attempt<'a, A> { + pub fn new( + artifacts: A, + factory_socket: PathBuf, + staging: PathBuf, + wall_us: u64, + cancellation: &'a AtomicBool, + ) -> io::Result { + if wall_us == 0 || wall_us > MAX_JOB_WALL_US { + return Err(io::Error::other("job deadline")); + } + StagePublication::recover(&staging)?; + let started = Instant::now(); + let deadline = started + .checked_add(std::time::Duration::from_micros(wall_us)) + .ok_or_else(|| io::Error::other("deadline overflow"))?; + Ok(Self { + artifacts, + stages: vec![], + cancellation, + factory: FactoryClient::new(factory_socket), + staging, + started, + deadline, + program: None, + }) + } + fn execute( + &mut self, + spec: ExecutionSpec, + publication: &mut StagePublication, + ) -> Result { + let id = spec.run_id.clone(); + let validated = spec + .clone() + .validate(&kf_sandbox::spec::SpecPolicy::default()) + .map_err(|_| stage_error(FailureCode::SandboxUnavailable, true))?; + let result = self.factory.execute_controlled( + spec, + publication.root(), + Some(self.deadline), + Some(self.cancellation), + ); + let cleanup = publication.cleanup().is_ok(); + let (result, pid) = + result.map_err(|_| stage_error(FailureCode::SandboxUnavailable, false))?; + if !cleanup { + return Err(stage_error(FailureCode::CleanupFailed, false)); + } + result + .report + .classify(&validated) + .map_err(|_| stage_error(FailureCode::SandboxUnavailable, false))?; + self.stages.push(StageEvidence { + run_id: id, + helper_pid: pid, + report: result.report.clone(), + }); + Ok(result) + } +} +impl CompilationPort for Attempt<'_, A> { + fn compile(&mut self, m: &JobManifest, run_id: &str) -> Result { + if m.toolchain.image_revision != 1 || m.toolchain.toolchain_id.as_str() != "gnu-cxx-v1" { + return Err(stage_error(FailureCode::SandboxUnavailable, true)); + } + let source = self + .artifacts + .read(&m.source) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + verify(&m.source, &source) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + let mut publication = StagePublication::create(&self.staging, run_id) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + publication + .put("source", &mut source.as_slice(), source.len() as u64) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, false))?; + drop(source); + let spec = compiler_spec( + m.language, + "source".into(), + run_id.into(), + &m.compile_limits, + ); + let validated = spec + .clone() + .validate(&kf_sandbox::spec::SpecPolicy::default()) + .map_err(|_| stage_error(FailureCode::SandboxUnavailable, true))?; + let result = self.execute(spec, &mut publication)?; + let status = match result + .report + .classify(&validated) + .map_err(|_| stage_error(FailureCode::SandboxUnavailable, false))? + { + ExecutionOutcome::Exited { code: 0 } => CompilationStatus::Succeeded, + ExecutionOutcome::Exited { .. } | ExecutionOutcome::Signaled { .. } => { + CompilationStatus::CompilationError + } + ExecutionOutcome::LimitExceeded { .. } => CompilationStatus::CompilationLimit, + ExecutionOutcome::Cancelled => CompilationStatus::Cancelled, + _ => CompilationStatus::SystemError, + }; + let mut raw = Vec::new(); + result + .stderr + .reader() + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))? + .take(65537) + .read_to_end(&mut raw) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + 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; + } + let program = if status == CompilationStatus::Succeeded { + if result.outputs.len() != 1 + || result.outputs[0].path != "main" + || result.outputs[0].file.bytes() == 0 + { + return Err(stage_error(FailureCode::ArtifactUnavailable, true)); + } + let output = result.outputs.into_iter().next().expect("one output").file; + let mut h = Sha256::new(); + let mut reader = output + .reader() + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + let mut buffer = [0u8; 16384]; + loop { + let n = reader + .read(&mut buffer) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + if n == 0 { + break; + } + h.update(&buffer[..n]); + } + let r = ArtifactReference { + digest: format!("{:x}", h.finalize()).try_into().expect("SHA256"), + bytes: output.bytes(), + }; + self.program = Some((r.clone(), output)); + Some(r) + } else { + None + }; + Ok(Compiled { + report: CompilationReport { + status, + program, + diagnostics: text.try_into().expect("bounded diagnostics"), + diagnostics_truncated: truncated, + cpu_time_us: metric(&result.report.usage.cpu_time_us), + memory_peak_bytes: metric(&result.report.usage.memory_peak_bytes), + }, + cleanup_complete: true, + failure: (status == CompilationStatus::SystemError) + .then_some(FailureCode::SandboxUnavailable), + }) + } +} +impl RunPort for Attempt<'_, A> { + fn run( + &mut self, + p: &ArtifactReference, + c: &TestCase, + b: &RunLimits, + run_id: &str, + ) -> Result { + let input = self + .artifacts + .read(&c.input) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + verify(&c.input, &input) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + let Some((reference, program)) = &self.program else { + return Err(stage_error(FailureCode::ArtifactUnavailable, true)); + }; + if reference != p { + return Err(stage_error(FailureCode::ArtifactUnavailable, true)); + } + let mut publication = StagePublication::create(&self.staging, run_id) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + publication + .put( + "program", + &mut program + .reader() + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?, + p.bytes, + ) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, false))?; + publication + .put("stdin", &mut input.as_slice(), input.len() as u64) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, false))?; + drop(input); + let spec = ExecutionSpec { + schema_version: kf_contracts::SchemaVersion, + run_id: run_id.into(), + profile: ExecutionProfile::Native, + argv: vec!["/work/main".into()], + env: Default::default(), + cwd: ".".into(), + inputs: vec![InputArtifact { + artifact_id: "program".into(), + path: "main".into(), + }], + stdin_artifact: Some("stdin".into()), + outputs: vec![], + limits: limits(b), + }; + let validated = spec + .clone() + .validate(&kf_sandbox::spec::SpecPolicy::default()) + .map_err(|_| stage_error(FailureCode::SandboxUnavailable, true))?; + let result = self.execute(spec, &mut publication)?; + let outcome = result + .report + .classify(&validated) + .map_err(|_| stage_error(FailureCode::SandboxUnavailable, false))?; + let mut stdout = Vec::new(); + result + .stdout + .reader() + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))? + .take(b.stdout_bytes + 1) + .read_to_end(&mut stdout) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + if stdout.len() as u64 > b.stdout_bytes { + return Err(stage_error(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 ExpectedPort for Attempt<'_, A> { + fn expected(&mut self, c: &TestCase) -> Result, StageFailure> { + let data = self + .artifacts + .read(&c.expected) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + verify(&c.expected, &data) + .map_err(|_| stage_error(FailureCode::ArtifactUnavailable, true))?; + Ok(data) + } +} +impl ControlPort for Attempt<'_, A> { + fn elapsed_us(&self) -> u64 { + self.started + .elapsed() + .as_micros() + .try_into() + .unwrap_or(u64::MAX) + } + fn cancelled(&self) -> bool { + self.cancellation.load(Ordering::Acquire) + } + fn stage(&mut self, _: Stage, _: Option) -> Result<(), StageFailure> { + Ok(()) + } +} +pub fn judge_local( + socket: &Path, + job: &Path, + artifacts: &Path, + staging: &Path, +) -> io::Result { + let job: Job = serde_json::from_slice(&kf_sandbox::execution::read_bounded_file(job, 262144)?) + .map_err(|_| io::Error::other("invalid submission"))?; + let canonical = + manifest_bytes(&job.manifest).map_err(|_| io::Error::other("invalid manifest"))?; + if digest(&canonical) != job.manifest_digest { + return Err(io::Error::other("manifest digest mismatch")); + } + let cancel = AtomicBool::new(false); + let mut adapter = Attempt::new( + LocalArtifacts::open(artifacts)?, + socket.into(), + staging.into(), + job.manifest.job_wall_time_us, + &cancel, + )?; + let result = judge_submission( + &mut adapter, + &job.manifest, + job.binding, + job.manifest_digest, + ) + .map_err(io::Error::other)?; + Ok(Judged { + result, + stages: adapter.stages, + }) +} +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn fixed_commands_and_source_digest() { + let m: JobManifest = serde_json::from_str(include_str!( + "../../../../tests/fixtures/contracts/judge-manifest.valid.json" + )) + .unwrap(); + for (lang, tool, std) in [ + (Language::C17, "/usr/bin/gcc", "-std=c17"), + (Language::Cpp20, "/usr/bin/g++", "-std=c++20"), + ] { + let s = compiler_spec(lang, "source".into(), "compile".into(), &m.compile_limits); + assert_eq!(s.argv[0], tool); + assert_eq!(s.argv[1], std); + assert_eq!(s.outputs, ["main"]); + assert!(s.validate(&kf_sandbox::spec::SpecPolicy::default()).is_ok()); + } + let r = ArtifactReference { + digest: digest(b"a"), + bytes: 1, + }; + assert!(verify(&r, b"b").is_err()); + assert!(verify(&r, b"a").is_ok()); + } +}