From 08db00b56ea7a0e7691a80a88941ceb358d60c57 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Tue, 6 Oct 2026 22:04:29 +0800 Subject: [PATCH] feat(sandbox): add bounded standalone launcher control --- judge/Cargo.lock | 26 ++ judge/README.md | 32 +- judge/crates/sandbox/Cargo.toml | 1 + judge/crates/sandbox/src/launcher/entry.rs | 79 +++++ judge/crates/sandbox/src/launcher/mod.rs | 9 +- judge/crates/sandbox/src/launcher/protocol.rs | 314 ++++++++++++++++++ judge/crates/sandbox/src/main.rs | 19 ++ .../crates/sandbox/tests/launcher_control.rs | 303 +++++++++++++++++ 8 files changed, 779 insertions(+), 4 deletions(-) create mode 100644 judge/crates/sandbox/src/launcher/entry.rs create mode 100644 judge/crates/sandbox/src/launcher/protocol.rs create mode 100644 judge/crates/sandbox/tests/launcher_control.rs diff --git a/judge/Cargo.lock b/judge/Cargo.lock index 1699e97..96cb418 100644 --- a/judge/Cargo.lock +++ b/judge/Cargo.lock @@ -119,6 +119,12 @@ dependencies = [ "tracing", ] +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + [[package]] name = "bumpalo" version = "3.20.3" @@ -379,6 +385,7 @@ version = "0.0.1" dependencies = [ "clap", "kf-contracts", + "rustix", "serde", "serde_json", ] @@ -406,6 +413,12 @@ version = "0.2.190" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + [[package]] name = "log" version = "0.4.34" @@ -518,6 +531,19 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "rustix" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys", +] + [[package]] name = "rustversion" version = "1.0.23" diff --git a/judge/README.md b/judge/README.md index 5171b60..6bf8f9b 100644 --- a/judge/README.md +++ b/judge/README.md @@ -91,8 +91,36 @@ FD、栈和其他信号仍按原退出状态记录，缺少专门证据时不猜 最先停止的原因只用于调度诊断，最终执行分类仍由结果模型依据全部观察确定。 事件仅由可信监督器在核验同一任务的实际操作后产生，不可从执行请求直接反序列化。 -该模块不执行系统调用、不证明隔离、不产生执行许可；有界 IPC、单线程启动器、真实终止与 -清理仍待实现和独立 VM 验证。当前 CLI `run` 保持拒绝执行。 +该状态模块不执行系统调用、不证明隔离、不产生执行许可；真实任务树终止与清理仍待实现和 +独立 VM 验证。当前 CLI `run` 保持拒绝执行。 + +## 独立启动入口与控制通信 + +`launcher/entry.rs` 在可信二进制的独立进程中运行，不启动 Tokio、不安装自定义 `pre_exec` +闭包，也不在 Worker 的 fork 子进程里运行 Rust 初始化逻辑。入口实际检查 `/proc/self/task` +只有一个线程；未来每次权限操作或 fork 前须再次检查。当前入口以普通身份运行，没有实现 +特权 helper、namespace、降权、seccomp 或用户程序执行，初始化返回准备阶段的能力不可用。 + +控制端必须先在可信 state 根下创建独占、权限 0700 的每次启动目录和 Unix listener，再通过 +清空环境的 `Command` 启动内部入口；socket 不放在用户工作区，控制通道与用户 stdio 分开。 +入口通过非阻塞、CLOEXEC socket 连接，队列满时直接失败，不在连接阶段无限等待。 +双方用内核 `SO_PEERCRED` 核对预期 PID/UID/GID；父端 PID 来源于仍持有的 `Child`， +子端父 PID 来源于内核。内部入口不提供用户命令参数，握手消息不能转换成生命周期确认事件。 + +`launcher/protocol.rs` 使用带版本的 8 字节帧头和最多 1 KiB JSON。先拒绝未知版本、标志及 +非法长度，再读取固定大小缓冲；发送序列化也使用固定缓冲。父子消息类型分开，拒绝多余字段、 +非法 run_id 和未知命令。所有 I/O 共用一个绝对截止时间（入口 5 秒，通道最大 10 秒）， +部分帧不能续期；EOF、解析、超时或写入错误关闭并锁定通道，不能重新同步后继续初始化。 + +新增 `rustix` 1.1.5 仅用于沙箱的安全 syscall/FD 封装，启用 `net`、`process`，由锁文件固定。 +选择它的 OwnedFd/AsFd 接口完成连接与凭据核对；目前无需 `nix` 或直接 `libc` 调用， +也未放宽 `unsafe` 限制。参见 [rustix](https://github.com/bytecodealliance/rustix)、 +[socket_peercred](https://docs.rs/rustix/1.1.5/rustix/net/sockopt/fn.socket_peercred.html) 和 +[fork 后的安全约束](https://doc.rust-lang.org/std/os/unix/process/trait.CommandExt.html#tymethod.pre_exec)。 + +普通 Linux 回归实际启动可信入口，验证单线程、认证握手、准备拒绝、取消和断开后的退出回收； +协议回归验证边界与失败锁定。这些测试没有运行用户代码或安全攻击，不是 Linux 隔离验收。 +目标进程的阻塞/放行、完整 FD 白名单、进程树监督、阶段回滚及独立 VM 安全验收仍待完成。 ## 可信运行配置与启动前检查 diff --git a/judge/crates/sandbox/Cargo.toml b/judge/crates/sandbox/Cargo.toml index c36bdf2..423b393 100644 --- a/judge/crates/sandbox/Cargo.toml +++ b/judge/crates/sandbox/Cargo.toml @@ -10,6 +10,7 @@ kf-contracts.workspace = true clap.workspace = true serde.workspace = true serde_json.workspace = true +rustix = { version = "1.1.5", features = ["net", "process"] } [lints] workspace = true diff --git a/judge/crates/sandbox/src/launcher/entry.rs b/judge/crates/sandbox/src/launcher/entry.rs new file mode 100644 index 0000000..254074a --- /dev/null +++ b/judge/crates/sandbox/src/launcher/entry.rs @@ -0,0 +1,79 @@ +//! Runs after a fresh exec of the trusted binary, without a Tokio runtime. +//! No custom pre_exec closure, fork, user program, or permission change occurs. + +use super::protocol::{ChildMessage, ControlChannel, ExpectedPeer, ParentMessage}; +use crate::outcome::{ErrorCode, ExecutionStage}; +use std::io; +use std::os::unix::net::UnixStream; +use std::path::Path; +use std::time::{Duration, Instant}; + +pub const STARTUP_TIMEOUT: Duration = Duration::from_secs(5); + +/// Parent must create a listener in a fresh private directory under its trusted +/// state root before spawning this entry. Socket path is trusted deployment data. +/// The kernel peer credentials, rather than that path, authenticate the parent. +pub fn serve(socket_path: &Path) -> io::Result<()> { + let deadline = Instant::now() + STARTUP_TIMEOUT; + ensure_single_threaded()?; + let parent = rustix::process::getppid() + .filter(|pid| *pid != rustix::process::Pid::INIT) + .ok_or_else(|| { + io::Error::new(io::ErrorKind::PermissionDenied, "missing launcher parent") + })?; + // A full/stalled listener queue must fail, rather than block before the + // channel deadline is installed. OwnedFd closes on every failure path. + let socket = rustix::net::socket_with( + rustix::net::AddressFamily::UNIX, + rustix::net::SocketType::STREAM, + rustix::net::SocketFlags::CLOEXEC | rustix::net::SocketFlags::NONBLOCK, + None, + )?; + let address = rustix::net::SocketAddrUnix::new(socket_path)?; + rustix::net::connect(&socket, &address)?; + let stream = UnixStream::from(socket); + let mut channel = ControlChannel::new( + stream, + ExpectedPeer { + pid: parent.as_raw_pid() as u32, + uid: rustix::process::geteuid().as_raw(), + gid: rustix::process::getegid().as_raw(), + }, + deadline, + )?; + channel.send(&ChildMessage::Hello { + pid: std::process::id(), + threads: 1, + })?; + match channel.receive::()? { + ParentMessage::Initialize { run_id } => { + // A transport handshake is never an isolation or exec permit. + channel.send(&ChildMessage::InitializationRefused { + run_id, + stage: ExecutionStage::Preparation, + code: ErrorCode::CapabilityUnavailable, + })?; + } + ParentMessage::Cancel {} => channel.send(&ChildMessage::Cancelled {})?, + } + Ok(()) +} + +/// Must be repeated immediately before future privileged/fork operations. +pub fn ensure_single_threaded() -> io::Result<()> { + let mut count = 0; + for entry in std::fs::read_dir("/proc/self/task")? { + entry?; + count += 1; + if count > 1 { + break; + } + } + if count != 1 { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "launcher requires one thread", + )); + } + Ok(()) +} diff --git a/judge/crates/sandbox/src/launcher/mod.rs b/judge/crates/sandbox/src/launcher/mod.rs index ed7e251..9e5b18a 100644 --- a/judge/crates/sandbox/src/launcher/mod.rs +++ b/judge/crates/sandbox/src/launcher/mod.rs @@ -1,7 +1,12 @@ //! Trusted supervisor/launcher coordination, separate from user stdio. //! //! Lifecycle events describe completed operations, not commands received from -//! an untrusted request. OS operations and authenticated bounded IPC are pending; -//! this module grants no execution authority and does not change CLI refusal. +//! an untrusted request. The standalone entry authenticates its parent and uses +//! bounded IPC, but refuses initialization until OS isolation is implemented. pub mod lifecycle; + +#[cfg(target_os = "linux")] +pub mod entry; +#[cfg(target_os = "linux")] +pub mod protocol; diff --git a/judge/crates/sandbox/src/launcher/protocol.rs b/judge/crates/sandbox/src/launcher/protocol.rs new file mode 100644 index 0000000..f0bc675 --- /dev/null +++ b/judge/crates/sandbox/src/launcher/protocol.rs @@ -0,0 +1,314 @@ +//! Private per-attempt Unix control transport, never user stdin/stdout. +//! Messages cannot produce Lifecycle events or prove completed OS operations. + +use crate::outcome::{ErrorCode, ExecutionStage}; +use serde::{Deserialize, Serialize, de::DeserializeOwned}; +use std::io::{self, Read, Write}; +use std::net::Shutdown; +use std::os::unix::net::UnixStream; +use std::time::{Duration, Instant}; + +pub const MAX_MESSAGE_BYTES: usize = 1024; +pub const MAX_CONTROL_TIMEOUT: Duration = Duration::from_secs(10); +const PREFIX: [u8; 4] = [b'K', b'F', 1, 0]; + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "command", rename_all = "snake_case", deny_unknown_fields)] +pub enum ParentMessage { + Initialize { run_id: String }, + Cancel {}, +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)] +pub enum ChildMessage { + Hello { + pid: u32, + threads: u8, + }, + InitializationRefused { + run_id: String, + stage: ExecutionStage, + code: ErrorCode, + }, + Cancelled {}, +} + +mod sealed { + pub trait Sealed {} +} + +/// Sealed to keep the control protocol separate from arbitrary execution data. +pub trait WireMessage: sealed::Sealed + Serialize + DeserializeOwned { + fn valid(&self) -> bool; +} + +impl sealed::Sealed for ParentMessage {} +impl WireMessage for ParentMessage { + fn valid(&self) -> bool { + match self { + Self::Initialize { run_id } => valid_run_id(run_id), + Self::Cancel {} => true, + } + } +} + +impl sealed::Sealed for ChildMessage {} +impl WireMessage for ChildMessage { + fn valid(&self) -> bool { + match self { + Self::Hello { pid, threads } => *pid > 0 && *threads == 1, + Self::InitializationRefused { run_id, .. } => valid_run_id(run_id), + Self::Cancelled {} => true, + } + } +} + +fn valid_run_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= 64 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"_-".contains(&byte)) +} + +/// Expected credentials come from the spawned Child handle / actual parent, +/// never the JSON payload. A separate stream is required for every attempt. +#[derive(Clone, Copy, Debug)] +pub struct ExpectedPeer { + pub pid: u32, + pub uid: u32, + pub gid: u32, +} + +pub struct ControlChannel { + stream: UnixStream, + deadline: Instant, + poisoned: bool, +} + +impl ControlChannel { + pub fn new(stream: UnixStream, peer: ExpectedPeer, deadline: Instant) -> io::Result { + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|duration| !duration.is_zero() && *duration <= MAX_CONTROL_TIMEOUT) + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "invalid control deadline") + })?; + let credentials = rustix::net::sockopt::socket_peercred(&stream)?; + if credentials.pid.as_raw_pid() as u32 != peer.pid + || credentials.uid.as_raw() != peer.uid + || credentials.gid.as_raw() != peer.gid + { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "unexpected control peer", + )); + } + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(remaining))?; + stream.set_write_timeout(Some(remaining))?; + Ok(Self { + stream, + deadline, + poisoned: false, + }) + } + + pub fn send(&mut self, message: &T) -> io::Result<()> { + let result = self.send_inner(message); + self.finish(result) + } + + pub fn receive(&mut self) -> io::Result { + let result = self.receive_inner(); + self.finish(result) + } + + fn send_inner(&mut self, message: &T) -> io::Result<()> { + self.remaining()?; + if !message.valid() { + return Err(invalid("invalid control message")); + } + let mut body = FixedBuffer::default(); + serde_json::to_writer(&mut body, message) + .map_err(|_| invalid("control message too large"))?; + let mut header = [0u8; 8]; + header[..4].copy_from_slice(&PREFIX); + header[4..].copy_from_slice(&(body.len as u32).to_be_bytes()); + self.write_bytes(&header)?; + self.write_bytes(&body.bytes[..body.len]) + } + + fn receive_inner(&mut self) -> io::Result { + let mut header = [0u8; 8]; + self.read_bytes(&mut header)?; + let length = decode_header(header)?; + let mut bytes = [0u8; MAX_MESSAGE_BYTES]; + self.read_bytes(&mut bytes[..length])?; + let message: T = serde_json::from_slice(&bytes[..length]) + .map_err(|_| invalid("invalid control JSON"))?; + if !message.valid() { + return Err(invalid("invalid control message")); + } + Ok(message) + } + + fn remaining(&self) -> io::Result { + if self.poisoned { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "control channel failed", + )); + } + self.deadline + .checked_duration_since(Instant::now()) + .filter(|duration| !duration.is_zero()) + .ok_or_else(|| io::Error::new(io::ErrorKind::TimedOut, "control deadline expired")) + } + + fn read_bytes(&mut self, mut bytes: &mut [u8]) -> io::Result<()> { + while !bytes.is_empty() { + self.stream.set_read_timeout(Some(self.remaining()?))?; + match self.stream.read(bytes) { + Ok(0) => return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "control EOF")), + Ok(count) => bytes = &mut bytes[count..], + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(error) => return Err(error), + } + } + Ok(()) + } + + fn write_bytes(&mut self, mut bytes: &[u8]) -> io::Result<()> { + while !bytes.is_empty() { + self.stream.set_write_timeout(Some(self.remaining()?))?; + match self.stream.write(bytes) { + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::WriteZero, + "control write failed", + )); + } + Ok(count) => bytes = &bytes[count..], + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(error) => return Err(error), + } + } + Ok(()) + } + + fn finish(&mut self, result: io::Result) -> io::Result { + if result.is_err() { + self.poisoned = true; + let _ = self.stream.shutdown(Shutdown::Both); + } + result + } +} + +fn invalid(message: &'static str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + +fn decode_header(header: [u8; 8]) -> io::Result { + if header[..4] != PREFIX { + return Err(invalid("unsupported control framing")); + } + let length = + u32::from_be_bytes(header[4..].try_into().expect("four-byte header length")) as usize; + if length == 0 || length > MAX_MESSAGE_BYTES { + return Err(invalid("invalid control frame length")); + } + Ok(length) +} + +struct FixedBuffer { + bytes: [u8; MAX_MESSAGE_BYTES], + len: usize, +} + +impl Default for FixedBuffer { + fn default() -> Self { + Self { + bytes: [0; MAX_MESSAGE_BYTES], + len: 0, + } + } +} + +impl Write for FixedBuffer { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if bytes.len() > MAX_MESSAGE_BYTES - self.len { + return Err(invalid("control message too large")); + } + self.bytes[self.len..self.len + bytes.len()].copy_from_slice(bytes); + self.len += bytes.len(); + Ok(bytes.len()) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn header_rejects_oversized_length_before_body_read() { + for length in [0, 1025, u32::MAX] { + let mut header = [0u8; 8]; + header[..4].copy_from_slice(&PREFIX); + header[4..].copy_from_slice(&length.to_be_bytes()); + assert!(decode_header(header).is_err()); + } + } + + #[test] + fn header_rejects_version_and_reserved_flags() { + for prefix in [[b'K', b'F', 2, 0], [b'K', b'F', 1, 1], [0; 4]] { + let mut header = [0u8; 8]; + header[..4].copy_from_slice(&prefix); + header[4..].copy_from_slice(&1u32.to_be_bytes()); + assert!(decode_header(header).is_err()); + } + } + + #[test] + fn serializer_has_fixed_storage() { + let mut buffer = FixedBuffer::default(); + buffer.write_all(&[0; MAX_MESSAGE_BYTES]).unwrap(); + assert!(buffer.write_all(b"extra").is_err()); + assert_eq!(buffer.len, MAX_MESSAGE_BYTES); + } + + #[test] + fn schemas_reject_unknown_fields_and_direction() { + assert!( + serde_json::from_str::(r#"{"command":"cancel","exec":true}"#).is_err() + ); + assert!(serde_json::from_str::(r#"{"command":"cancel"}"#).is_err()); + assert!(serde_json::from_str::(r#"{"command":"release_exec"}"#).is_err()); + } + + #[test] + fn identifiers_and_thread_claims_are_bounded() { + for run_id in ["", "../work", "bad id"] { + assert!( + !ParentMessage::Initialize { + run_id: run_id.into() + } + .valid() + ); + } + assert!( + !ParentMessage::Initialize { + run_id: "x".repeat(65) + } + .valid() + ); + assert!(!ChildMessage::Hello { pid: 1, threads: 2 }.valid()); + assert!(!ChildMessage::Hello { pid: 0, threads: 1 }.valid()); + } +} diff --git a/judge/crates/sandbox/src/main.rs b/judge/crates/sandbox/src/main.rs index 7810e1e..8036895 100644 --- a/judge/crates/sandbox/src/main.rs +++ b/judge/crates/sandbox/src/main.rs @@ -28,6 +28,13 @@ enum Command { }, /// Reserved execution entrypoint; refuses execution until isolation is implemented. Run, + /// Internal trusted launcher control entry; never runs a user program. + #[cfg(target_os = "linux")] + #[command(hide = true)] + Launcher { + #[arg(long)] + control_socket: PathBuf, + }, } fn main() -> ExitCode { @@ -50,6 +57,18 @@ fn main() -> ExitCode { } Command::Validate { spec } => validate_spec(spec), Command::CheckRuntime { config } => check_runtime(config), + #[cfg(target_os = "linux")] + Command::Launcher { control_socket } => { + match kf_sandbox::launcher::entry::serve(&control_socket) { + Ok(()) => ExitCode::from(2), + Err(error) => spec_error( + "launcher_control_failed", + Some(serde_json::json!({ + "io_kind": format!("{:?}", error.kind()) + })), + ), + } + } } } diff --git a/judge/crates/sandbox/tests/launcher_control.rs b/judge/crates/sandbox/tests/launcher_control.rs new file mode 100644 index 0000000..9eb8162 --- /dev/null +++ b/judge/crates/sandbox/tests/launcher_control.rs @@ -0,0 +1,303 @@ +#![cfg(target_os = "linux")] + +use kf_sandbox::launcher::protocol::{ChildMessage, ControlChannel, ExpectedPeer, ParentMessage}; +use kf_sandbox::outcome::{ErrorCode, ExecutionStage}; +use std::fs::{self, DirBuilder}; +use std::io::{self, Write}; +use std::os::unix::fs::DirBuilderExt; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::PathBuf; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +static ATTEMPT: AtomicU64 = AtomicU64::new(0); + +struct Attempt { + directory: PathBuf, + child: Option, +} + +impl Attempt { + fn start() -> (Self, ControlChannel, u32) { + let directory = std::env::temp_dir().join(format!( + "kf-l-{}-{}", + std::process::id(), + ATTEMPT.fetch_add(1, Ordering::Relaxed) + )); + DirBuilder::new().mode(0o700).create(&directory).unwrap(); + let mut attempt = Self { + directory, + child: None, + }; + let socket = attempt.directory.join("control.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + listener.set_nonblocking(true).unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + let child = Command::new(env!("CARGO_BIN_EXE_kf-sandbox")) + .arg("launcher") + .arg("--control-socket") + .arg(&socket) + .env_clear() + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let pid = child.id(); + attempt.child = Some(child); + let stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < deadline, + "launcher must connect within deadline" + ); + assert!( + attempt + .child + .as_mut() + .unwrap() + .try_wait() + .unwrap() + .is_none() + ); + std::thread::sleep(Duration::from_millis(5)); + } + Err(error) => panic!("cannot accept launcher: {error}"), + } + }; + let channel = ControlChannel::new(stream, own_peer(pid), deadline).unwrap(); + assert_eq!( + fs::read_dir(format!("/proc/{pid}/task")).unwrap().count(), + 1 + ); + (attempt, channel, pid) + } + + fn reap(&mut self) { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(status) = self.child.as_mut().unwrap().try_wait().unwrap() { + assert_eq!( + status.code(), + Some(2), + "entry never reports execution success" + ); + self.child = None; + return; + } + assert!( + Instant::now() < deadline, + "launcher must exit and be reaped" + ); + std::thread::sleep(Duration::from_millis(5)); + } + } +} + +impl Drop for Attempt { + fn drop(&mut self) { + if let Some(child) = &mut self.child { + let _ = child.kill(); + let _ = child.wait(); + } + let _ = fs::remove_file(self.directory.join("control.sock")); + let _ = fs::remove_dir(&self.directory); + } +} + +fn own_peer(pid: u32) -> ExpectedPeer { + ExpectedPeer { + pid, + uid: rustix::process::geteuid().as_raw(), + gid: rustix::process::getegid().as_raw(), + } +} + +fn pair() -> (ControlChannel, UnixStream) { + let (stream, writer) = UnixStream::pair().unwrap(); + ( + ControlChannel::new( + stream, + own_peer(std::process::id()), + Instant::now() + Duration::from_secs(2), + ) + .unwrap(), + writer, + ) +} + +#[test] +fn fresh_single_thread_entry_exchanges_authenticated_messages_then_refuses() { + let (mut attempt, mut channel, pid) = Attempt::start(); + assert_eq!( + channel.receive::().unwrap(), + ChildMessage::Hello { pid, threads: 1 } + ); + channel + .send(&ParentMessage::Initialize { + run_id: "attempt-001".into(), + }) + .unwrap(); + assert_eq!( + channel.receive::().unwrap(), + ChildMessage::InitializationRefused { + run_id: "attempt-001".into(), + stage: ExecutionStage::Preparation, + code: ErrorCode::CapabilityUnavailable, + } + ); + attempt.reap(); + assert_eq!( + channel.receive::().unwrap_err().kind(), + io::ErrorKind::UnexpectedEof + ); +} + +#[test] +fn parent_can_cancel_before_initialization_and_reap_entry() { + let (mut attempt, mut channel, pid) = Attempt::start(); + assert_eq!( + channel.receive::().unwrap(), + ChildMessage::Hello { pid, threads: 1 } + ); + channel.send(&ParentMessage::Cancel {}).unwrap(); + assert_eq!( + channel.receive::().unwrap(), + ChildMessage::Cancelled {} + ); + attempt.reap(); +} + +#[test] +fn parent_disconnect_does_not_leave_a_waiting_launcher() { + let (mut attempt, mut channel, _) = Attempt::start(); + assert!(matches!( + channel.receive::().unwrap(), + ChildMessage::Hello { .. } + )); + drop(channel); + attempt.reap(); +} + +#[test] +fn kernel_credentials_reject_an_unexpected_peer() { + for peer in [ + ExpectedPeer { + pid: std::process::id() + 1, + ..own_peer(std::process::id()) + }, + ExpectedPeer { + uid: rustix::process::geteuid().as_raw() + 1, + ..own_peer(std::process::id()) + }, + ExpectedPeer { + gid: rustix::process::getegid().as_raw() + 1, + ..own_peer(std::process::id()) + }, + ] { + let (stream, _writer) = UnixStream::pair().unwrap(); + let result = ControlChannel::new(stream, peer, Instant::now() + Duration::from_secs(1)); + assert!(matches!(result, Err(error) if error.kind() == io::ErrorKind::PermissionDenied)); + } +} + +#[test] +fn truncated_frame_poisoning_prevents_resynchronization() { + let (mut channel, mut writer) = pair(); + writer.write_all(b"KF\x01").unwrap(); + drop(writer); + assert_eq!( + channel.receive::().unwrap_err().kind(), + io::ErrorKind::UnexpectedEof + ); + assert_eq!( + channel.send(&ParentMessage::Cancel {}).unwrap_err().kind(), + io::ErrorKind::BrokenPipe + ); +} + +#[test] +fn two_frames_remain_separate_and_keep_their_direction() { + let (mut channel, stream) = pair(); + let mut writer = ControlChannel::new( + stream, + own_peer(std::process::id()), + Instant::now() + Duration::from_secs(2), + ) + .unwrap(); + writer + .send(&ParentMessage::Initialize { + run_id: "run-1".into(), + }) + .unwrap(); + writer.send(&ParentMessage::Cancel {}).unwrap(); + assert_eq!( + channel.receive::().unwrap(), + ParentMessage::Initialize { + run_id: "run-1".into() + } + ); + assert_eq!( + channel.receive::().unwrap(), + ParentMessage::Cancel {} + ); +} + +#[test] +fn unknown_fields_poison_channel_before_any_later_command() { + let (mut channel, mut writer) = pair(); + let bytes = br#"{"command":"cancel","extra":true}"#; + writer.write_all(b"KF\x01\x00").unwrap(); + writer + .write_all(&(bytes.len() as u32).to_be_bytes()) + .unwrap(); + writer.write_all(bytes).unwrap(); + assert_eq!( + channel.receive::().unwrap_err().kind(), + io::ErrorKind::InvalidData + ); + assert_eq!( + channel.receive::().unwrap_err().kind(), + io::ErrorKind::BrokenPipe + ); +} + +#[test] +fn partial_body_cannot_reset_the_absolute_deadline() { + let (stream, mut writer) = UnixStream::pair().unwrap(); + let mut channel = ControlChannel::new( + stream, + own_peer(std::process::id()), + Instant::now() + Duration::from_millis(80), + ) + .unwrap(); + writer.write_all(b"KF\x01\x00\x00\x00\x00\x10{").unwrap(); + let started = Instant::now(); + let error = channel.receive::().unwrap_err(); + assert!(matches!( + error.kind(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock + )); + assert!(started.elapsed() < Duration::from_secs(1)); + assert_eq!( + channel.send(&ParentMessage::Cancel {}).unwrap_err().kind(), + io::ErrorKind::BrokenPipe + ); +} + +#[test] +fn deadline_must_be_finite_short_and_in_the_future() { + for deadline in [ + Instant::now() - Duration::from_secs(1), + Instant::now() + Duration::from_secs(30), + ] { + let (stream, _writer) = UnixStream::pair().unwrap(); + assert!( + matches!(ControlChannel::new(stream, own_peer(std::process::id()), deadline), + Err(error) if error.kind() == io::ErrorKind::InvalidInput) + ); + } +}