feat(sandbox): retain replay records across explicit factory cycles

This commit is contained in:
2026-10-13 01:36:36 +08:00
parent a55de23fd6
commit 033d31d420
5 changed files with 306 additions and 25 deletions
+12
View File
@@ -39,6 +39,18 @@ API 存活检查和前端预览不依赖这些容器,可以先独立启动。
回收后安排新记录周期,不自动删除记录。清理失败保留身份租约并停止新分配;
重启只根据已记录归属和 inode 回收,未知/外来内容保留并拒绝启动。
工厂 `--max-requests` 由可信 Root 启动参数选择,范围1–4096,默认4096;
它限制当前记录周期的受理次数,不改变学生预算或已记录 ID 的拒绝规则。
`kf-sandbox rotate-factory-cycle --config ROOT_CONFIG --cycle-id ID` 是独立的
Root 维护入口:在受限监督器作用域中,核对受保护配置/镜像、取得工厂排他锁,
并读回任务、身份、捕获目录和委派 cgroup 为空、控制 socket 已消失后,才把
当前请求记录保留到 Root 私有 `factory/cycles/`,fsync文件与目录后清空当前记录。
最多保留64个有界周期;历史 ID 在后续启动仍用于拒绝重放。正常启动、重启、
寿命结束和容量用尽不会调用轮换或删除任何历史记录。归档完成但当前记录未清空
的中断状态拒绝启动,需在停止/排空后显式使用同一个 cycle ID 重试;未知或
不完整记录保留并拒绝,不能通过删日志恢复。该入口不启动服务,也不证明隔离。
新安装路径下的实际轮换、崩溃恢复、跨周期重放拒绝与资源读回仍需独立 VM 验收。
`kf-worker judge-submission` 是受控本地适配入口。常驻 `kf-worker serve --config`
使用有界 AMQP 通知和经过服务身份认证的 Coordinator claim;配置与依赖见
[判题工程说明](../judge/README.md#单机持久化协调与常驻-worker)。
+31 -23
View File
@@ -1,6 +1,7 @@
//! 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::factory_cycles::MAX_IDS;
use crate::{
artifacts::ArtifactRoot, execution, launcher::protocol::WireMessage, rpc, spec::ExecutionSpec,
};
@@ -10,9 +11,8 @@ use rustix::{
};
use serde::{Deserialize, Serialize};
use std::{
collections::HashSet,
fs::File,
io::{self, Read, Write},
io::{self, Write},
os::unix::{
fs::{FileTypeExt, MetadataExt, PermissionsExt},
net::{UnixListener, UnixStream},
@@ -21,7 +21,6 @@ use std::{
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")
}
@@ -220,7 +219,7 @@ struct Active {
child: Child,
_guardian: UnixStream,
}
fn private_file(path: &Path) -> io::Result<File> {
pub(crate) fn private_file(path: &Path) -> io::Result<File> {
let fd = match rustix::fs::open(
path,
OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
@@ -277,7 +276,7 @@ fn idle_recovery(config: &crate::runtime::ValidatedRuntimeConfig) -> io::Result<
std::thread::sleep(Duration::from_millis(10));
}
}
fn drained(config: &crate::runtime::ValidatedRuntimeConfig) -> io::Result<bool> {
pub(crate) fn drained(config: &crate::runtime::ValidatedRuntimeConfig) -> io::Result<bool> {
let p = config.config();
for root in [
PathBuf::from(&p.tasks_root),
@@ -305,10 +304,7 @@ fn drained(config: &crate::runtime::ValidatedRuntimeConfig) -> io::Result<bool>
/// 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());
}
pub(crate) fn clean_environment() -> io::Result<()> {
// Root management receives no ambient business/credential environment.
if std::env::vars_os().any(|(k, v)| {
!matches!(
@@ -320,6 +316,23 @@ pub fn serve(config_path: &Path, lifetime_seconds: u64) -> io::Result<()> {
}) {
return Err(denied());
}
Ok(())
}
pub fn serve(config_path: &Path, lifetime_seconds: u64) -> io::Result<()> {
serve_with_limit(config_path, lifetime_seconds, MAX_IDS)
}
/// Limits are selected by the trusted operator, never by a Worker request.
pub fn serve_with_limit(
config_path: &Path,
lifetime_seconds: u64,
max_requests: usize,
) -> io::Result<()> {
if !(1..=86400).contains(&lifetime_seconds) || !(1..=MAX_IDS).contains(&max_requests) {
return Err(denied());
}
clean_environment()?;
let config = execution::load_config(config_path)?;
execution::authorize_registry(&config)?;
let policy = config.config();
@@ -374,22 +387,16 @@ pub fn serve(config_path: &Path, lifetime_seconds: u64) -> io::Result<()> {
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()) {
let current =
crate::factory_cycles::parse_ids(&crate::factory_cycles::read_ledger(&mut ledger)?)?;
let mut used = current.len();
let mut ids = crate::factory_cycles::historical_ids(&root)?;
for id in current {
if !ids.insert(id) {
// Incomplete explicit rotation is preserved for operator recovery.
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 {
@@ -463,7 +470,7 @@ pub fn serve(config_path: &Path, lifetime_seconds: u64) -> io::Result<()> {
continue;
}
// Replay/ledger exhaustion is terminal even while all slots are busy.
if ids.len() >= MAX_IDS || ids.contains(&request.run_id) {
if used >= max_requests || ids.contains(&request.run_id) {
let _ = channel.send(&Allocation::Refused {});
continue;
}
@@ -474,6 +481,7 @@ pub fn serve(config_path: &Path, lifetime_seconds: u64) -> io::Result<()> {
// Persist before allocating; IDs remain consumed across restarts.
writeln!(ledger, "{}", request.run_id)?;
ledger.sync_all()?;
used += 1;
ids.insert(request.run_id.clone());
let (guardian, child_guardian) = UnixStream::pair()?;
let child = Command::new(&binary)
+232
View File
@@ -0,0 +1,232 @@
//! Explicit stopped-factory ledger rotation. Historical IDs still reject replay.
//! No rotation is performed by startup, restart, expiry or allocation.
use crate::runtime::ValidatedRuntimeConfig;
use rustix::fs::FlockOperation;
use std::collections::HashSet;
use std::fs::File;
use std::io::{self, Read, Write};
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use std::path::{Path, PathBuf};
pub(crate) const MAX_IDS: usize = 4096;
const MAX_CYCLES: usize = 64;
const MAX_LEDGER_BYTES: u64 = (MAX_IDS * 65) as u64;
fn denied() -> io::Error {
io::Error::new(io::ErrorKind::PermissionDenied, "factory cycle refused")
}
pub(crate) fn parse_ids(bytes: &[u8]) -> io::Result<HashSet<String>> {
if bytes.len() as u64 > MAX_LEDGER_BYTES {
return Err(denied());
}
let text = std::str::from_utf8(bytes).map_err(|_| denied())?;
if !text.is_empty() && !text.ends_with('\n') {
return Err(denied());
}
let mut ids = HashSet::new();
for id in text.lines() {
if crate::spec::identifier(id, "run_id").is_err() || !ids.insert(id.to_owned()) {
return Err(denied());
}
}
if ids.len() > MAX_IDS {
return Err(denied());
}
Ok(ids)
}
pub(crate) fn read_ledger(file: &mut File) -> io::Result<Vec<u8>> {
if file.metadata()?.len() > MAX_LEDGER_BYTES {
return Err(denied());
}
let mut bytes = Vec::new();
file.take(MAX_LEDGER_BYTES + 1).read_to_end(&mut bytes)?;
parse_ids(&bytes)?;
Ok(bytes)
}
fn cycle_name(name: &str) -> io::Result<&str> {
let id = name
.strip_prefix("cycle-")
.and_then(|s| s.strip_suffix(".requests"));
match id {
Some(id) if crate::spec::identifier(id, "cycle_id").is_ok() => Ok(id),
_ => Err(denied()),
}
}
fn archives(root: &Path) -> io::Result<Vec<PathBuf>> {
let directory = root.join("cycles");
if !directory.try_exists()? {
if std::fs::symlink_metadata(&directory).is_ok() {
return Err(denied());
}
return Ok(Vec::new());
}
crate::isolation::root::open_directory(&directory)?;
if std::fs::metadata(&directory)?.mode() & 0o7777 != 0o700 {
return Err(denied());
}
let mut paths = Vec::new();
for entry in std::fs::read_dir(&directory)? {
let entry = entry?;
let name = entry.file_name();
cycle_name(name.to_str().ok_or_else(denied)?)?;
let info = std::fs::symlink_metadata(entry.path())?;
if !info.is_file()
|| info.uid() != 0
|| info.gid() != 0
|| info.mode() & 0o7777 != 0o600
|| info.nlink() != 1
|| info.len() == 0
|| info.len() > MAX_LEDGER_BYTES
{
return Err(denied());
}
paths.push(entry.path());
if paths.len() > MAX_CYCLES {
return Err(denied());
}
}
paths.sort();
Ok(paths)
}
pub(crate) fn historical_ids(root: &Path) -> io::Result<HashSet<String>> {
let mut all = HashSet::new();
for path in archives(root)? {
let mut file = crate::factory::private_file(&path)?;
for id in parse_ids(&read_ledger(&mut file)?)? {
if !all.insert(id) {
return Err(denied());
}
}
}
Ok(all)
}
/// An operator uses a bounded root maintenance scope and the same protected
/// runtime/images. The factory lock and actual empty resources are mandatory.
/// An archived-but-not-truncated interruption requires repeating this explicit
/// operation with the same cycle ID; ordinary factory startup refuses it.
pub fn rotate(config_path: &Path, cycle_id: &str) -> io::Result<serde_json::Value> {
crate::spec::identifier(cycle_id, "cycle_id").map_err(|_| denied())?;
crate::factory::clean_environment()?;
let config: ValidatedRuntimeConfig = crate::execution::load_config(config_path)?;
crate::execution::authorize_registry(&config)?;
let root = Path::new(&config.config().state_root).join("factory");
crate::isolation::root::open_directory(&root)?;
let lock = crate::factory::private_file(&root.join("lock"))?;
rustix::fs::flock(&lock, FlockOperation::NonBlockingLockExclusive)?;
if !crate::factory::drained(&config)?
|| std::fs::symlink_metadata(root.join("control.sock")).is_ok()
{
return Err(denied());
}
let directory = root.join("cycles");
let old = archives(&root)?;
let target = directory.join(format!("cycle-{cycle_id}.requests"));
let exists = old.contains(&target);
if !exists && old.len() == MAX_CYCLES {
return Err(denied());
}
let mut ledger = crate::factory::private_file(&root.join("requests"))?;
let bytes = read_ledger(&mut ledger)?;
let current = parse_ids(&bytes)?;
let mut previous = HashSet::new();
let mut archived_target = None;
for path in old {
let mut file = crate::factory::private_file(&path)?;
let archived = read_ledger(&mut file)?;
if path == target {
archived_target = Some(archived.clone());
}
for id in parse_ids(&archived)? {
if !previous.insert(id.clone()) || (path != target && current.contains(&id)) {
return Err(denied());
}
}
}
if let Some(archived) = archived_target {
if !bytes.is_empty() && archived != bytes {
return Err(denied());
}
// Empty current ledger means this explicit rotation already finished.
// Its historical IDs remain intact and counted on the next startup.
} else {
if bytes.is_empty() {
return Err(denied());
}
if !directory.try_exists()? {
std::fs::create_dir(&directory)?;
std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o700))?;
}
let fd = crate::isolation::root::open_directory(&directory)?;
let mut archive = crate::factory::private_file(&target)?;
if archive.metadata()?.len() != 0 {
return Err(denied());
}
archive.write_all(&bytes)?;
archive.sync_all()?;
rustix::fs::fsync(&fd)?;
}
// Never erase a ledger before the retained copy is durable and matches.
let mut archived = crate::factory::private_file(&target)?;
let retained = read_ledger(&mut archived)?;
if !bytes.is_empty() && retained != bytes {
return Err(denied());
}
archived.sync_all()?;
let archive_fd = crate::isolation::root::open_directory(&directory)?;
rustix::fs::fsync(&archive_fd)?;
let root_fd = crate::isolation::root::open_directory(&root)?;
// Persist the newly created archive directory in its parent before the
// active ledger can be truncated, including an explicit interrupted retry.
rustix::fs::fsync(&root_fd)?;
if !crate::factory::drained(&config)? {
return Err(denied());
}
ledger.set_len(0)?;
ledger.sync_all()?;
rustix::fs::fsync(&root_fd)?;
Ok(serde_json::json!({
"cycle_id": cycle_id, "archived_requests": parse_ids(&retained)?.len(),
"current_requests": 0, "historical_replay_ids_retained": true,
"services_started": false, "isolation_verified": false,
}))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ledger_parsing_keeps_unique_complete_bounded_identifiers() {
assert!(parse_ids(b"").unwrap().is_empty());
assert_eq!(parse_ids(b"run-one\nrun-two\n").unwrap().len(), 2);
for bytes in [
b"run-one".as_slice(),
b"run-one\nrun-one\n",
b"../foreign\n",
b"\xff\n",
] {
assert!(parse_ids(bytes).is_err());
}
assert!(parse_ids(&vec![b'x'; MAX_LEDGER_BYTES as usize + 1]).is_err());
}
#[test]
fn cycle_names_cannot_choose_an_external_path() {
assert_eq!(cycle_name("cycle-cycle-one.requests").unwrap(), "cycle-one");
for name in [
"requests",
"cycle-../foreign.requests",
"cycle-.requests",
"cycle-one.tmp",
] {
assert!(cycle_name(name).is_err());
}
}
}
+2
View File
@@ -39,3 +39,5 @@ pub mod rpc;
#[cfg(target_os = "linux")]
pub mod factory;
#[cfg(target_os = "linux")]
pub mod factory_cycles;
+29 -2
View File
@@ -53,6 +53,15 @@ enum Command {
config: PathBuf,
#[arg(long, default_value_t = 3600)]
lifetime_seconds: u64,
#[arg(long, default_value_t = 4096)]
max_requests: usize,
},
/// Explicit root maintenance: retain historical IDs after proven drain.
RotateFactoryCycle {
#[arg(long)]
config: PathBuf,
#[arg(long)]
cycle_id: String,
},
#[cfg(target_os = "linux")]
#[command(hide = true)]
@@ -160,10 +169,13 @@ fn main() -> ExitCode {
Command::Factory {
config,
lifetime_seconds,
max_requests,
} => {
#[cfg(target_os = "linux")]
{
if kf_sandbox::factory::serve(&config, lifetime_seconds).is_ok() {
if kf_sandbox::factory::serve_with_limit(&config, lifetime_seconds, max_requests)
.is_ok()
{
ExitCode::SUCCESS
} else {
spec_error("factory_refused", None)
@@ -171,7 +183,22 @@ fn main() -> ExitCode {
}
#[cfg(not(target_os = "linux"))]
{
let _ = (config, lifetime_seconds);
let _ = (config, lifetime_seconds, max_requests);
spec_error("unsupported_platform", None)
}
}
Command::RotateFactoryCycle { config, cycle_id } => {
#[cfg(target_os = "linux")]
match kf_sandbox::factory_cycles::rotate(&config, &cycle_id) {
Ok(report) => {
println!("{report}");
ExitCode::SUCCESS
}
Err(_) => spec_error("factory_cycle_refused", None),
}
#[cfg(not(target_os = "linux"))]
{
let _ = (config, cycle_id);
spec_error("unsupported_platform", None)
}
}