feat(judge): compile once and run isolated points through helper factory
This commit is contained in:
@@ -4,8 +4,8 @@
|
||||
前端使用 Vue 3、TypeScript、Element Plus 和 Vite。
|
||||
|
||||
当前版本为 **0.0.1**：沙箱已有受控隔离、C/C++ 编译及 native 执行；平台已实现真实身份、
|
||||
不可变题目、工件及有效租约读取基础。完整提交投递、协调器调度、多测试点评测与前端业务
|
||||
接入仍在开发，不能将前端预览成绩当作实际结果。
|
||||
不可变题目、工件及有效租约读取基础。受控 helper 工厂与本地 C/C++ 编译一次、多测试点评测
|
||||
已经接通；完整提交投递、协调器调度与前端业务接入仍在开发，不能将前端预览成绩当作实际结果。
|
||||
|
||||
## 工程划分
|
||||
|
||||
|
||||
@@ -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 库。
|
||||
|
||||
@@ -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
|
||||
Generated
+73
@@ -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"
|
||||
|
||||
+29
-2
@@ -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。
|
||||
|
||||
## 监督和恢复
|
||||
|
||||
|
||||
@@ -118,7 +118,7 @@ pub fn judge_submission<P: CompilationPort + RunPort + ExpectedPort + ControlPor
|
||||
};
|
||||
let mut cleanup_complete;
|
||||
let mut failure_code;
|
||||
let compilation = match compilation {
|
||||
let mut compilation = match compilation {
|
||||
Ok(c) => {
|
||||
c.report.validate()?;
|
||||
cleanup_complete = c.cleanup_complete;
|
||||
@@ -142,6 +142,13 @@ pub fn judge_submission<P: CompilationPort + RunPort + ExpectedPort + ControlPor
|
||||
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 {
|
||||
@@ -175,13 +182,18 @@ pub fn judge_submission<P: CompilationPort + RunPort + ExpectedPort + ControlPor
|
||||
}
|
||||
}
|
||||
Ok(o) => {
|
||||
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<P: CompilationPort + RunPort + ExpectedPort + ControlPor
|
||||
}
|
||||
},
|
||||
}
|
||||
} else {
|
||||
classify(&o, &[])
|
||||
};
|
||||
if verdict == Verdict::SystemError && failure_code.is_none() {
|
||||
failure_code = Some(FailureCode::SandboxUnavailable);
|
||||
@@ -306,6 +316,7 @@ mod tests {
|
||||
expected_calls: usize,
|
||||
cancel_after: usize,
|
||||
clock: u64,
|
||||
advance_us: u64,
|
||||
fail_compile: bool,
|
||||
}
|
||||
impl CompilationPort for Port {
|
||||
@@ -353,6 +364,7 @@ mod tests {
|
||||
impl ControlPort for Port {
|
||||
fn elapsed_us(&self) -> 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,3 +29,6 @@ workspace = true
|
||||
[[example]]
|
||||
name = "identity_probe"
|
||||
test = false
|
||||
|
||||
[dev-dependencies]
|
||||
sha2 = "=0.10.9"
|
||||
|
||||
@@ -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<StageEntry>,
|
||||
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<u64> {
|
||||
value
|
||||
.rsplit_once(") ")
|
||||
.and_then(|(_, s)| s.split_whitespace().nth(19))
|
||||
.and_then(|s| s.parse::<u64>().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<Self> {
|
||||
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<u32> {
|
||||
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::<io::Result<_>>()?;
|
||||
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::<StageEntry>(l)
|
||||
.map_err(|_| refused("stage entry journal"))
|
||||
})
|
||||
.collect::<io::Result<Vec<_>>>()?;
|
||||
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::<io::Result<_>>()?;
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -77,7 +77,7 @@ fn read_pinned(fd: OwnedFd, limit: u64) -> io::Result<Vec<u8>> {
|
||||
Ok(bytes)
|
||||
}
|
||||
pub fn read_bounded_file(path: &Path, limit: u64) -> io::Result<Vec<u8>> {
|
||||
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(
|
||||
|
||||
@@ -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<Instant>,
|
||||
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>()? {
|
||||
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<File> {
|
||||
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<Active> = 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::<Acquire>() 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::<Acquire>(data);
|
||||
assert!(decoded.is_err() || !decoded.unwrap().valid());
|
||||
}
|
||||
assert!(
|
||||
serde_json::from_str::<Acquire>(r#"{"run_id":"a-1"}"#)
|
||||
.unwrap()
|
||||
.valid()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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<LimitObservation>) {
|
||||
}
|
||||
}
|
||||
|
||||
fn peer_closed(fd: &rustix::fd::OwnedFd) -> io::Result<bool> {
|
||||
pub(crate) fn peer_closed(fd: &rustix::fd::OwnedFd) -> io::Result<bool> {
|
||||
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<bool> {
|
||||
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) {
|
||||
|
||||
@@ -33,3 +33,6 @@ pub mod execution;
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub mod rpc;
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub mod factory;
|
||||
|
||||
@@ -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")]
|
||||
|
||||
@@ -64,7 +64,7 @@ impl WireMessage for Reply {
|
||||
}
|
||||
}
|
||||
}
|
||||
fn channel(stream: UnixStream, uid: u32, gid: u32) -> io::Result<ControlChannel> {
|
||||
pub(crate) fn channel(stream: UnixStream, uid: u32, gid: u32) -> io::Result<ControlChannel> {
|
||||
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<Executed> {
|
||||
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<serde_json::Va
|
||||
}
|
||||
};
|
||||
drop(listener);
|
||||
serve_stream(&config, stream, None, None)
|
||||
}
|
||||
pub(crate) fn serve_stream(
|
||||
config: &crate::runtime::ValidatedRuntimeConfig,
|
||||
stream: UnixStream,
|
||||
binding: Option<&str>,
|
||||
guardian: Option<&OwnedFd>,
|
||||
) -> io::Result<serde_json::Value> {
|
||||
let ids = &config.config().identities;
|
||||
let mut c = channel(stream, ids.worker_uid, ids.worker_gid)?;
|
||||
let request = c.receive::<Request>()?;
|
||||
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 {
|
||||
|
||||
@@ -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 <stdio.h>
|
||||
#include <unistd.h>
|
||||
#include <fcntl.h>
|
||||
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<Child>,
|
||||
}
|
||||
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::<serde_json::Map<_, _>>()
|
||||
.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::<serde_json::Map<_, _>>()
|
||||
.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})
|
||||
);
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<StageEvidence>,
|
||||
}
|
||||
pub trait ArtifactPort {
|
||||
fn read(&mut self, reference: &ArtifactReference) -> io::Result<Vec<u8>>;
|
||||
}
|
||||
pub struct LocalArtifacts {
|
||||
root: ArtifactRoot,
|
||||
}
|
||||
impl LocalArtifacts {
|
||||
pub fn open(path: &Path) -> io::Result<Self> {
|
||||
Ok(Self {
|
||||
root: ArtifactRoot::open(path)?,
|
||||
})
|
||||
}
|
||||
}
|
||||
impl ArtifactPort for LocalArtifacts {
|
||||
fn read(&mut self, r: &ArtifactReference) -> io::Result<Vec<u8>> {
|
||||
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<u64>) -> Option<u64> {
|
||||
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<StageEvidence>,
|
||||
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<Self> {
|
||||
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<kf_sandbox::rpc::Executed, StageFailure> {
|
||||
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<A: ArtifactPort> CompilationPort for Attempt<'_, A> {
|
||||
fn compile(&mut self, m: &JobManifest, run_id: &str) -> Result<Compiled, StageFailure> {
|
||||
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<A: ArtifactPort> RunPort for Attempt<'_, A> {
|
||||
fn run(
|
||||
&mut self,
|
||||
p: &ArtifactReference,
|
||||
c: &TestCase,
|
||||
b: &RunLimits,
|
||||
run_id: &str,
|
||||
) -> Result<Observation, StageFailure> {
|
||||
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<A: ArtifactPort> ExpectedPort for Attempt<'_, A> {
|
||||
fn expected(&mut self, c: &TestCase) -> Result<Vec<u8>, 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<A: ArtifactPort> 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<u16>) -> Result<(), StageFailure> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
pub fn judge_local(
|
||||
socket: &Path,
|
||||
job: &Path,
|
||||
artifacts: &Path,
|
||||
staging: &Path,
|
||||
) -> io::Result<Judged> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user