diff --git a/judge/Cargo.lock b/judge/Cargo.lock index 4b825da..1587721 100644 --- a/judge/Cargo.lock +++ b/judge/Cargo.lock @@ -377,6 +377,7 @@ name = "kf-engine" version = "0.0.1" dependencies = [ "kf-contracts", + "serde", ] [[package]] @@ -400,6 +401,7 @@ dependencies = [ "kf-contracts", "kf-engine", "kf-sandbox", + "serde", "serde_json", ] diff --git a/judge/README.md b/judge/README.md index f75e75a..d96bcfb 100644 --- a/judge/README.md +++ b/judge/README.md @@ -33,6 +33,31 @@ kf-sandbox run --config /etc/kf-judge/runtime.json --spec request.json --artif 完成任务树终止、cgroup 空读回、owned 子进程回收、最终计量和导出后，才移除资源并释放 UID。stdout/stderr 和输出文件为不可变快照。普通文件入口按 run ID 排他发布 `stdout`、`stderr`、`artifact-N`，先写自身临时目录再原子安装，不覆盖已有目录；清理未完成时不发布。 +## Worker 和 Engine 的最小链路 + +服务管理器先在有限监督器 cgroup 内启动一个新的 `kf-sandbox serve-one --config /etc/kf-judge/runtime.json --socket /run/kf-judge/task.sock`。socket 的祖先须由 root 持有、不可被 Worker 写入并允许其搜索；socket 排他创建为 root:worker 的 0660，既有路径不覆盖。一次新进程只接收一次请求，等待连接和准备帧有绝对截止时间，不提供常驻消费循环。 + +普通用户运行： + +```sh +kf-worker judge-once --socket /run/kf-judge/task.sock \ + --job native-job.json --artifacts /srv/kf-worker/publication +``` + +本地单用例作业示例（不是平台任务通知协议）： + +```json +{"request":{"run_id":"case-001","program_artifact":"program-001","stdin_artifact":"input-001"},"expected_artifact":"expected-001"} +``` + +Worker 选取业务授权的自身发布目录，准备有界封印的期望输出，并向 Engine 注入 `ExecutionPort`。适配器构造 native 规格，仅绑定程序与 stdin，不将期望输出放入用户运行根。helper 根据 `SO_PEERCRED` 核验配置中的 Worker UID/GID；客户端核验 helper 的 root 凭据。有界私有帧与 `SCM_RIGHTS` 只传持有的目录能力和只读封印输出，收到的 FD 重新检查类型、长度、访问模式、封印和 CLOEXEC，不传任意宿主挂载路径。 + +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、不修改账号或权限。 + ## 监督和恢复 CPU 总时间与墙钟使用可信监督器和单调时钟；CPU 带宽仅用于节流。内存 OOM 和 pids 超限依据所属 cgroup 的真实事件及计量，不能凭 SIGKILL 推断 MLE。并发排空二进制 stdout/stderr，独立限额、截断和提前 stdin 关闭均保留实际证据。监督器最终报告保留计量不可用状态、全部限制观察、取消、退出状态和清理错误；比较与成绩属于 Engine。 diff --git a/judge/crates/engine/Cargo.toml b/judge/crates/engine/Cargo.toml index c725af3..7df5187 100644 --- a/judge/crates/engine/Cargo.toml +++ b/judge/crates/engine/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true [dependencies] kf-contracts.workspace = true +serde.workspace = true [lints] workspace = true diff --git a/judge/crates/engine/src/lib.rs b/judge/crates/engine/src/lib.rs index 6956846..8f23a1e 100644 --- a/judge/crates/engine/src/lib.rs +++ b/judge/crates/engine/src/lib.rs @@ -1,6 +1,229 @@ -//! Reserved for deterministic judging orchestration (JS-05/JS-06). -//! -//! Owns language adaptation, case order, output comparison and verdict aggregation. -//! Execution is supplied through a port; this crate must not access SQL, AMQP, -//! HTTP, privilege setup or hidden host paths directly. -//! No fake verdict or user-program execution is provided by this scaffold. +//! Deterministic comparison and aggregation. The injected port owns execution; +//! this crate has no OS, sandbox, privilege or storage dependency. +use serde::{Deserialize, Serialize}; +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CaseRequest { + pub run_id: String, + pub program_artifact: String, + pub stdin_artifact: Option, +} +pub struct Case { + pub request: CaseRequest, + pub expected: Vec, +} +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ExecutionKind { + Exited { code: u8 }, + Signaled, + TimeLimit, + MemoryLimit, + OutputLimit, + OtherLimit, + Cancelled, + InfrastructureFailure, +} +pub struct Observation { + pub kind: ExecutionKind, + pub stdout: Vec, + pub cpu_time_us: Option, + pub memory_peak_bytes: Option, +} +pub trait ExecutionPort { + fn execute(&mut self, request: &CaseRequest) -> Result; +} +#[derive(Debug)] +pub struct PortError; +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Verdict { + Accepted, + WrongAnswer, + RuntimeError, + TimeLimitExceeded, + MemoryLimitExceeded, + OutputLimitExceeded, + Cancelled, + SystemError, +} +#[derive(Debug, Serialize, Deserialize)] +pub struct CaseResult { + pub run_id: String, + pub verdict: Verdict, + pub cpu_time_us: Option, + pub memory_peak_bytes: Option, +} +#[derive(Debug, Serialize, Deserialize)] +pub struct Judgement { + pub verdict: Verdict, + pub cases: Vec, +} +pub fn judge(port: &mut impl ExecutionPort, cases: &[Case]) -> Result { + if cases.is_empty() || cases.len() > 256 { + return Err(PortError); + } + let mut seen = std::collections::HashSet::new(); + if cases.iter().any(|c| !seen.insert(&c.request.run_id)) { + return Err(PortError); + } + let mut results = Vec::with_capacity(cases.len()); + for c in cases { + let result = match port.execute(&c.request) { + Err(_) => CaseResult { + run_id: c.request.run_id.clone(), + verdict: Verdict::SystemError, + cpu_time_us: None, + memory_peak_bytes: None, + }, + Ok(o) => { + let verdict = match o.kind { + ExecutionKind::Exited { code: 0 } => { + if o.stdout == c.expected { + Verdict::Accepted + } else { + Verdict::WrongAnswer + } + } + ExecutionKind::Exited { .. } + | ExecutionKind::Signaled + | ExecutionKind::OtherLimit => Verdict::RuntimeError, + ExecutionKind::TimeLimit => Verdict::TimeLimitExceeded, + ExecutionKind::MemoryLimit => Verdict::MemoryLimitExceeded, + ExecutionKind::OutputLimit => Verdict::OutputLimitExceeded, + ExecutionKind::Cancelled => Verdict::Cancelled, + ExecutionKind::InfrastructureFailure => Verdict::SystemError, + }; + CaseResult { + run_id: c.request.run_id.clone(), + verdict, + cpu_time_us: o.cpu_time_us, + memory_peak_bytes: o.memory_peak_bytes, + } + } + }; + let stop = matches!(result.verdict, Verdict::SystemError | Verdict::Cancelled); + results.push(result); + if stop { + break; + } + } + let verdict = results + .iter() + .find(|r| matches!(r.verdict, Verdict::SystemError | Verdict::Cancelled)) + .or_else(|| results.iter().find(|r| r.verdict != Verdict::Accepted)) + .map_or(Verdict::Accepted, |r| r.verdict); + Ok(Judgement { + verdict, + cases: results, + }) +} +#[cfg(test)] +mod tests { + use super::*; + struct Port { + kinds: std::collections::VecDeque>, + calls: usize, + } + impl ExecutionPort for Port { + fn execute(&mut self, _: &CaseRequest) -> Result { + self.calls += 1; + Ok(Observation { + kind: self.kinds.pop_front().unwrap()?, + stdout: vec![0, 255, 10], + cpu_time_us: Some(42), + memory_peak_bytes: None, + }) + } + } + fn case(id: &str, expected: Vec) -> Case { + Case { + request: CaseRequest { + run_id: id.into(), + program_artifact: "program".into(), + stdin_artifact: None, + }, + expected, + } + } + #[test] + fn exact_binary_comparison_and_nonzero_precedence() { + let mut p = Port { + kinds: [ + Ok(ExecutionKind::Exited { code: 0 }), + Ok(ExecutionKind::Exited { code: 0 }), + Ok(ExecutionKind::Exited { code: 7 }), + ] + .into(), + calls: 0, + }; + let j = judge( + &mut p, + &[ + case("a", vec![0, 255, 10]), + case("b", vec![0, 255]), + case("c", vec![0, 255, 10]), + ], + ) + .unwrap(); + assert_eq!(j.verdict, Verdict::WrongAnswer); + assert_eq!( + j.cases.iter().map(|c| c.verdict).collect::>(), + vec![ + Verdict::Accepted, + Verdict::WrongAnswer, + Verdict::RuntimeError + ] + ); + assert_eq!(j.cases[0].cpu_time_us, Some(42)); + assert_eq!(j.cases[0].memory_peak_bytes, None); + } + #[test] + fn every_execution_failure_has_a_verdict() { + for (k, v) in [ + (ExecutionKind::TimeLimit, Verdict::TimeLimitExceeded), + (ExecutionKind::MemoryLimit, Verdict::MemoryLimitExceeded), + (ExecutionKind::OutputLimit, Verdict::OutputLimitExceeded), + (ExecutionKind::OtherLimit, Verdict::RuntimeError), + (ExecutionKind::Signaled, Verdict::RuntimeError), + (ExecutionKind::Cancelled, Verdict::Cancelled), + (ExecutionKind::InfrastructureFailure, Verdict::SystemError), + ] { + let mut p = Port { + kinds: [Ok(k)].into(), + calls: 0, + }; + assert_eq!(judge(&mut p, &[case("a", vec![])]).unwrap().verdict, v); + } + } + #[test] + fn infrastructure_failure_stops_and_overrides_prior_answer() { + let mut p = Port { + kinds: [ + Ok(ExecutionKind::Exited { code: 0 }), + Err(PortError), + Ok(ExecutionKind::Exited { code: 0 }), + ] + .into(), + calls: 0, + }; + let j = judge( + &mut p, + &[case("a", vec![]), case("b", vec![]), case("c", vec![])], + ) + .unwrap(); + assert_eq!(j.verdict, Verdict::SystemError); + assert_eq!(p.calls, 2); + assert_eq!(j.cases[1].cpu_time_us, None); + } + #[test] + fn empty_or_reused_run_ids_do_not_execute() { + let mut p = Port { + kinds: [].into(), + calls: 0, + }; + assert!(judge(&mut p, &[]).is_err()); + assert!(judge(&mut p, &[case("a", vec![]), case("a", vec![])]).is_err()); + assert_eq!(p.calls, 0); + } +} diff --git a/judge/crates/sandbox/src/artifacts.rs b/judge/crates/sandbox/src/artifacts.rs index 31d8cc4..b3bfad6 100644 --- a/judge/crates/sandbox/src/artifacts.rs +++ b/judge/crates/sandbox/src/artifacts.rs @@ -35,6 +35,23 @@ pub struct SnapshotFile { } impl SnapshotFile { + pub(crate) fn from_received(fd: OwnedFd, bytes: u64) -> io::Result { + let st = rustix::fs::fstat(&fd)?; + if FileType::from_raw_mode(st.st_mode) != FileType::RegularFile + || st.st_size < 0 + || st.st_size as u64 != bytes + || bytes > MAX_SNAPSHOT_BYTES + || rustix::fs::fcntl_getfl(&fd)?.intersection(OFlags::ACCMODE) != OFlags::RDONLY + || !rustix::fs::fcntl_get_seals(&fd)?.contains(SEALS) + || !rustix::io::fcntl_getfd(&fd)?.contains(rustix::io::FdFlags::CLOEXEC) + { + return Err(refused("invalid received snapshot")); + } + Ok(Self { + file: File::from(fd), + bytes, + }) + } pub fn bytes(&self) -> u64 { self.bytes } @@ -106,10 +123,40 @@ pub struct ArtifactRoot { owner: u32, } +impl AsFd for ArtifactRoot { + fn as_fd(&self) -> BorrowedFd<'_> { + self.fd.as_fd() + } +} impl ArtifactRoot { + pub fn open(path: &std::path::Path) -> io::Result { + Self::from_directory(rustix::fs::open( + path, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?) + } + pub fn snapshot_artifact(&self, id: &str, limit: u64) -> io::Result { + if limit == 0 || limit > MAX_SNAPSHOT_BYTES { + return Err(refused("snapshot ceiling")); + } + let mut budget = SnapshotBudget { + file_bytes: limit, + remaining_bytes: limit, + }; + self.snapshot_id(id, &mut budget, false) + } + pub(crate) fn from_worker_directory(fd: OwnedFd, owner: u32) -> io::Result { + if owner < 1000 { + return Err(refused("ordinary publication owner")); + } + Self::with_owner(fd, owner) + } pub fn from_directory(fd: OwnedFd) -> io::Result { + Self::with_owner(fd, rustix::process::geteuid().as_raw()) + } + fn with_owner(fd: OwnedFd, owner: u32) -> io::Result { let stat = rustix::fs::fstat(&fd)?; - let owner = rustix::process::geteuid().as_raw(); if FileType::from_raw_mode(stat.st_mode) != FileType::Directory || stat.st_uid != owner || stat.st_mode & 0o022 != 0 diff --git a/judge/crates/sandbox/src/execution.rs b/judge/crates/sandbox/src/execution.rs index 117f549..2c90ffe 100644 --- a/judge/crates/sandbox/src/execution.rs +++ b/judge/crates/sandbox/src/execution.rs @@ -76,13 +76,19 @@ fn read_pinned(fd: OwnedFd, limit: u64) -> io::Result> { } Ok(bytes) } -pub fn read_spec(path: &Path) -> io::Result { +pub fn read_bounded_file(path: &Path, limit: u64) -> io::Result> { + if limit == 0 || limit > crate::spec::MAX_SPEC_BYTES { + return Err(invalid("bounded JSON ceiling")); + } let fd = rustix::fs::open( path, OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty(), )?; - serde_json::from_slice(&read_pinned(fd, crate::spec::MAX_SPEC_BYTES)?) + read_pinned(fd, limit) +} +pub fn read_spec(path: &Path) -> io::Result { + serde_json::from_slice(&read_bounded_file(path, crate::spec::MAX_SPEC_BYTES)?) .map_err(|_| invalid("invalid execution JSON")) } pub fn load_config(path: &Path) -> io::Result { @@ -207,7 +213,7 @@ pub(crate) fn execute( config: &ValidatedRuntimeConfig, request: ExecutionSpec, artifacts: &ArtifactRoot, - mode: run::MonitorMode, + mode: run::MonitorMode<'_>, ) -> Result<(run::Completed, ValidatedSpec), RunFailure> { authorize(config).map_err(|e| refused("runtime_not_ready", &e))?; let spec = request diff --git a/judge/crates/sandbox/src/launcher/mod.rs b/judge/crates/sandbox/src/launcher/mod.rs index 52cb4c8..45aac03 100644 --- a/judge/crates/sandbox/src/launcher/mod.rs +++ b/judge/crates/sandbox/src/launcher/mod.rs @@ -1,9 +1,5 @@ -//! Trusted supervisor/launcher coordination, separate from user stdio. -//! -//! Lifecycle events describe completed operations, not commands received from -//! an untrusted request. The standalone entry authenticates its parent and uses -//! bounded IPC, and refuses user execution. The separate namespace bootstrap initializes -//! only fixed OS isolation operations and remains gated at the missing root stage. +//! Trusted launcher coordination with per-stage kernel acknowledgement, bounded +//! private control and one final exec gate; user stdio never carries commands. pub mod lifecycle; diff --git a/judge/crates/sandbox/src/launcher/protocol.rs b/judge/crates/sandbox/src/launcher/protocol.rs index 8749107..63b569d 100644 --- a/judge/crates/sandbox/src/launcher/protocol.rs +++ b/judge/crates/sandbox/src/launcher/protocol.rs @@ -43,7 +43,7 @@ pub enum ChildMessage { } /// A distinct fixed initialization protocol. There is no argv, host path, -/// readiness claim. VM acceptance builds add a separately verified final gate. +/// readiness claim. The final execution gate requires actual kernel acknowledgements. #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "command", rename_all = "snake_case", deny_unknown_fields)] pub enum NamespaceCommand { @@ -105,7 +105,7 @@ pub enum NamespaceEvent { Cancelled {}, } -mod sealed { +pub(crate) mod sealed { pub trait Sealed {} } diff --git a/judge/crates/sandbox/src/launcher/run.rs b/judge/crates/sandbox/src/launcher/run.rs index 75d9ec9..fafda42 100644 --- a/judge/crates/sandbox/src/launcher/run.rs +++ b/judge/crates/sandbox/src/launcher/run.rs @@ -39,7 +39,8 @@ pub(crate) struct RunProof { pub order: Vec<&'static str>, } #[derive(Clone, Copy)] -pub(crate) enum MonitorMode { +pub(crate) enum MonitorMode<'a> { + Peer(&'a rustix::fd::OwnedFd), Normal, #[cfg(feature = "vm-acceptance")] CancelAfter(u64), @@ -78,15 +79,28 @@ fn consume(event: BudgetEvent) -> (ResourceUsage, Vec) { } } +fn peer_closed(fd: &rustix::fd::OwnedFd) -> io::Result { + use rustix::event::{PollFd, PollFlags, poll}; + let mut f = [PollFd::new( + fd, + PollFlags::IN | PollFlags::HUP | PollFlags::ERR, + )]; + poll( + &mut f, + Some(&rustix::event::Timespec { + tv_sec: 0, + tv_nsec: 0, + }), + )?; + Ok(!f[0].revents().is_empty()) +} pub(crate) fn execute( config: &ValidatedRuntimeConfig, spec: &ValidatedSpec, binary: &Path, artifacts: &ArtifactRoot, - mode: MonitorMode, + mode: MonitorMode<'_>, ) -> Result { - #[cfg(not(feature = "vm-acceptance"))] - let _ = mode; let mut owned = OwnedChildren::begin().map_err(|e| InitializationFailure { failure: error(ExecutionStage::Preparation, ErrorCode::OperationFailed, &e), cleanup_failure: None, @@ -114,10 +128,7 @@ pub(crate) fn execute( let bootstrap = target.root().bootstrap_pid(); let mut errors = Vec::new(); let mut output_limits = Vec::new(); - #[cfg(feature = "vm-acceptance")] let mut cancelled = None; - #[cfg(not(feature = "vm-acceptance"))] - let cancelled: Option = None; let mut main = None; let mut reaped_descendants = 0; let mut order = vec!["verified_isolation", "exec_gate"]; @@ -146,6 +157,14 @@ pub(crate) fn execute( }); } let released = (|| { + if let MonitorMode::Peer(fd) = mode + && peer_closed(fd)? + { + return Err(io::Error::new( + io::ErrorKind::Interrupted, + "client disconnected before exec", + )); + } ns.control.send(&NamespaceCommand::ReleaseExecution {})?; budget.release_prepared_gate(&[bootstrap, init], &ns.init_pidfd)?; if ns.control.receive::()? != (NamespaceEvent::ExecutionSpawned {}) { @@ -256,8 +275,10 @@ pub(crate) fn execute( monitoring_failed = true; } } - #[cfg(feature = "vm-acceptance")] match mode { + MonitorMode::Peer(fd) if peer_closed(fd).unwrap_or(true) => { + cancelled = Some(wall(&usage)); + } #[cfg(feature = "vm-acceptance")] MonitorMode::CancelAfter(t) if wall(&usage) >= t => { cancelled = Some(wall(&usage)) diff --git a/judge/crates/sandbox/src/lib.rs b/judge/crates/sandbox/src/lib.rs index 10a0645..eb5eaa5 100644 --- a/judge/crates/sandbox/src/lib.rs +++ b/judge/crates/sandbox/src/lib.rs @@ -1,7 +1,5 @@ -//! Linux execution boundary: policy/contract checks and read-only observations. -//! -//! Execution contracts and lifecycle ordering precede OS launcher/isolation work. -//! Until that work is verified, the CLI cannot execute any user program. +//! Linux execution boundary: policy, kernel gates, bounded supervision and recovery. +//! Ordinary operator/Worker entries require trusted authority and active isolation. pub mod launcher; pub mod outcome; @@ -32,3 +30,6 @@ mod isolation; #[cfg(target_os = "linux")] pub mod execution; + +#[cfg(target_os = "linux")] +pub mod rpc; diff --git a/judge/crates/sandbox/src/main.rs b/judge/crates/sandbox/src/main.rs index 7982c8d..744e7b0 100644 --- a/judge/crates/sandbox/src/main.rs +++ b/judge/crates/sandbox/src/main.rs @@ -40,6 +40,13 @@ enum Command { #[arg(long)] output: Option, }, + /// One authenticated Worker request, in a dedicated finite root helper scope. + ServeOne { + #[arg(long)] + config: PathBuf, + #[arg(long)] + socket: PathBuf, + }, /// Internal trusted launcher control entry; never runs a user program. #[cfg(target_os = "linux")] #[command(hide = true)] @@ -120,6 +127,21 @@ fn main() -> ExitCode { spec_error("unsupported_platform", None) } } + Command::ServeOne { config, socket } => { + #[cfg(target_os = "linux")] + match kf_sandbox::rpc::serve_one(&config, &socket) { + Ok(result) => { + println!("{result}"); + ExitCode::SUCCESS + } + Err(_) => spec_error("worker_helper_refused", None), + } + #[cfg(not(target_os = "linux"))] + { + let _ = (config, socket); + spec_error("unsupported_platform", None) + } + } Command::Validate { spec } => validate_spec(spec), Command::CheckRuntime { config } => check_runtime(config), #[cfg(target_os = "linux")] diff --git a/judge/crates/sandbox/src/outcome.rs b/judge/crates/sandbox/src/outcome.rs index 4aeb8f1..c99842a 100644 --- a/judge/crates/sandbox/src/outcome.rs +++ b/judge/crates/sandbox/src/outcome.rs @@ -247,7 +247,7 @@ impl std::error::Error for ReportError {} impl ExecutionReport { /// Consistency checks on trusted observations. This does not verify OS - /// isolation and never changes the CLI's refusal to execute. + /// isolation or supply an execution permit. pub fn classify(&self, spec: &ValidatedSpec) -> Result { let limits = &spec.spec().limits; if self.run_id != spec.spec().run_id || self.errors.len() > 16 || self.limits.len() > 8 { diff --git a/judge/crates/sandbox/src/rpc.rs b/judge/crates/sandbox/src/rpc.rs new file mode 100644 index 0000000..81be740 --- /dev/null +++ b/judge/crates/sandbox/src/rpc.rs @@ -0,0 +1,245 @@ +//! One request per fresh privileged process. Kernel peer credentials authenticate +//! both endpoints; only the configured Worker may provide its own publication FD. +use crate::{ + artifacts::{ArtifactRoot, OutputSnapshot, SnapshotFile}, + execution, + launcher::{ + protocol::{ControlChannel, ExpectedPeer, WireMessage}, + run, + }, + outcome::{ExecutionError, ExecutionReport}, + spec::{ExecutionSpec, SpecPolicy}, +}; +use rustix::{ + fd::OwnedFd, + fs::{Mode, OFlags}, +}; +use serde::{Deserialize, Serialize}; +use std::{ + io, + os::unix::{ + fs::{MetadataExt, PermissionsExt}, + net::{UnixListener, UnixStream}, + }, + path::Path, + time::{Duration, Instant}, +}; +fn denied() -> io::Error { + io::Error::new(io::ErrorKind::PermissionDenied, "sandbox RPC refused") +} +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Request { + spec: ExecutionSpec, +} +#[derive(Serialize, Deserialize)] +#[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)] +pub(crate) enum Reply { + Completed { + report: Box, + }, + Rejected { + failure: ExecutionError, + cleanup_failure: Option, + }, +} +impl crate::launcher::protocol::sealed::Sealed for Request {} +impl crate::launcher::protocol::sealed::Sealed for Reply {} +impl WireMessage for Request { + const MAX_BYTES: usize = crate::spec::MAX_SPEC_BYTES as usize + 128; + fn valid(&self) -> bool { + self.spec.run_id.len() <= 64 + } +} +impl WireMessage for Reply { + const MAX_BYTES: usize = crate::spec::MAX_SPEC_BYTES as usize + 128; + fn valid(&self) -> bool { + match self { + Self::Completed { report } => { + report.artifacts.len() <= 16 + && report.errors.len() <= 16 + && report.limits.len() <= 8 + } + Self::Rejected { .. } => true, + } + } +} +fn channel(stream: UnixStream, uid: u32, gid: u32) -> io::Result { + let c = rustix::net::sockopt::socket_peercred(&stream)?; + ControlChannel::new( + stream, + ExpectedPeer { + pid: c.pid.as_raw_pid() as u32, + uid, + gid, + }, + Instant::now() + Duration::from_secs(5), + ) +} +#[derive(Debug)] +pub struct Rejected { + pub failure: ExecutionError, + pub cleanup_failure: Option, +} +impl std::fmt::Display for Rejected { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("sandbox request rejected") + } +} +impl std::error::Error for Rejected {} +pub struct Executed { + pub report: ExecutionReport, + pub stdout: SnapshotFile, + pub stderr: SnapshotFile, + pub outputs: Vec, +} +/// No privileges are acquired: this endpoint requires an ordinary uid/gid. +pub fn execute( + socket: &Path, + spec: ExecutionSpec, + publication: &ArtifactRoot, +) -> io::Result { + if rustix::process::geteuid().as_raw() < 1000 || rustix::process::getegid().as_raw() < 1000 { + return Err(denied()); + } + let validated = spec + .clone() + .validate(&SpecPolicy::default()) + .map_err(|_| denied())?; + let mut c = channel(UnixStream::connect(socket)?, 0, 0)?; + c.send(&Request { spec })?; + c.send_fd(publication)?; + c.execution_deadline(validated.spec().limits.wall_time_us)?; + let report = match c.receive::()? { + Reply::Completed { report } => *report, + Reply::Rejected { + failure, + cleanup_failure, + } => { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + Rejected { + failure, + cleanup_failure, + }, + )); + } + }; + report.classify(&validated).map_err(|_| denied())?; + let stdout = SnapshotFile::from_received(c.receive_fd()?, report.stdout.retained_bytes)?; + let stderr = SnapshotFile::from_received(c.receive_fd()?, report.stderr.retained_bytes)?; + let mut outputs = Vec::new(); + for a in &report.artifacts { + outputs.push(OutputSnapshot { + path: a.path.clone(), + file: SnapshotFile::from_received(c.receive_fd()?, a.bytes)?, + }); + } + Ok(Executed { + report, + stdout, + stderr, + outputs, + }) +} +struct SocketLease<'a> { + parent: OwnedFd, + name: &'a std::ffi::OsStr, + dev: u64, + ino: u64, +} +impl Drop for SocketLease<'_> { + fn drop(&mut self) { + if let Ok(s) = rustix::fs::statat( + &self.parent, + self.name, + rustix::fs::AtFlags::SYMLINK_NOFOLLOW, + ) && s.st_dev == self.dev + && s.st_ino == self.ino + { + let _ = rustix::fs::unlinkat(&self.parent, self.name, rustix::fs::AtFlags::empty()); + } + } +} +/// Operator supplies protected config/socket paths; a service manager provisions +/// the finite helper scope. No internal sudo, account changes or deployment. +pub fn serve_one(config_path: &Path, socket: &Path) -> io::Result { + let config = execution::load_config(config_path)?; + execution::authorize(&config)?; + let parent = crate::isolation::root::open_directory(socket.parent().ok_or_else(denied)?)?; + let name = socket.file_name().ok_or_else(denied)?; + // Refuse replacement, including an existing stale/foreign socket. + if rustix::fs::statat(&parent, name, rustix::fs::AtFlags::SYMLINK_NOFOLLOW).is_ok() { + return Err(denied()); + } + let listener = UnixListener::bind(socket)?; + let m = std::fs::symlink_metadata(socket)?; + let lease = SocketLease { + parent, + name, + dev: m.dev(), + ino: m.ino(), + }; + let socket_fd = rustix::fs::openat( + &lease.parent, + name, + OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + )?; + let stat = rustix::fs::fstat(&socket_fd)?; + if stat.st_dev != lease.dev || stat.st_ino != lease.ino || stat.st_uid != 0 { + return Err(denied()); + } + // Ancestors are non-writable to Worker. Operations cannot follow a substituted path. + std::os::unix::fs::chown(socket, Some(0), Some(config.config().identities.worker_gid))?; + std::fs::set_permissions(socket, std::fs::Permissions::from_mode(0o660))?; + listener.set_nonblocking(true)?; + let deadline = Instant::now() + Duration::from_secs(5); + let stream = loop { + match listener.accept() { + Ok((s, _)) => break s, + Err(e) if e.kind() == io::ErrorKind::WouldBlock && Instant::now() < deadline => { + std::thread::sleep(Duration::from_millis(5)) + } + Err(e) => return Err(e), + } + }; + drop(listener); + let ids = &config.config().identities; + let mut c = channel(stream, ids.worker_uid, ids.worker_gid)?; + let request = c.receive::()?; + 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, + request.spec, + &artifacts, + run::MonitorMode::Peer(&peer), + ) { + 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() { + return Ok(serde_json::json!({"report":report,"proof":result.proof})); + } + c.send(&Reply::Completed { + report: Box::new(report.clone()), + })?; + c.send_fd(&result.stdout)?; + c.send_fd(&result.stderr)?; + for o in result.outputs { + c.send_fd(&o.file)?; + } + Ok(serde_json::json!({"report":report,"proof":result.proof})) + } + Err(e) => { + let failure = e.failure.ok_or_else(denied)?; + c.send(&Reply::Rejected { + failure, + cleanup_failure: e.cleanup_failure, + })?; + Err(denied()) + } + } +} diff --git a/judge/crates/sandbox/tests/fixtures/worker_disconnect.c b/judge/crates/sandbox/tests/fixtures/worker_disconnect.c new file mode 100644 index 0000000..67cc1eb --- /dev/null +++ b/judge/crates/sandbox/tests/fixtures/worker_disconnect.c @@ -0,0 +1,25 @@ +/* Fixed independent-VM controller only. Not installed in the ordinary CLI. */ +#include +#include +#include +#include +#include +#include +#include +int main(int argc,char **argv) { + if(argc!=4 || getuid()!=0 || setgid(1000) || setuid(1000))return 100; + int d=open(argv[2],O_RDONLY|O_DIRECTORY|O_CLOEXEC); + int f=open(argv[3],O_RDONLY|O_CLOEXEC); + char body[66000];int n=(int)read(f,body,sizeof body);if(d<0||f<0||n<=0||n>=66000)return 101;close(f); + int s=socket(AF_UNIX,SOCK_STREAM|SOCK_CLOEXEC,0);struct sockaddr_un a={0};a.sun_family=AF_UNIX; + if(strlen(argv[1])>=sizeof a.sun_path){return 102;} + strcpy(a.sun_path,argv[1]); + if(connect(s,(struct sockaddr*)&a,sizeof a))return 103; + unsigned char header[8]={'K','F',1,0,(unsigned char)((unsigned)n>>24),(unsigned char)((unsigned)n>>16),(unsigned char)((unsigned)n>>8),(unsigned char)n}; + if(write(s,header,8)!=8||write(s,body,(size_t)n)!=n)return 104; + char marker=(char)0xa5;struct iovec i={&marker,1};union {struct cmsghdr align;char bytes[CMSG_SPACE(sizeof(int))];}control={0}; + struct msghdr m={0};m.msg_iov=&i;m.msg_iovlen=1;m.msg_control=control.bytes;m.msg_controllen=sizeof control.bytes; + struct cmsghdr *c=CMSG_FIRSTHDR(&m);c->cmsg_level=SOL_SOCKET;c->cmsg_type=SCM_RIGHTS;c->cmsg_len=CMSG_LEN(sizeof(int));memcpy(CMSG_DATA(c),&d,sizeof d); + if(sendmsg(s,&m,MSG_NOSIGNAL)!=1)return 105; + close(d);sleep(2);close(s);return 0; +} diff --git a/judge/crates/sandbox/tests/fixtures/worker_native.c b/judge/crates/sandbox/tests/fixtures/worker_native.c new file mode 100644 index 0000000..30a297e --- /dev/null +++ b/judge/crates/sandbox/tests/fixtures/worker_native.c @@ -0,0 +1,10 @@ +#include +int main(void) { + unsigned char b[4096]; + ssize_t n=read(0,b,sizeof b); + if(n<0)return 8; + if(n>0 && b[0]=='#')return 7; + if(n>0 && b[0]=='!'){if(write(1,"live",4)!=4)return 9;sleep(10);return 0;} + do {if(n>0 && write(1,b,(size_t)n)!=n)return 10;}while((n=read(0,b,sizeof b))>0); + return n<0 ? 11 : 0; +} diff --git a/judge/crates/sandbox/tests/worker_vm.rs b/judge/crates/sandbox/tests/worker_vm.rs new file mode 100644 index 0000000..c20c9b2 --- /dev/null +++ b/judge/crates/sandbox/tests/worker_vm.rs @@ -0,0 +1,240 @@ +#![cfg(all(target_os = "linux", feature = "vm-acceptance"))] +#[allow(dead_code)] +mod support; +use std::{ + fs, + os::unix::{ + fs::{PermissionsExt, chown}, + net::UnixStream, + process::CommandExt, + }, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + time::{Duration, Instant}, +}; +struct Fixture { + root: support::Root, + dir: PathBuf, + config: PathBuf, + socket: PathBuf, +} +impl Fixture { + fn new() -> Self { + support::require_vm(); + let mut root = support::Root::new(); + root.config.aggregate.task_cpu_quota_us = 100_000; + let dir = PathBuf::from(format!("/kf/worker-{}", std::process::id())); + 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(); + let p = dir.join("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(); + Self::worker_file(&p.join("program")); + let socket = dir.join("socket"); + Self { + root, + dir, + config, + socket, + } + } + fn worker_file(p: &Path) { + chown(p, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(p, fs::Permissions::from_mode(0o600)).unwrap(); + } + fn job(&self, id: &str, input: &[u8], expected: &[u8]) { + for (name, data) in [("stdin", input), ("expected", expected)] { + let p = self.dir.join(format!("inputs/{name}")); + fs::write(&p, data).unwrap(); + Self::worker_file(&p); + } + let j = serde_json::json!({"request":{"run_id":id,"program_artifact":"program","stdin_artifact":"stdin"},"expected_artifact":"expected"}); + fs::write(self.dir.join("job.json"), serde_json::to_vec(&j).unwrap()).unwrap(); + Self::worker_file(&self.dir.join("job.json")); + } + fn helper(&self) -> Child { + let mut c = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap()) + .env_clear() + .args(["serve-one", "--config"]) + .arg(&self.config) + .arg("--socket") + .arg(&self.socket) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + while !self.socket.exists() { + assert!( + c.try_wait().unwrap().is_none(), + "helper exited before socket" + ); + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(5)); + } + // Socket creation precedes mode installation; wait for actual published permissions. + while fs::metadata(&self.socket).unwrap().permissions().mode() & 0o777 != 0o660 { + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(2)); + } + c + } + fn empty(&self) { + for p in ["tasks", "identities", "captures"] { + let p = PathBuf::from(format!("/var/lib/kf-judge/{p}")); + if p.exists() { + assert_eq!(fs::read_dir(p).unwrap().count(), 0); + } + } + assert!( + !fs::read_dir(&self.root.path).unwrap().any(|e| e + .unwrap() + .file_type() + .unwrap() + .is_dir()) + ); + assert!(!self.socket.exists()); + } + fn execute(&self, id: &str, input: &[u8], expected: &[u8], verdict: &str) { + self.job(id, input, expected); + let h = self.helper(); + let w = Command::new(std::env::var_os("KF_WORKER_TEST_BIN").unwrap()) + .env_clear() + .args(["judge-once", "--socket"]) + .arg(&self.socket) + .arg("--job") + .arg(self.dir.join("job.json")) + .arg("--artifacts") + .arg(self.dir.join("inputs")) + .uid(1000) + .gid(1000) + .output() + .unwrap(); + assert!(w.status.success(), "{}", String::from_utf8_lossy(&w.stderr)); + let j: serde_json::Value = serde_json::from_slice(&w.stdout).unwrap(); + assert_eq!(j["verdict"], verdict, "{j}"); + assert_eq!(j["cases"][0]["run_id"], id); + assert!(j["cases"][0]["cpu_time_us"].as_u64().unwrap() > 0); + let o = h.wait_with_output().unwrap(); + assert!(o.status.success(), "{}", String::from_utf8_lossy(&o.stderr)); + let v: serde_json::Value = serde_json::from_slice(&o.stdout).unwrap(); + assert_eq!(v["report"]["cleanup"], "complete"); + assert_eq!(v["report"]["errors"], serde_json::json!([])); + self.empty(); + println!( + "KF_WORKER_CHAIN_PROOF {}", + serde_json::json!({"worker_uid":1000,"worker_gid":1000,"kernel_peer_authentication":true,"judgement":j,"sandbox":v,"expected_artifact_not_mounted":true}) + ); + } +} +impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.dir); + } +} +#[test] +#[ignore = "independent capped VM: actual ordinary Worker -> Engine -> default-build root helper"] +fn native_binary_comparison_and_execution_failures() { + let f = Fixture::new(); + f.execute("worker-ac", &[0, 255, 10], &[0, 255, 10], "accepted"); + f.execute("worker-wa", &[0, 255, 10], &[0, 255], "wrong_answer"); + f.execute("worker-re", b"#", b"", "runtime_error"); + f.execute("worker-tle", b"!", b"live", "time_limit_exceeded"); +} +#[test] +#[ignore = "independent capped VM: real kernel peer credential denial and foreign socket preservation"] +fn mismatched_peer_and_existing_socket_are_refused() { + let f = Fixture::new(); + let h = f.helper(); + let c = UnixStream::connect(&f.socket).unwrap(); + drop(c); + let o = h.wait_with_output().unwrap(); + assert_eq!(o.status.code(), Some(2)); + assert!(o.stdout.is_empty()); + f.empty(); + f.job("worker-owner", b"!", b""); + let publication = f.dir.join("inputs"); + chown(&publication, Some(0), Some(0)).unwrap(); + fs::set_permissions(&publication, fs::Permissions::from_mode(0o555)).unwrap(); + let h = f.helper(); + let mut spec: kf_sandbox::spec::ExecutionSpec = + serde_json::from_str(include_str!("fixtures/spec.valid.json")).unwrap(); + spec.run_id = "worker-owner".into(); + spec.inputs[0].artifact_id = "program".into(); + spec.stdin_artifact = Some("stdin".into()); + let path = f.dir.join("wire.json"); + fs::write( + &path, + serde_json::to_vec(&serde_json::json!({"spec":spec})).unwrap(), + ) + .unwrap(); + Fixture::worker_file(&path); + let c = Command::new("/kf/bin/worker-disconnect") + .arg(&f.socket) + .arg(&publication) + .arg(&path) + .output() + .unwrap(); + assert!(c.status.success()); + let o = h.wait_with_output().unwrap(); + assert_eq!(o.status.code(), Some(2)); + assert!(o.stdout.is_empty()); + f.empty(); + chown(&publication, Some(1000), Some(1000)).unwrap(); + fs::set_permissions(&publication, fs::Permissions::from_mode(0o700)).unwrap(); + fs::write(&f.socket, b"foreign socket name").unwrap(); + let o = Command::new(std::env::var_os("KF_SANDBOX_PRODUCTION_BIN").unwrap()) + .args(["serve-one", "--config"]) + .arg(&f.config) + .arg("--socket") + .arg(&f.socket) + .output() + .unwrap(); + assert_eq!(o.status.code(), Some(2)); + assert_eq!(fs::read(&f.socket).unwrap(), b"foreign socket name"); + fs::remove_file(&f.socket).unwrap(); + f.empty(); + println!( + "KF_WORKER_AUTH_PROOF wrong_root_peer_refused=1 root_owned_publication_refused=1 foreign_socket_preserved=1 zero_resources=1" + ); +} +#[test] +#[ignore = "independent capped VM: authenticated ordinary client disconnect while actual task is live"] +fn client_disconnect_cancels_entire_task_and_cleans() { + let f = Fixture::new(); + f.job("worker-cancel", b"!", b""); + let h = f.helper(); + let mut spec: kf_sandbox::spec::ExecutionSpec = + serde_json::from_str(include_str!("fixtures/spec.valid.json")).unwrap(); + spec.run_id = "worker-cancel".into(); + spec.inputs[0].artifact_id = "program".into(); + spec.stdin_artifact = Some("stdin".into()); + let path = f.dir.join("wire.json"); + fs::write( + &path, + serde_json::to_vec(&serde_json::json!({"spec":spec})).unwrap(), + ) + .unwrap(); + Fixture::worker_file(&path); + let c = Command::new("/kf/bin/worker-disconnect") + .arg(&f.socket) + .arg(f.dir.join("inputs")) + .arg(&path) + .output() + .unwrap(); + assert!(c.status.success(), "{}", String::from_utf8_lossy(&c.stderr)); + let o = h.wait_with_output().unwrap(); + assert!(o.status.success(), "{}", String::from_utf8_lossy(&o.stderr)); + let v: serde_json::Value = serde_json::from_slice(&o.stdout).unwrap(); + assert!(v["report"]["cancelled_at_wall_time_us"].as_u64().unwrap() > 0); + assert_eq!(v["report"]["stdout"]["retained_bytes"], 4); + assert_eq!(v["report"]["cleanup"], "complete"); + assert_eq!(v["report"]["errors"], serde_json::json!([])); + f.empty(); + println!("KF_WORKER_DISCONNECT_PROOF {v}"); +} diff --git a/judge/crates/worker/Cargo.toml b/judge/crates/worker/Cargo.toml index 71152e8..daed9a8 100644 --- a/judge/crates/worker/Cargo.toml +++ b/judge/crates/worker/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true [dependencies] kf-contracts.workspace = true +serde.workspace = true kf-engine.workspace = true kf-sandbox.workspace = true clap.workspace = true diff --git a/judge/crates/worker/src/lib.rs b/judge/crates/worker/src/lib.rs new file mode 100644 index 0000000..ec734d1 --- /dev/null +++ b/judge/crates/worker/src/lib.rs @@ -0,0 +1,113 @@ +//! Ordinary Worker adapter. Privileged execution and artifact capabilities stay +//! in sandbox; comparison and the final summary stay in Engine. +#[cfg(target_os = "linux")] +pub mod native { + use kf_engine::{Case, CaseRequest, ExecutionKind, ExecutionPort, Observation, PortError}; + use kf_sandbox::{ + artifacts::ArtifactRoot, + outcome::{ExecutionOutcome, LimitKind, Measurement}, + spec::{ExecutionProfile, ExecutionSpec, InputArtifact, ResourceLimits}, + }; + use serde::Deserialize; + use std::{ + io::{self, Read}, + path::{Path, PathBuf}, + }; + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + pub struct Job { + pub request: CaseRequest, + pub expected_artifact: String, + } + pub struct SandboxPort { + socket: PathBuf, + artifacts: ArtifactRoot, + } + impl ExecutionPort for SandboxPort { + fn execute(&mut self, request: &CaseRequest) -> Result { + let spec = ExecutionSpec { + schema_version: kf_contracts::SchemaVersion, + run_id: request.run_id.clone(), + profile: ExecutionProfile::Native, + argv: vec!["/work/main".into()], + env: std::collections::BTreeMap::new(), + cwd: ".".into(), + inputs: vec![InputArtifact { + artifact_id: request.program_artifact.clone(), + path: "main".into(), + }], + stdin_artifact: request.stdin_artifact.clone(), + outputs: vec![], + limits: ResourceLimits::for_profile(ExecutionProfile::Native), + }; + let validated = spec + .clone() + .validate(&kf_sandbox::spec::SpecPolicy::default()) + .map_err(|_| PortError)?; + let result = kf_sandbox::rpc::execute(&self.socket, spec, &self.artifacts) + .map_err(|_| PortError)?; + let outcome = result.report.classify(&validated).map_err(|_| PortError)?; + let kind = match outcome { + 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 + } + }; + let mut stdout = Vec::new(); + result + .stdout + .reader() + .map_err(|_| PortError)? + .read_to_end(&mut stdout) + .map_err(|_| PortError)?; + let value = |m: Measurement| match m { + Measurement::Measured { value } => Some(value), + Measurement::Unavailable { .. } => None, + }; + Ok(Observation { + kind, + stdout, + cpu_time_us: value(result.report.usage.cpu_time_us), + memory_peak_bytes: value(result.report.usage.memory_peak_bytes), + }) + } + } + pub fn judge_once( + socket: &Path, + job_path: &Path, + artifact_path: &Path, + ) -> io::Result { + let data = + kf_sandbox::execution::read_bounded_file(job_path, kf_sandbox::spec::MAX_SPEC_BYTES)?; + let job: Job = + serde_json::from_slice(&data).map_err(|_| io::Error::other("invalid native job"))?; + let artifacts = ArtifactRoot::open(artifact_path)?; + let expected = artifacts.snapshot_artifact( + &job.expected_artifact, + ResourceLimits::for_profile(ExecutionProfile::Native).stdout_bytes, + )?; + let mut bytes = Vec::new(); + expected.reader()?.read_to_end(&mut bytes)?; + let case = Case { + request: job.request, + expected: bytes, + }; + let mut port = SandboxPort { + socket: socket.to_owned(), + artifacts, + }; + kf_engine::judge(&mut port, &[case]).map_err(|_| io::Error::other("invalid judge request")) + } +} diff --git a/judge/crates/worker/src/main.rs b/judge/crates/worker/src/main.rs index 893ae94..98881ca 100644 --- a/judge/crates/worker/src/main.rs +++ b/judge/crates/worker/src/main.rs @@ -1,28 +1,64 @@ -use clap::Parser; -use std::process::ExitCode; - +use clap::{Parser, Subcommand}; +use std::{path::PathBuf, process::ExitCode}; #[derive(Parser)] -#[command(version, about = "KF judge worker entrypoint")] +#[command( + version, + about = "KF ordinary judge worker", + args_conflicts_with_subcommands = true +)] struct Cli { - /// Check the process entrypoint without consuming or executing tasks. #[arg(long)] check: bool, + #[command(subcommand)] + command: Option, +} +#[derive(Subcommand)] +enum Command { + JudgeOnce { + #[arg(long)] + socket: PathBuf, + #[arg(long)] + job: PathBuf, + #[arg(long)] + artifacts: PathBuf, + }, } - fn main() -> ExitCode { - if !Cli::parse().check { - eprintln!("judge worker task loop is not implemented; use --check"); - return ExitCode::from(2); + let cli = Cli::parse(); + if cli.check { + println!( + "{}", + serde_json::json!({"service":"judge-worker","version":env!("CARGO_PKG_VERSION"),"status":"entry_ready","contract_version":kf_contracts::SCHEMA_VERSION,"consuming_tasks":false}) + ); + return ExitCode::SUCCESS; + } + match cli.command { + Some(Command::JudgeOnce { + socket, + job, + artifacts, + }) => { + #[cfg(target_os = "linux")] + match kf_worker::native::judge_once(&socket, &job, &artifacts) { + Ok(result) => { + println!("{}", serde_json::json!(result)); + ExitCode::SUCCESS + } + Err(_) => { + eprintln!("{}", serde_json::json!({"error":"judge_request_failed"})); + ExitCode::from(2) + } + } + #[cfg(not(target_os = "linux"))] + { + let _ = (socket, job, artifacts); + eprintln!("unsupported worker platform"); + ExitCode::from(2) + } + } + None => { + eprintln!("worker consumer loop unavailable; use --check or judge-once"); + ExitCode::from(2) + } } - println!( - "{}", - serde_json::json!({ - "service": "judge-worker", - "version": env!("CARGO_PKG_VERSION"), - "status": "scaffold", - "contract_version": kf_contracts::SCHEMA_VERSION, - "consuming_tasks": false - }) - ); - ExitCode::SUCCESS }