From 71918399c932d5850ac366ab1a84b3c109f2eca2 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Tue, 13 Oct 2026 04:11:58 +0800 Subject: [PATCH] feat(delivery): observe bounded node readiness and owner state --- .../src/kf_oj/modules/problems/operations.py | 64 ++ backend/src/kf_oj/workers/problem_status.py | 33 + backend/tests/test_delivery.py | 118 ++++ judge/crates/coordinator/src/api.rs | 25 + judge/crates/coordinator/src/lib.rs | 1 + judge/crates/coordinator/src/operations.rs | 91 +++ judge/crates/sandbox/src/cgroup.rs | 39 ++ judge/crates/sandbox/src/factory.rs | 124 +++- judge/crates/sandbox/src/factory_health.rs | 63 ++ judge/crates/sandbox/src/lib.rs | 2 + judge/crates/sandbox/src/main.rs | 23 + scripts/release/README.md | 34 + scripts/release/broker_health.py | 218 +++++++ scripts/release/factory_health.py | 46 ++ scripts/release/import_health.py | 44 ++ scripts/release/node_health.py | 584 ++++++++++++++++++ scripts/release/operator_units.py | 231 +++++++ 17 files changed, 1738 insertions(+), 2 deletions(-) create mode 100644 backend/src/kf_oj/modules/problems/operations.py create mode 100644 backend/src/kf_oj/workers/problem_status.py create mode 100644 judge/crates/coordinator/src/operations.rs create mode 100644 judge/crates/sandbox/src/factory_health.rs create mode 100644 scripts/release/broker_health.py create mode 100644 scripts/release/factory_health.py create mode 100644 scripts/release/import_health.py create mode 100644 scripts/release/node_health.py create mode 100644 scripts/release/operator_units.py diff --git a/backend/src/kf_oj/modules/problems/operations.py b/backend/src/kf_oj/modules/problems/operations.py new file mode 100644 index 0000000..f1003d8 --- /dev/null +++ b/backend/src/kf_oj/modules/problems/operations.py @@ -0,0 +1,64 @@ +"""Read-only import projection for the ordinary operator CLI; no private archive data.""" + +from sqlalchemy import text + +from kf_oj.modules.problems.import_models import ImportPool +from kf_oj.modules.problems.imports import ImportUnavailable, policy_digest + + +class ImportOperationQueries: + def __init__(self, sessions, policy): + self.sessions, self.policy = sessions, policy + + async def status(self) -> dict: + async with self.sessions() as session, session.begin(): + await session.execute(text("SET TRANSACTION READ ONLY")) + await session.execute(text("SET LOCAL statement_timeout='1500'")) + await session.execute(text("SET LOCAL lock_timeout='500'")) + authorized = await session.scalar( + text(""" + SELECT rolcanlogin AND NOT(rolsuper OR rolcreatedb OR rolcreaterole + OR rolreplication OR rolbypassrls) + AND pg_has_role(current_user,'kf_oj_problem_import','USAGE') + AND NOT pg_has_role(current_user,'kf_oj_platform','MEMBER') + AND NOT pg_has_role(current_user,'kf_oj_coordinator','MEMBER') + AND NOT pg_has_role(current_user,'kf_oj_identity_operator','MEMBER') + AND NOT pg_has_role(current_user,'kf_oj_identity_authority_owner','MEMBER') + FROM pg_roles WHERE rolname=session_user + """) + ) + if authorized is not True: + raise ImportUnavailable("dedicated import observation role required") + pool = await session.get(ImportPool, 1) + row = ( + ( + await session.execute( + text(""" + SELECT clock_timestamp() now, + (SELECT count(*) FROM (SELECT 1 FROM problem_import_operations + WHERE state='uploaded' LIMIT 129) q) uploaded, + (SELECT count(*) FROM (SELECT 1 FROM problem_import_operations + WHERE state='validating' LIMIT 129) q) validating, + (SELECT coalesce(sum(reserved_bytes),0) FROM ( + SELECT reserved_bytes FROM problem_import_operations + WHERE reserved_bytes>0 LIMIT 129) q) actual_reserved + """) + ) + ) + .mappings() + .one() + ) + if pool is None or pool.reserved_bytes != row["actual_reserved"]: + raise ImportUnavailable("import reservation consistency unknown") + return { + "schema_version": 1, + "worker_ready": bool( + pool.worker_ready_until + and pool.worker_ready_until > row["now"] + and pool.worker_policy_digest == policy_digest(self.policy) + ), + "reserved_bytes": pool.reserved_bytes, + "uploaded_capped": min(row["uploaded"], 128), + "validating_capped": min(row["validating"], 128), + "truncated": row["uploaded"] > 128 or row["validating"] > 128, + } diff --git a/backend/src/kf_oj/workers/problem_status.py b/backend/src/kf_oj/workers/problem_status.py new file mode 100644 index 0000000..1752817 --- /dev/null +++ b/backend/src/kf_oj/workers/problem_status.py @@ -0,0 +1,33 @@ +"""Ordinary import-only read status; no parser, staging mutation or judging writes.""" + +import asyncio +import json + +from kf_oj.core.settings import Settings +from kf_oj.infra.db.session import build_engine, build_session_factory +from kf_oj.modules.problems.operations import ImportOperationQueries + + +async def run() -> dict: + settings = Settings() + if settings.judge_admission is None: + raise ValueError("import admission required") + engine = build_engine(settings) + try: + return await ImportOperationQueries( + build_session_factory(engine), settings.judge_admission + ).status() + finally: + await engine.dispose() + + +def main() -> None: + try: + result = asyncio.run(asyncio.wait_for(run(), timeout=6)) + print(json.dumps(result, sort_keys=True)) + except Exception: + raise SystemExit("import status unavailable") from None + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index 19fdd60..79045ea 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -23,6 +23,8 @@ factory_service = importlib.import_module("scripts.release.factory_service") dependencies = importlib.import_module("scripts.release.dependencies") dependency_install = importlib.import_module("scripts.release.dependency_install") dependency_initialize = importlib.import_module("scripts.release.dependency_initialize") +node_health = importlib.import_module("scripts.release.node_health") +broker_health = importlib.import_module("scripts.release.broker_health") def test_delivery_authorities_and_layout(): @@ -725,3 +727,119 @@ def test_database_initialization_receipt_does_not_prove_migration_or_authenticat receipt["cluster_identity"]["uid"] = 0 with pytest.raises(policy.DeliveryError, match="cluster identity"): dependency_initialize.validate_receipt(receipt) + + +@pytest.mark.parametrize( + "field,value", + [ + ("superuser", True), + ("set_role", True), + ("connection_limit", -1), + ("login", "postgres"), + ("inherited", False), + ], +) +def test_readiness_rejects_wrong_or_excess_database_authority(field, value): + observed = { + "login": "kf_coordinator", + "database": "kf_onlinejudge", + "canlogin": True, + "superuser": False, + "createdb": False, + "createrole": False, + "replication": False, + "bypassrls": False, + "member": True, + "inherited": True, + "set_role": False, + "connection_limit": 8, + "privileges": node_health.PRIVILEGES["coordinator"].copy(), + } + node_health.validate_role("coordinator", observed) + observed[field] = value + with pytest.raises(policy.DeliveryError, match="role authority"): + node_health.validate_role("coordinator", observed) + + +def test_readiness_refuses_context_swap_or_claimed_isolation_certificate(): + value = { + "schema_version": 1, + "configuration_sha256": "b" * 64, + "supervisor_authorized": True, + "execution_gate": "required_per_execution", + "isolation_verified": False, + "max_concurrent_runs": 1, + "active_helpers": 0, + "cleanup_complete": True, + "remaining_requests": 4096, + "remaining_lifetime_seconds": 3600, + } + node_health.validate_factory(value, "b" * 64) + with pytest.raises(policy.DeliveryError, match="factory"): + node_health.validate_factory(value, "c" * 64) + value["isolation_verified"] = True + with pytest.raises(policy.DeliveryError, match="factory"): + node_health.validate_factory(value, "b" * 64) + + +def test_busy_unknown_cleanup_or_expiring_factory_never_admits_new_work(): + root = { + "cleanup_complete": True, + "active_helpers": 0, + "remaining_requests": 4096, + "remaining_lifetime_seconds": 3600, + } + counters = { + "in_flight": {"count_capped": 0, "truncated": False}, + "cleanup_unconfirmed": {"count_capped": 0, "truncated": False}, + } + admission = { + "legacy": {"max_job_wall_time_us": 60000000}, + "multi_runtime": {"max_job_wall_time_us": 60000000}, + } + assert node_health.admission_ready(root, counters, {"consumers": 1}, admission) + root["remaining_requests"] = 256 + assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission) + root["remaining_requests"] = 4096 + root["remaining_lifetime_seconds"] = 119 + assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission) + root["remaining_lifetime_seconds"] = 3600 + root["cleanup_complete"] = False + assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission) + root["cleanup_complete"] = True + counters["cleanup_unconfirmed"]["count_capped"] = 1 + assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission) + + +def test_actual_broker_argument_parser_keeps_typed_budgets_and_rejects_duplicates(): + value = [ + [key, "longstr" if key == "x-overflow" else "long", val] + for key, val in dependencies.QUEUE_ARGUMENTS.items() + ] + assert broker_health.queue_arguments(value) == dependencies.QUEUE_ARGUMENTS + value[-1] = value[0].copy() + with pytest.raises(policy.DeliveryError, match="duplicate"): + broker_health.queue_arguments(value) + with pytest.raises(policy.DeliveryError, match="budgets"): + broker_health.queue_arguments({**dependencies.QUEUE_ARGUMENTS, "x-max-length": 1024}) + + +def test_counter_projection_rejects_truncated_empty_and_source_admission_mismatch(): + value = { + "schema_version": 1, + "admission": {"fixture": True}, + "configured_workers": 1, + "lease_seconds": 120, + "counters": { + "queued": {"count_capped": 0, "truncated": False}, + "in_flight": {"count_capped": 0, "truncated": False}, + "cleanup_unconfirmed": {"count_capped": 0, "truncated": False}, + "database_clock_unix_ms": 1791830000000, + }, + } + node_health.validate_execution(value, value["admission"]) + with pytest.raises(policy.DeliveryError, match="admission"): + node_health.validate_execution(value, {"fixture": False}) + value["counters"]["cleanup_unconfirmed"]["truncated"] = True + with pytest.raises(policy.DeliveryError, match="bound"): + node_health.validate_execution(value, value["admission"]) diff --git a/judge/crates/coordinator/src/api.rs b/judge/crates/coordinator/src/api.rs index 758a7ff..5a61b0b 100644 --- a/judge/crates/coordinator/src/api.rs +++ b/judge/crates/coordinator/src/api.rs @@ -132,6 +132,29 @@ async fn ready(State(app): State) -> ApiResult> { app.repository.database.ready().await?; Ok(Json(json!({"status":"ready"}))) } +async fn operations(State(app): State, headers: HeaderMap) -> ApiResult> { + // The general internal guard also recognizes Workers; this projection is + // separately restricted to platform/operator authority, never worker tokens. + platform(&app, &headers)?; + let counters = crate::operations::snapshot(&app.repository.database).await?; + Ok(Json(json!({ + "schema_version": 1, + "admission": app.config.admission, + "configured_workers": app.config.workers.len(), + "lease_seconds": app.config.lease_seconds, + "counters": counters, + }))) +} +async fn notification_probe_id( + State(app): State, + headers: HeaderMap, +) -> ApiResult> { + platform(&app, &headers)?; + let id = crate::operations::absent_notification_id(&app.repository.database).await?; + Ok(Json( + json!({"schema_version":1,"evaluation_id":id,"absent":true}), + )) +} async fn claim(State(app): State, h: HeaderMap, b: Bytes) -> ApiResult { let w = worker(&app, &h)?; let n: JudgeTaskNotification = decode(&b)?; @@ -198,6 +221,8 @@ pub fn application(app: App) -> Router { Router::new() .route("/health/live", get(live)) .route("/health/ready", get(ready)) + .route("/internal/operations/status", get(operations)) + .route("/internal/operations/probe-id", get(notification_probe_id)) .route("/internal/claim", post(claim)) .route("/internal/renew", post(renew)) .route("/internal/stage", post(stage)) diff --git a/judge/crates/coordinator/src/lib.rs b/judge/crates/coordinator/src/lib.rs index 4c706be..613aacd 100644 --- a/judge/crates/coordinator/src/lib.rs +++ b/judge/crates/coordinator/src/lib.rs @@ -2,5 +2,6 @@ pub mod api; pub mod config; pub mod database; +pub mod operations; pub mod performance; pub mod repository; diff --git a/judge/crates/coordinator/src/operations.rs b/judge/crates/coordinator/src/operations.rs new file mode 100644 index 0000000..52e947d --- /dev/null +++ b/judge/crates/coordinator/src/operations.rs @@ -0,0 +1,91 @@ +//! Bounded read-only execution counters; no task identifiers or authority writes. +use crate::database::{Database, Error, Result}; +use serde::Serialize; + +#[derive(Debug, Serialize)] +pub struct BoundedCount { + pub count_capped: u16, + pub truncated: bool, +} +impl BoundedCount { + fn new(count: i64) -> Result { + if !(0..=129).contains(&count) { + return Err(Error::Invalid); + } + Ok(Self { + count_capped: count.min(128) as u16, + truncated: count == 129, + }) + } +} +#[derive(Debug, Serialize)] +pub struct ExecutionCounters { + pub queued: BoundedCount, + pub in_flight: BoundedCount, + pub cleanup_unconfirmed: BoundedCount, + pub database_clock_unix_ms: i64, +} +pub async fn snapshot(database: &Database) -> Result { + let mut session = database.session().await?; + let transaction = session + .client + .build_transaction() + .read_only(true) + .isolation_level(tokio_postgres::IsolationLevel::RepeatableRead) + .start() + .await?; + // LIMIT applies inside each count: a timeout or truncated projection can + // never be interpreted as empty capacity. The fixed DB deadline still + // bounds a planner that cannot produce the first 129 rows cheaply. + let row = transaction.query_one( + "SELECT + (SELECT count(*) FROM (SELECT 1 FROM evaluations WHERE state='queued' LIMIT 129) q) queued, + (SELECT count(*) FROM (SELECT 1 FROM evaluations WHERE state IN + ('claimed','compiling','running','cancelling') LIMIT 129) a) in_flight, + (SELECT count(*) FROM (SELECT 1 FROM evaluations e JOIN judge_attempts a + ON e.current_attempt_id=a.id AND a.evaluation_id=e.id + WHERE a.cleanup_complete IS DISTINCT FROM TRUE LIMIT 129) c) cleanup_unconfirmed, + floor(extract(epoch FROM clock_timestamp())*1000)::bigint database_clock_unix_ms", + &[], + ).await?; + let value = ExecutionCounters { + queued: BoundedCount::new(row.get("queued"))?, + in_flight: BoundedCount::new(row.get("in_flight"))?, + cleanup_unconfirmed: BoundedCount::new(row.get("cleanup_unconfirmed"))?, + database_clock_unix_ms: row.get("database_clock_unix_ms"), + }; + transaction.commit().await?; + Ok(value) +} + +pub async fn absent_notification_id(database: &Database) -> Result { + let session = database.session().await?; + for _ in 0..4 { + let identifier = uuid::Uuid::new_v4(); + if session + .client + .query_opt("SELECT 1 FROM evaluations WHERE id=$1", &[&identifier]) + .await? + .is_none() + { + return Ok(identifier); + } + } + Err(Error::Unavailable) +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn truncated_counts_never_become_an_empty_drain_proof() { + let empty = BoundedCount::new(0).unwrap(); + assert_eq!(empty.count_capped, 0); + assert!(!empty.truncated); + let crowded = BoundedCount::new(129).unwrap(); + assert_eq!(crowded.count_capped, 128); + assert!(crowded.truncated); + assert!(BoundedCount::new(-1).is_err()); + assert!(BoundedCount::new(130).is_err()); + } +} diff --git a/judge/crates/sandbox/src/cgroup.rs b/judge/crates/sandbox/src/cgroup.rs index 47172a3..94de847 100644 --- a/judge/crates/sandbox/src/cgroup.rs +++ b/judge/crates/sandbox/src/cgroup.rs @@ -59,6 +59,45 @@ pub struct DelegatedCgroup { } impl DelegatedCgroup { + /// Read-only status: never repairs a missing controller or rewrites limits. + pub fn inspect(config: &ValidatedRuntimeConfig) -> Result<(), CgroupError> { + let policy = config.config(); + let fd = open_directory(Path::new(&policy.delegated_cgroup_root))?; + let stat = rustix::fs::fstat(&fd).map_err(io::Error::from)?; + if stat.st_uid != rustix::process::geteuid().as_raw() + || stat.st_gid != 0 + || stat.st_mode & 0o022 != 0 + || rustix::fs::fstatfs(&fd).map_err(io::Error::from)?.f_type as u64 != CGROUP2_MAGIC + || read(&fd, "cgroup.type")?.trim() != "domain" + || !read(&fd, "cgroup.procs")?.trim().is_empty() + || read(&fd, "cgroup.max.depth")?.trim() != "1" + || read(&fd, "cgroup.max.descendants")?.trim() != policy.max_concurrent_runs.to_string() + { + return Err(invalid("delegation status differs").into()); + } + for name in ["cgroup.controllers", "cgroup.subtree_control"] { + let observed = read(&fd, name)?; + if !["cpu", "memory", "pids"] + .iter() + .all(|required| observed.split_whitespace().any(|value| value == *required)) + { + return Err(invalid("delegation controllers unavailable").into()); + } + } + let budget = &policy.aggregate; + verify_limits( + &fd, + &AppliedLimits { + memory_bytes: budget.memory_bytes, + swap_bytes: 0, + processes: budget.processes, + cpu_quota_us: budget.cpu_quota_us, + cpu_period_us: budget.cpu_period_us, + oom_group: true, + }, + )?; + Ok(()) + } /// Config comes from the operator. Structural validation is not proof of /// config-file ownership, service delegation or an exec permit. /// The root must be empty, owned by this helper and not group/other writable. diff --git a/judge/crates/sandbox/src/factory.rs b/judge/crates/sandbox/src/factory.rs index 9f1af22..c5bf9f2 100644 --- a/judge/crates/sandbox/src/factory.rs +++ b/judge/crates/sandbox/src/factory.rs @@ -10,6 +10,7 @@ use rustix::{ fs::{Mode, OFlags}, }; use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; use std::{ fs::File, io::{self, Write}, @@ -30,6 +31,17 @@ struct Acquire { run_id: String, #[serde(default, skip_serializing_if = "is_false")] check_stopped: bool, + #[serde( + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "health_version" + )] + health_version: Option, +} +fn health_version<'de, D: serde::Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + u8::deserialize(deserializer).map(Some) } fn is_false(v: &bool) -> bool { !v @@ -37,16 +49,29 @@ fn is_false(v: &bool) -> bool { #[derive(Serialize, Deserialize)] #[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)] enum Allocation { - Ready { run_id: String, helper_pid: u32 }, + Ready { + run_id: String, + helper_pid: u32, + }, Refused {}, Busy {}, - Stopped { cleanup_complete: bool }, + Stopped { + cleanup_complete: bool, + }, + Health { + status: crate::factory_health::FactoryHealth, + }, } 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() + && match self.health_version { + None => true, + Some(1) => self.run_id == "health-check" && !self.check_stopped, + Some(_) => false, + } } } impl WireMessage for Allocation { @@ -56,6 +81,7 @@ impl WireMessage for Allocation { *helper_pid > 1 && crate::spec::identifier(run_id, "run_id").is_ok() } Self::Refused {} | Self::Busy {} | Self::Stopped { .. } => true, + Self::Health { status } => status.valid(), } } } @@ -86,12 +112,30 @@ impl FactoryClient { channel.send(&Acquire { run_id: "stop-check".into(), check_stopped: true, + health_version: None, })?; match channel.receive::()? { Allocation::Stopped { cleanup_complete } => Ok(cleanup_complete), _ => Err(denied()), } } + /// Root-authenticated status; does not recover, allocate or execute a task. + pub fn health(&self) -> io::Result { + if rustix::process::geteuid().as_raw() < 1000 || rustix::process::getegid().as_raw() < 1000 + { + return Err(denied()); + } + let mut channel = rpc::channel(UnixStream::connect(&self.socket)?, 0, 0)?; + channel.send(&Acquire { + run_id: "health-check".into(), + check_stopped: false, + health_version: Some(1), + })?; + match channel.receive::()? { + Allocation::Health { status } => Ok(status), + _ => Err(denied()), + } + } /// The caller holds only ordinary artifact and Unix socket capabilities. pub fn execute( &self, @@ -177,6 +221,7 @@ impl FactoryClient { channel.send(&Acquire { run_id: spec.run_id.clone(), check_stopped: false, + health_version: None, })?; let pid = match channel.receive::()? { Allocation::Ready { run_id, helper_pid } if run_id == spec.run_id => helper_pid, @@ -334,6 +379,13 @@ pub fn serve_with_limit( } clean_environment()?; let config = execution::load_config(config_path)?; + let configuration_sha256 = format!( + "{:x}", + Sha256::digest(execution::read_bounded_file( + config_path, + crate::runtime::MAX_CONFIG_BYTES + )?) + ); execution::authorize_registry(&config)?; let policy = config.config(); if policy.aggregate.supervisor_processes < 4 + 8 * policy.max_concurrent_runs { @@ -417,6 +469,8 @@ pub fn serve_with_limit( let _delegated = crate::cgroup::DelegatedCgroup::open(&config).map_err(|_| denied())?; let mut active: Vec = Vec::new(); let deadline = Instant::now() + Duration::from_secs(lifetime_seconds); + let mut health_checks = 0_usize; + let mut last_health = None; let result = (|| { while Instant::now() < deadline { let mut i = 0; @@ -451,6 +505,47 @@ pub fn serve_with_limit( let Ok(request) = channel.receive::() else { continue; }; + if request.health_version == Some(1) { + if health_checks >= MAX_IDS + || last_health + .is_some_and(|last: Instant| last.elapsed() < Duration::from_secs(1)) + { + let _ = channel.send(&Allocation::Busy {}); + continue; + } + health_checks += 1; + last_health = Some(Instant::now()); + // The fixed root supervisor context and delegation must still + // be authorized. Namespace/root/seccomp/exec acknowledgements + // remain mandatory for every actual execution, not inferred. + let current = + execution::read_bounded_file(config_path, crate::runtime::MAX_CONFIG_BYTES)?; + if format!("{:x}", Sha256::digest(current)) != configuration_sha256 + || execution::authorize(&config).is_err() + || crate::cgroup::DelegatedCgroup::inspect(&config).is_err() + { + let _ = channel.send(&Allocation::Refused {}); + continue; + } + let cleanup_complete = active.is_empty() && drained(&config)?; + channel.send(&Allocation::Health { + status: crate::factory_health::FactoryHealth { + schema_version: 1, + configuration_sha256: configuration_sha256.clone(), + supervisor_authorized: true, + execution_gate: "required_per_execution".into(), + isolation_verified: false, + max_concurrent_runs: policy.max_concurrent_runs, + active_helpers: active.len() as u32, + cleanup_complete, + remaining_requests: (max_requests - used) as u32, + remaining_lifetime_seconds: deadline + .saturating_duration_since(Instant::now()) + .as_secs(), + }, + })?; + continue; + } if request.check_stopped { if !active.is_empty() { let _ = channel.send(&Allocation::Stopped { @@ -578,4 +673,29 @@ mod tests { .valid() ); } + #[test] + fn health_is_explicitly_versioned_and_old_allocation_bytes_are_unchanged() { + let allocation = Acquire { + run_id: "a-1".into(), + check_stopped: false, + health_version: None, + }; + assert_eq!( + serde_json::to_string(&allocation).unwrap(), + r#"{"run_id":"a-1"}"# + ); + for data in [ + r#"{"run_id":"health-check","health_version":2}"#, + r#"{"run_id":"work-id","health_version":1}"#, + r#"{"run_id":"health-check","health_version":1,"check_stopped":true}"#, + ] { + let request = serde_json::from_str::(data).unwrap(); + assert!(!request.valid()); + } + assert!( + serde_json::from_str::(r#"{"run_id":"health-check","health_version":1}"#) + .unwrap() + .valid() + ); + } } diff --git a/judge/crates/sandbox/src/factory_health.rs b/judge/crates/sandbox/src/factory_health.rs new file mode 100644 index 0000000..721aaf1 --- /dev/null +++ b/judge/crates/sandbox/src/factory_health.rs @@ -0,0 +1,63 @@ +//! Versioned read-only factory observation, never an isolation/execution permit. +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FactoryHealth { + pub schema_version: u8, + pub configuration_sha256: String, + pub supervisor_authorized: bool, + pub execution_gate: String, + pub isolation_verified: bool, + pub max_concurrent_runs: u32, + pub active_helpers: u32, + pub cleanup_complete: bool, + pub remaining_requests: u32, + pub remaining_lifetime_seconds: u64, +} +impl FactoryHealth { + pub(crate) fn valid(&self) -> bool { + self.schema_version == 1 + && self.configuration_sha256.len() == 64 + && self + .configuration_sha256 + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + && self.supervisor_authorized + && self.execution_gate == "required_per_execution" + && !self.isolation_verified + && (1..=32).contains(&self.max_concurrent_runs) + && self.active_helpers <= self.max_concurrent_runs + && (!self.cleanup_complete || self.active_helpers == 0) + && self.remaining_requests <= super::factory_cycles::MAX_IDS as u32 + && self.remaining_lifetime_seconds <= 86400 + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn health_never_certifies_namespace_execution_or_busy_cleanup() { + let mut health = FactoryHealth { + schema_version: 1, + configuration_sha256: "b".repeat(64), + supervisor_authorized: true, + execution_gate: "required_per_execution".into(), + isolation_verified: false, + max_concurrent_runs: 1, + active_helpers: 0, + cleanup_complete: true, + remaining_requests: 4096, + remaining_lifetime_seconds: 3600, + }; + assert!(health.valid()); + health.isolation_verified = true; + assert!(!health.valid()); + health.isolation_verified = false; + health.active_helpers = 1; + assert!(!health.valid()); + health.cleanup_complete = false; + assert!(health.valid()); + } +} diff --git a/judge/crates/sandbox/src/lib.rs b/judge/crates/sandbox/src/lib.rs index ed89af8..88a237e 100644 --- a/judge/crates/sandbox/src/lib.rs +++ b/judge/crates/sandbox/src/lib.rs @@ -42,4 +42,6 @@ pub mod factory; #[cfg(target_os = "linux")] pub mod factory_cycles; #[cfg(target_os = "linux")] +pub mod factory_health; +#[cfg(target_os = "linux")] pub mod factory_preparation; diff --git a/judge/crates/sandbox/src/main.rs b/judge/crates/sandbox/src/main.rs index 04b68a7..57aa0c6 100644 --- a/judge/crates/sandbox/src/main.rs +++ b/judge/crates/sandbox/src/main.rs @@ -16,6 +16,11 @@ struct Cli { enum Command { /// Observe the current Linux environment without changing it. Probe, + /// Ordinary authenticated, read-only factory status; never certifies isolation. + FactoryHealth { + #[arg(long)] + socket: PathBuf, + }, #[cfg(all(target_os = "linux", feature = "vm-acceptance"))] #[command(hide = true)] VmCase { case: String }, @@ -107,6 +112,24 @@ enum Command { fn main() -> ExitCode { match Cli::parse().command { + Command::FactoryHealth { socket } => { + #[cfg(target_os = "linux")] + match kf_sandbox::factory::FactoryClient::new(socket).health() { + Ok(status) => { + println!( + "{}", + serde_json::to_string(&status).expect("bounded health") + ); + ExitCode::SUCCESS + } + Err(_) => spec_error("factory_health_unavailable", None), + } + #[cfg(not(target_os = "linux"))] + { + let _ = socket; + spec_error("unsupported_platform", None) + } + } Command::Probe => match serde_json::to_string_pretty(&kf_sandbox::probe::observe()) { Ok(report) => { println!("{report}"); diff --git a/scripts/release/README.md b/scripts/release/README.md index e24609c..ebf7257 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -243,3 +243,37 @@ reads the transient unit result and confirms stop; foreign/residual units and unrecorded/partial targets are refused. These are implementation interfaces, not actual VM acceptance. See [PostgreSQL initdb](https://www.postgresql.org/docs/18/app-initdb.html) and [systemd credentials](https://systemd.io/CREDENTIALS/) for the underlying interfaces. + +`python3.13 -m scripts.release.node_health` performs the installed service +admission-precondition checks in a fixed root operator unit. It uses bounded +ordinary broker/factory/import observer units and accepts no executable, account, +port or command override. Each helper checks its effective node/service/process +budgets, identity and no-new-privileges/capability boundary. Unknown or concurrent +units are refused; transient helper stop is read back. This is an operator CLI, +not an additional public HTTP control endpoint, and has no schema/state writer. + +The projection checks selected unit pointers and effective identities/limits, +actual DB-role/column privileges, the import owner's fresh policy-bound heartbeat +through its Unix socket, and broker queue semantics/permissions/vhost limits via +ordinary-cookie read-only CLI queries. Coordinator owns the authenticated bounded +execution/cleanup counters; truncated or unconfirmed data never proves a drain. +Its `/internal/operations/` routes require platform authority and are not proxied +publicly. Counters contain no submission IDs, source, archives or credential bytes. + +Factory health is a separate version-1 authenticated observation, rate/call-count +bounded for the finite factory lifetime. It reads current supervisor/delegation +authorization, the exact runtime-config digest, live helpers, remaining requests +and lifetime, without recovery, task allocation, ledger writes or limit repair. +The old allocation request bytes are unchanged. Health explicitly reports +`isolation_verified=false` and `execution_gate=required_per_execution`: every real +execution still needs its namespace/root/identity/seccomp/exec acknowledgements. +Admission readiness requires confirmed empty execution/cleanup state and enough +factory requests/lifetime for the configured maximum job. It does not certify +TLS/browser behavior or replace the independent VM's execution acceptance. + +The optional explicit `--probe-notification` sends one small confirmed notification +for a fresh identifier checked absent by Coordinator. It reads no message body, +creates no evaluation/grade and starts no program; ordinary periodic observation +does not publish messages. Probe failure is reported separately. Underlying +interfaces: [RabbitMQ CLI](https://www.rabbitmq.com/docs/man/rabbitmqctl.8), +[PostgreSQL privilege functions](https://www.postgresql.org/docs/18/functions-info.html). diff --git a/scripts/release/broker_health.py b/scripts/release/broker_health.py new file mode 100644 index 0000000..becbcaf --- /dev/null +++ b/scripts/release/broker_health.py @@ -0,0 +1,218 @@ +"""Finite ordinary-cookie metadata queries; no queue/user mutation or message reads.""" + +import argparse +import json +import stat +import subprocess +import tempfile +from pathlib import Path + +from . import dependencies, operator_units, preflight +from .policy import DeliveryError, canonical, fields, integer + +CTL = Path("/usr/lib/rabbitmq/sbin/rabbitmqctl") + + +def queue_arguments(value) -> dict: + if isinstance(value, dict): + result = value + elif isinstance(value, list) and len(value) == 4: + result = {} + for row in value: + if not isinstance(row, list) or len(row) != 3 or row[0] in result: + raise DeliveryError("duplicate/unknown broker argument") + name, kind, argument = row + if name not in dependencies.QUEUE_ARGUMENTS or kind != ( + "longstr" if name == "x-overflow" else "long" + ): + raise DeliveryError("broker argument type differs") + result[name] = argument + else: + raise DeliveryError("broker argument shape differs") + if canonical(result) != canonical(dependencies.QUEUE_ARGUMENTS): + raise DeliveryError("actual broker queue budgets differ") + return result + + +def validate_snapshot(value: dict) -> dict: + fields(value, {"queues", "vhost_limits", "users", "permissions"}, "broker snapshot") + rows = value["queues"] + if not isinstance(rows, list) or len(rows) != 1: + raise DeliveryError("actual broker queue inventory differs") + queue = fields( + rows[0], + { + "name", + "durable", + "auto_delete", + "exclusive", + "type", + "state", + "arguments", + "messages_ready", + "messages_unacknowledged", + "message_bytes", + "consumers", + "policy", + "operator_policy", + "effective_policy_definition", + }, + "broker queue", + ) + if ( + queue["name"] != dependencies.QUEUE + or queue["durable"] is not True + or queue["auto_delete"] is not False + or queue["exclusive"] is not False + or queue["type"] != "classic" + or queue["state"] != "running" + or queue["policy"] not in (None, "") + or queue["operator_policy"] not in (None, "") + or queue["effective_policy_definition"] not in ({}, []) + ): + raise DeliveryError("actual broker queue semantics differ") + queue_arguments(queue["arguments"]) + limits = value["vhost_limits"] + if not isinstance(limits, list) or len(limits) != 1: + raise DeliveryError("actual broker vhost limits unavailable") + row = fields(limits[0], {"vhost", "limits"}, "broker vhost limits") + if row["vhost"] != "kf" or not isinstance(row["limits"], str): + raise DeliveryError("actual broker vhost binding differs") + if canonical(operator_units.decode(row["limits"].encode(), 1024)) != canonical( + {"max-connections": 4, "max-queues": 1} + ): + raise DeliveryError("actual broker vhost budgets differ") + users = value["users"] + if ( + not isinstance(users, list) + or len(users) != 2 + or canonical(sorted(users, key=lambda u: u.get("user", ""))) + != canonical( + [ + {"user": "kf_outbox", "tags": []}, + {"user": "kf_worker", "tags": []}, + ] + ) + ): + raise DeliveryError("actual broker users/tags differ") + permissions = value["permissions"] + expected = [ + { + "user": "kf_" + role, + "configure": r"^kf\.judge\.evaluations\.v1$", + "write": r"^(amq\.default|kf\.judge\.evaluations\.v1)$" if role == "outbox" else "^$", + "read": r"^kf\.judge\.evaluations\.v1$" if role == "worker" else "^$", + } + for role in ("outbox", "worker") + ] + if ( + not isinstance(permissions, list) + or len(permissions) != 2 + or canonical(sorted(permissions, key=lambda u: u.get("user", ""))) != canonical(expected) + ): + raise DeliveryError("actual broker authority differs") + return { + "schema_version": 1, + "metadata_verified": True, + "messages_ready": integer(queue["messages_ready"], 0, 64, "broker ready count"), + "messages_unacknowledged": integer( + queue["messages_unacknowledged"], 0, 1, "broker unacknowledged count" + ), + "message_bytes": integer(queue["message_bytes"], 0, 32768 + 4096, "broker bytes"), + "consumers": integer(queue["consumers"], 0, 1, "broker consumer count"), + } + + +def command(arguments: list[str]): + # A file-backed capped stream prevents malicious metadata/log output from + # allocating an unbounded capture buffer. The unit adds a hard file limit. + env = { + "PATH": "/usr/bin:/bin", + "LC_ALL": "C.UTF-8", + "HOME": dependencies.STORAGE["rabbitmq"]["path"], + "ERL_EPMD_ADDRESS": "127.0.0.1", + "RABBITMQ_NODENAME": "kf@localhost", + "RABBITMQ_CTL_ERL_ARGS": ( + "+S 1:1 +A 2 +P 1024 +Q 1024 -kernel inet_dist_use_interface {127,0,0,1}" + ), + } + with tempfile.TemporaryFile() as output: + result = subprocess.run( + [ + str(CTL), + "-q", + "-n", + "kf@localhost", + "--timeout", + "5", + *arguments, + "--formatter=json", + ], + stdout=output, + stderr=subprocess.DEVNULL, + timeout=8, + env=env, + ) + if result.returncode or output.tell() > 32768: + raise DeliveryError("bounded broker metadata query unavailable") + output.seek(0) + return operator_units.decode(output.read(32769)) + + +def observe() -> dict: + operator_units.guarded("broker-health") + preflight.protected(CTL) + cookie = Path(dependencies.STORAGE["rabbitmq"]["path"]) / ".erlang.cookie" + info = cookie.lstat() + if ( + not stat.S_ISREG(info.st_mode) + or info.st_nlink != 1 + or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (1204, 1204, 0o400) + ): + raise DeliveryError("ordinary private broker cookie required") + data = { + "queues": command( + [ + "list_queues", + "-p", + "kf", + "name", + "durable", + "auto_delete", + "exclusive", + "type", + "state", + "arguments", + "messages_ready", + "messages_unacknowledged", + "message_bytes", + "consumers", + "policy", + "operator_policy", + "effective_policy_definition", + ] + ), + "vhost_limits": command(["list_vhost_limits", "--global"]), + "users": command(["list_users"]), + "permissions": command(["list_permissions", "-p", "kf"]), + } + return validate_snapshot(data) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Bounded ordinary broker metadata") + parser.add_argument("--inside", action="store_true", required=True) + parser.parse_args() + try: + print(json.dumps(observe(), sort_keys=True)) + except (OSError, ValueError, subprocess.SubprocessError) as error: + parser.exit( + 1, + "Broker health refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/factory_health.py b/scripts/release/factory_health.py new file mode 100644 index 0000000..9de490d --- /dev/null +++ b/scripts/release/factory_health.py @@ -0,0 +1,46 @@ +"""Ordinary bounded status client; no execution request, credentials or privileges.""" + +import argparse +import os +import re +from pathlib import Path + +from . import install, operator_units, preflight +from .policy import DeliveryError + + +def main() -> None: + parser = argparse.ArgumentParser(description="Fixed ordinary factory observation") + parser.add_argument("--inside", action="store_true", required=True) + parser.parse_args() + try: + operator_units.guarded("factory-health") + payload = Path(__file__).resolve().parents[3] + if ( + payload.parent.parent != install.RELEASES + or payload.name != "payload" + or not re.fullmatch(r"sd-[0-9a-f]{12}", payload.parent.name) + ): + raise DeliveryError("factory observer is not an installed delivery") + binary = payload / "bin/kf-sandbox" + preflight.protected(binary) + os.execve( + binary, + [str(binary), "factory-health", "--socket", "/var/lib/kf-judge/factory/control.sock"], + { + "PATH": "/usr/bin:/bin", + "LC_ALL": "C.UTF-8", + "LD_LIBRARY_PATH": str(payload / "lib"), + }, + ) + except (OSError, ValueError) as error: + parser.exit( + 1, + "Factory health refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/import_health.py b/scripts/release/import_health.py new file mode 100644 index 0000000..8111d8a --- /dev/null +++ b/scripts/release/import_health.py @@ -0,0 +1,44 @@ +"""Fixed ordinary import-only read projection in its private PID/network view.""" + +import argparse +import os +from pathlib import Path + +from . import install, operator_units, service_bootstrap +from .policy import DeliveryError + + +def main() -> None: + parser = argparse.ArgumentParser(description="Private ordinary import observation") + parser.add_argument("--inside", action="store_true", required=True) + parser.parse_args() + try: + operator_units.guarded("import-health") + status = dict( + line.split(":", 1) + for line in Path("/proc/self/status").read_text().splitlines() + if ":" in line + ) + interfaces = { + line.split(":", 1)[0].strip() + for line in Path("/proc/net/dev").read_text().splitlines()[2:] + } + if status.get("NSpid", "").split()[-1:] != ["1"] or interfaces != {"lo"}: + raise DeliveryError("private import observation view required") + payload = Path(__file__).resolve().parents[3] + if payload.parent.parent != install.RELEASES or payload.name != "payload": + raise DeliveryError("installed import observer required") + command, env = service_bootstrap.python_command("problem-import", payload.parent.name) + command[-1] = "kf_oj.workers.problem_status" + os.execve(command[0], command, env) + except (OSError, ValueError, KeyError) as error: + parser.exit( + 1, + "Import health refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/node_health.py b/scripts/release/node_health.py new file mode 100644 index 0000000..b48176f --- /dev/null +++ b/scripts/release/node_health.py @@ -0,0 +1,584 @@ +"""Bounded operator readiness: fixed services, ACLs, broker and gated factory capacity.""" + +import argparse +import asyncio +import json +import logging +import os +import subprocess +import urllib.request +from pathlib import Path +from uuid import UUID + +from . import ( + configuration, + database, + dependencies, + dependency_install, + factory_service, + install, + operator_units, + preflight, + service_install, + services, +) +from .policy import DeliveryError, canonical, digest, fields, read_json + +# Table/column privilege introspection only. Business state counters come from +# the execution owner, not a second privileged database state reader/writer. +PRIVILEGES = { + "api": { + "accounts:SELECT": True, + "evaluations:SELECT": True, + "evaluations:state:UPDATE": False, + "accounts:is_administrator:UPDATE": False, + }, + "coordinator": { + "evaluations:SELECT": True, + "evaluations:state:UPDATE": True, + "accounts:SELECT": False, + "problems:active_version_id:UPDATE": False, + }, + "import": { + "problem_import_operations:SELECT": True, + "problem_import_operations:state:UPDATE": True, + "evaluations:state:UPDATE": False, + "accounts:SELECT": False, + }, + "identity": { + "accounts:SELECT": True, + "accounts:is_administrator:UPDATE": True, + "evaluations:SELECT": False, + "evaluations:state:UPDATE": False, + }, +} + + +def binding() -> tuple[dict, dict, Path]: + selected = install.selection(check_configuration=False) + payload = install.RELEASES / selected["release_id"] / "payload" + for name in ( + "node_health.py", + "broker_health.py", + "factory_health.py", + "import_health.py", + "operator_units.py", + ): + shipped = payload / "delivery/scripts/release" / name + preflight.protected(shipped) + if digest(shipped) != digest(Path(__file__).parent / name): + raise DeliveryError("health tool differs from selected delivery") + receipt = read_json( + configuration.installed_receipt_path(selected["release_id"], selected["configuration_id"]), + 16384, + ) + configuration.validate_receipt(receipt, installed=True) + paths = {} + for name in configuration.CONFIG_FILES: + path = configuration.installed_path( + selected["release_id"], selected["configuration_id"], name + ) + preflight.protected(path) + configuration.private(path) + if digest(path) != receipt["configuration_files"][str(path)]: + raise DeliveryError("live configuration changed since selection") + paths[name] = path + manifest_path = payload.parent / "manifest.json" + preflight.protected(manifest_path) + if digest(manifest_path) != selected["manifest_sha256"]: + raise DeliveryError("selected manifest changed") + data = read_json(manifest_path) + # Immutable payload bytes/images are verified by installation and every + # actual Sandbox request. A frequent status query is not a rebuild/certifier. + configuration.validate_contents( + paths, payload, data, attest=receipt["operator_attested_measurement"] + ) + return selected, paths, payload + + +def service_profile(name: str) -> tuple[int, int, int, int, int]: + if name in services.PROFILES: + user, memory, pids, cpu, _, _ = services.PROFILES[name] + return preflight.ACCOUNTS[user], memory, pids, cpu, 64 + if name in {"postgres", "rabbitmq"}: + return (1203 if name == "postgres" else 1204), 512, 64, 100, 256 + if name == "sandbox-factory": + return 0, 256, 32, 50, 64 + raise DeliveryError("unknown health service") + + +def inspect_unit(name: str, selected: dict) -> dict: + uid, memory, pids, cpu, file_mib = service_profile(name) + unit = "kf-" + name + ".service" + link = service_install.SYSTEM_UNITS / unit + if name in services.PROFILES: + directory = service_install.known_link(link) + expected = service_install.UNITS / ( + selected["release_id"] + "." + selected["configuration_id"] + ) + elif name == "sandbox-factory": + directory = factory_service.known_link(link) + expected = factory_service.UNITS / ( + selected["release_id"] + "." + selected["configuration_id"] + ) + else: + directory = dependency_install.known_link(link) + expected = dependency_install.ROOT / ( + selected["release_id"] + "." + selected["configuration_id"] + ) + if directory != expected: + raise DeliveryError("service snapshot is not selected") + value = dict( + line.split("=", 1) + for line in service_install.systemctl( + "show", + unit, + "--property=LoadState,ActiveState,SubState,MainPID,FragmentPath,ControlGroup", + ).splitlines() + if "=" in line + ) + if ( + value.get("LoadState") != "loaded" + or value.get("ActiveState") != "active" + or value.get("SubState") != "running" + or value.get("FragmentPath") not in {str(link), str(directory / unit)} + or value.get("ControlGroup") != "/kfonlinejudge.slice/" + unit + ): + raise DeliveryError("selected service is not running") + pid = int(value.get("MainPID", "0")) + if not 1 < pid < 2**31: + raise DeliveryError("running service PID differs") + info = dict( + line.split(":", 1) + for line in Path(f"/proc/{pid}/status").read_text().splitlines() + if ":" in line + ) + if ( + info.get("Uid", "").split() != [str(uid)] * 4 + or info.get("Gid", "").split() != [str(uid)] * 4 + ): + raise DeliveryError("effective service identity differs") + if not set(info.get("Groups", "").split()) <= {str(uid)}: + raise DeliveryError("unexpected service supplementary authority") + if name != "sandbox-factory" and ( + info.get("NoNewPrivs", "").strip() != "1" + or any( + not info.get(k, "").strip() or int(info[k].strip(), 16) != 0 + for k in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb") + ) + ): + raise DeliveryError("effective ordinary privilege boundary differs") + if name == "sandbox-factory" and ( + info.get("NoNewPrivs", "").strip() != "0" or info.get("Seccomp", "").strip() != "0" + ): + raise DeliveryError("factory inherited incompatible hardening") + if name in {"api", "outbox", "problem-import"} and info.get("NSpid", "").split()[-1:] != ["1"]: + raise DeliveryError("Python service private PID namespace missing") + group = Path("/sys/fs/cgroup" + value["ControlGroup"]) + actual = { + k: (group / k).read_text().strip() + for k in ("memory.max", "memory.swap.max", "pids.max", "cpu.max") + } + if actual != { + "memory.max": str(memory * 1024**2), + "memory.swap.max": "0", + "pids.max": str(pids), + "cpu.max": f"{cpu * 1000} 100000", + }: + raise DeliveryError("effective service limits differ") + if Path(f"/proc/{pid}/cgroup").read_text().strip() != "0::" + value["ControlGroup"]: + raise DeliveryError("effective service cgroup differs") + expected_file = str(file_mib * 1024**2) + # /proc format has names with spaces; fixed relevant prefixes are unambiguous. + limits = Path(f"/proc/{pid}/limits").read_text().splitlines() + expected = { + "Max core file size": "0", + "Max file size": expected_file, + "Max open files": "64" if name == "problem-import" else "256", + "Max stack size": str(16 * 1024**2), + } + for prefix, bound in expected.items(): + rows = [line[len(prefix) :].split() for line in limits if line.startswith(prefix)] + if len(rows) != 1 or rows[0][:2] != [bound, bound]: + raise DeliveryError("effective service process limits differ") + return { + "active": True, + "uid": uid, + "limits": actual, + "memory_current": int((group / "memory.current").read_text()), + "pids_current": int((group / "pids.current").read_text()), + } + + +def validate_role(role: str, observed: dict) -> None: + expected = { + "login": configuration.PROFILES[role], + "database": database.DATABASE, + "canlogin": True, + "superuser": False, + "createdb": False, + "createrole": False, + "replication": False, + "bypassrls": False, + "member": True, + "inherited": True, + "set_role": False, + "connection_limit": 8, + "privileges": PRIVILEGES[role], + } + if canonical(observed) != canonical(expected): + raise DeliveryError("database role authority differs") + + +async def roles(endpoints: dict) -> dict: + import asyncpg + + result = {} + for role, requirements in PRIVILEGES.items(): + connection = await asyncpg.connect( + endpoints["database"][role], + timeout=2, + command_timeout=2, + server_settings={ + "application_name": "kf-node-health", + "default_transaction_read_only": "on", + "statement_timeout": "2000", + "lock_timeout": "500", + "idle_in_transaction_session_timeout": "3000", + }, + ) + try: + row = await connection.fetchrow( + "SELECT session_user login,current_database() database,rolcanlogin canlogin," + "rolsuper superuser,rolcreatedb createdb,rolcreaterole createrole," + "rolreplication replication," + "rolbypassrls bypassrls,rolconnlimit connection_limit," + "pg_has_role(session_user,$1,'MEMBER') member," + "pg_has_role(session_user,$1,'USAGE') inherited," + "pg_has_role(session_user,$1,'SET') set_role " + "FROM pg_roles WHERE rolname=session_user", + database.MEMBERSHIP[configuration.PROFILES[role]], + ) + observed = dict(row) + observed["privileges"] = {} + for key in requirements: + parts = key.split(":") + observed["privileges"][key] = await connection.fetchval( + "SELECT has_table_privilege(current_user,$1,$2)" + if len(parts) == 2 + else "SELECT has_column_privilege(current_user,$1,$2,$3)", + *parts, + ) + validate_role(role, observed) + result[role] = {"authority_verified": True} + finally: + await connection.close(timeout=1) + return result + + +def http(url: str, *, token: str | None = None) -> dict: + # Fixed numeric loopback paths only; inherited proxy settings and redirects + # may not turn an operator query into a request to another service/host. + if url not in { + "http://127.0.0.1:8000/api/v1/health/live", + "http://127.0.0.1:8100/health/ready", + "http://127.0.0.1:8100/internal/operations/status", + "http://127.0.0.1:8100/internal/operations/probe-id", + }: + raise DeliveryError("unknown health HTTP target") + + class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *_): + return None + + opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()) + request = urllib.request.Request( + url, headers={"Authorization": "Bearer " + token} if token else {} + ) + with opener.open(request, timeout=5) as response: + if response.status != 200: + raise DeliveryError("health HTTP unavailable") + result = operator_units.decode(response.read(32769)) + if not isinstance(result, dict): + raise DeliveryError("health HTTP object required") + return result + + +def validate_factory(value: dict, expected: str) -> None: + fields( + value, + { + "schema_version", + "configuration_sha256", + "supervisor_authorized", + "execution_gate", + "isolation_verified", + "max_concurrent_runs", + "active_helpers", + "cleanup_complete", + "remaining_requests", + "remaining_lifetime_seconds", + }, + "factory health", + ) + if ( + value["schema_version"] != 1 + or type(value["schema_version"]) is not int + or value["configuration_sha256"] != expected + or value["supervisor_authorized"] is not True + or value["execution_gate"] != "required_per_execution" + or value["isolation_verified"] is not False + or type(value["max_concurrent_runs"]) is not int + or value["max_concurrent_runs"] != 1 + or type(value["active_helpers"]) is not int + or not 0 <= value["active_helpers"] <= 1 + or type(value["cleanup_complete"]) is not bool + or (value["cleanup_complete"] and value["active_helpers"]) + or type(value["remaining_requests"]) is not int + or not 0 <= value["remaining_requests"] <= 4096 + or type(value["remaining_lifetime_seconds"]) is not int + or not 0 <= value["remaining_lifetime_seconds"] <= 86400 + ): + raise DeliveryError("factory authorization/binding status differs") + + +def validate_execution(value: dict, admission: dict) -> dict: + fields( + value, + {"schema_version", "admission", "configured_workers", "lease_seconds", "counters"}, + "execution snapshot", + ) + if ( + type(value["schema_version"]) is not int + or value["schema_version"] != 1 + or type(value["configured_workers"]) is not int + or value["configured_workers"] != 1 + or value["lease_seconds"] != 120 + or type(value["lease_seconds"]) is not int + or canonical(value["admission"]) != canonical(admission) + ): + raise DeliveryError("live coordinator admission differs") + counters = fields( + value["counters"], + {"queued", "in_flight", "cleanup_unconfirmed", "database_clock_unix_ms"}, + "execution counters", + ) + for name in ("queued", "in_flight", "cleanup_unconfirmed"): + item = fields(counters[name], {"count_capped", "truncated"}, "bounded execution counter") + if ( + type(item["count_capped"]) is not int + or not 0 <= item["count_capped"] <= 128 + or type(item["truncated"]) is not bool + or (item["truncated"] and item["count_capped"] != 128) + ): + raise DeliveryError("execution counter bound differs") + if ( + type(counters["database_clock_unix_ms"]) is not int + or counters["database_clock_unix_ms"] <= 0 + ): + raise DeliveryError("execution counter clock unavailable") + return counters + + +def admission_ready(root: dict, counters: dict, broker: dict, admission: dict) -> bool: + reserve_seconds = ( + max( + admission["legacy"]["max_job_wall_time_us"], + admission["multi_runtime"]["max_job_wall_time_us"], + ) + + 999999 + ) // 1000000 + 60 + return bool( + root["cleanup_complete"] + and root["active_helpers"] == 0 + and root["remaining_requests"] >= 257 + and root["remaining_lifetime_seconds"] >= reserve_seconds + and counters["in_flight"] == {"count_capped": 0, "truncated": False} + and counters["cleanup_unconfirmed"] == {"count_capped": 0, "truncated": False} + and broker["consumers"] == 1 + ) + + +async def notification_probe(url: str, identifier: str) -> None: + import aio_pika + from pamqp.commands import Basic + + for name in ("aio_pika", "aiormq", "pamqp"): + logging.getLogger(name).setLevel(logging.CRITICAL) + connection = await aio_pika.connect(url, timeout=2, heartbeat=30) + try: + channel = await connection.channel(publisher_confirms=True, on_return_raises=True) + await channel.declare_queue(dependencies.QUEUE, passive=True, timeout=2) + # A synthetically absent evaluation notification cannot create a grade or + # execute a program: Coordinator must find and lease an existing row. + body = canonical( + {"schema_version": 1, "kind": "judge.evaluation", "evaluation_id": identifier} + ).strip() + reply = await channel.default_exchange.publish( + aio_pika.Message( + body, + content_type="application/json", + delivery_mode=aio_pika.DeliveryMode.PERSISTENT, + ), + routing_key=dependencies.QUEUE, + mandatory=True, + timeout=2, + ) + if not isinstance(reply, Basic.Ack): + raise DeliveryError("broker delivery not confirmed") + finally: + await asyncio.wait_for(connection.close(), timeout=1) + + +def observe(*, probe: bool) -> dict: + operator_units.guarded("node-health") + selected, paths, payload = binding() + result = { + "schema_version": 1, + "source_sha": read_json(payload.parent / "manifest.json")["source_sha"], + "configuration_id": selected["configuration_id"], + "ready_for_admission": False, + "isolation_verified": False, + "components": {}, + "errors": [], + } + components, errors = result["components"], result["errors"] + for name in [*services.PROFILES, "postgres", "rabbitmq", "sandbox-factory"]: + try: + components[name] = inspect_unit(name, selected) + except (OSError, ValueError): + errors.append(name + "_unavailable") + endpoints = dependency_install.installed_endpoints(selected) + try: + components["db_authority"] = asyncio.run(roles(endpoints)) + except Exception: + errors.append("db_authority_unavailable") + try: + live = http("http://127.0.0.1:8000/api/v1/health/live") + if live.get("status") != "ok" or live.get("service") != "platform": + raise DeliveryError("API live shape differs") + components["api_live"] = True + except (OSError, ValueError): + errors.append("api_live_unavailable") + try: + coordinator = read_json(paths["coordinator.json"], 32768) + counters = validate_execution( + http( + "http://127.0.0.1:8100/internal/operations/status", + token=coordinator["platform_token"], + ), + coordinator["admission"], + ) + components["execution"] = counters + except (OSError, ValueError): + errors.append("execution_owner_unavailable") + try: + components["broker"] = operator_units.run("broker-health", payload) + except (OSError, ValueError, subprocess.SubprocessError): + errors.append("broker_metadata_unavailable") + try: + state = operator_units.run("import-health", payload) + fields( + state, + { + "schema_version", + "worker_ready", + "reserved_bytes", + "uploaded_capped", + "validating_capped", + "truncated", + }, + "import status", + ) + if ( + type(state["schema_version"]) is not int + or state["schema_version"] != 1 + or state["worker_ready"] is not True + or state["truncated"] is not False + ): + raise DeliveryError("import heartbeat or counter unavailable") + for key, maximum in ( + ("reserved_bytes", 268435456), + ("uploaded_capped", 128), + ("validating_capped", 1), + ): + if type(state[key]) is not int or not 0 <= state[key] <= maximum: + raise DeliveryError("import status bound differs") + components["import_pipeline"] = state + except (OSError, ValueError, subprocess.SubprocessError): + errors.append("import_pipeline_unavailable") + try: + status = operator_units.run("factory-health", payload) + validate_factory(status, digest(paths["runtime.json"])) + components["factory_status"] = status + except (OSError, ValueError, subprocess.SubprocessError): + errors.append("factory_status_unavailable") + if probe: + try: + value = fields( + http( + "http://127.0.0.1:8100/internal/operations/probe-id", + token=coordinator["platform_token"], + ), + {"schema_version", "evaluation_id", "absent"}, + "notification probe identity", + ) + identifier = UUID(value["evaluation_id"]) + if ( + type(value["schema_version"]) is not int + or value["schema_version"] != 1 + or value["absent"] is not True + or identifier.int == 0 + or str(identifier) != value["evaluation_id"] + ): + raise DeliveryError("notification probe authority unavailable") + asyncio.run( + asyncio.wait_for( + notification_probe(endpoints["amqp"]["outbox"], str(identifier)), timeout=6 + ) + ) + components["notification_delivery_confirmed"] = True + except Exception: + errors.append("notification_delivery_unconfirmed") + if install.selection(check_configuration=False) != selected: + errors.append("selected_binding_changed") + if not errors: + root = components["factory_status"] + counters = components["execution"] + result["ready_for_admission"] = admission_ready( + root, counters, components["broker"], coordinator["admission"] + ) + return result + + +def main() -> None: + parser = argparse.ArgumentParser(description="Bounded installed service readiness") + parser.add_argument("--inside", action="store_true") + parser.add_argument("--probe-notification", action="store_true") + args = parser.parse_args() + try: + if args.inside: + result = observe(probe=args.probe_notification) + else: + if os.getuid() != 0 or os.geteuid() != 0: + raise DeliveryError("root operator required") + selected = install.selection(check_configuration=False) + result = operator_units.run( + "node-health", + install.RELEASES / selected["release_id"] / "payload", + notification_probe=args.probe_notification, + ) + print(json.dumps(result, sort_keys=True)) + if not args.inside and not result["ready_for_admission"]: + raise SystemExit(2) + except (OSError, ValueError, subprocess.SubprocessError) as error: + parser.exit( + 1, + "Node health refused: " + + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) + + "\n", + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/release/operator_units.py b/scripts/release/operator_units.py new file mode 100644 index 0000000..ebe1ddd --- /dev/null +++ b/scripts/release/operator_units.py @@ -0,0 +1,231 @@ +"""Finite fixed-purpose operator helpers; no caller-controlled executable or identity.""" + +import json +import os +import resource +import subprocess +from pathlib import Path + +from . import configuration, install, preflight, service_install +from .policy import DeliveryError, invalid_constant, unique_object +from .services import python_environment + +PROFILES = { + "node-health": ("root", 0, 128, 16, 50, 90), + "broker-health": ("rabbitmq", 1204, 256, 64, 50, 45), + "factory-health": ("kf-worker", 1202, 32, 4, 10, 10), + "import-health": ("kf-oj", 1200, 128, 16, 25, 10), +} + + +def guarded(name: str) -> None: + _, uid, memory, pids, cpu, _ = PROFILES[name] + if ( + os.getuid() != uid + or os.geteuid() != uid + or os.getgid() != uid + or not set(os.getgroups()) <= {uid} + ): + raise DeliveryError("fixed operator helper identity required") + observed = dict( + line.split(":", 1) + for line in Path("/proc/self/status").read_text().splitlines() + if ":" in line + ) + if observed.get("NoNewPrivs", "").strip() != "1" or any( + not observed.get(k, "").strip() or int(observed[k].strip(), 16) != 0 + for k in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb") + ): + raise DeliveryError("operator helper privilege boundary differs") + group = preflight.current_cgroup() + if str(group) != "/sys/fs/cgroup/kfonlinejudge.slice/kf-" + name + ".service": + raise DeliveryError("operator helper cgroup differs") + for directory, expected in ( + (group, [str(memory * 1024**2), "0", str(pids), f"{cpu * 1000} 100000"]), + ( + Path("/sys/fs/cgroup/kfonlinejudge.slice"), + [str(3840 * 1024**2), "0", "384", "200000 100000"], + ), + ): + actual = [ + (directory / k).read_text().strip() + for k in ("memory.max", "memory.swap.max", "pids.max", "cpu.max") + ] + if actual != expected: + raise DeliveryError("operator helper/node budgets differ") + for key, maximum in ( + (resource.RLIMIT_CORE, 0), + (resource.RLIMIT_FSIZE, 16 * 1024**2), + (resource.RLIMIT_NOFILE, 256), + (resource.RLIMIT_STACK, 16 * 1024**2), + ): + if resource.getrlimit(key) != (maximum, maximum): + raise DeliveryError("operator helper process budgets differ") + + +def decode(contents: bytes, maximum: int = 32768): + if len(contents) > maximum: + raise DeliveryError("operator output exceeds bound") + return json.loads( + contents, + object_pairs_hook=unique_object, + parse_constant=invalid_constant, + ) + + +def run(name: str, payload: Path, *, notification_probe: bool = False) -> dict: + if os.getuid() != 0 or os.geteuid() != 0 or name not in PROFILES: + raise DeliveryError("fixed root operator helper required") + user, _, memory, pids, cpu, seconds = PROFILES[name] + unit = "kf-" + name + ".service" + current = dict( + line.split("=", 1) + for line in service_install.systemctl( + "show", unit, "--property=LoadState,ActiveState,MainPID" + ).splitlines() + if "=" in line + ) + if ( + current.get("LoadState") != "not-found" + or current.get("ActiveState") != "inactive" + or current.get("MainPID") != "0" + ): + raise DeliveryError("operator helper already active or unknown") + release_id = payload.parent.name + selected = install.selection(check_configuration=False) + if ( + payload != install.RELEASES / selected["release_id"] / "payload" + or release_id != selected["release_id"] + ): + raise DeliveryError("operator helper is not the selected payload") + description = "KF fixed " + name + " " + release_id + "." + selected["configuration_id"] + command = [ + "/usr/bin/systemd-run", + "--quiet", + "--wait", + "--pipe", + "--collect", + "--unit=" + unit, + "--description=" + description, + "--slice=kfonlinejudge.slice", + ] + properties = { + "Type": "exec", + "User": user, + "Group": user, + "UMask": "0077", + "MemoryMax": str(memory) + "M", + "MemorySwapMax": "0", + "TasksMax": str(pids), + "CPUQuota": str(cpu) + "%", + "LimitCORE": "0", + "LimitFSIZE": "16M", + "LimitNOFILE": "256", + "LimitSTACK": "16M", + "RuntimeMaxSec": str(seconds), + "TimeoutStartSec": "10", + "TimeoutStopSec": "10", + "Restart": "no", + "KillMode": "control-group", + "NoNewPrivileges": "yes", + "CapabilityBoundingSet": "", + "AmbientCapabilities": "", + "ProtectSystem": "strict", + "ProtectHome": "yes", + "PrivateTmp": "yes", + "PrivateDevices": "yes", + "ProtectControlGroups": "yes", + "RestrictAddressFamilies": "AF_UNIX AF_INET", + "WorkingDirectory": str(payload), + } + for key, value in properties.items(): + command.append("--property=" + key + "=" + value) + if name == "import-health": + path = configuration.installed_path( + selected["release_id"], selected["configuration_id"], "import.env" + ) + command += [ + "--property=EnvironmentFile=" + str(path), + "--property=PrivatePIDs=yes", + "--property=PrivateNetwork=yes", + "--property=RestrictAddressFamilies=AF_UNIX", + ] + if name == "import-health": + if notification_probe: + raise DeliveryError("unexpected import probe override") + command += [ + "--property=Environment=PATH=/usr/bin:/bin LC_ALL=C.UTF-8 " + + python_environment(payload) + ] + command += [ + str(payload / "python/bin/python3.13"), + "-s", + "-P", + "-m", + "scripts.release.import_health", + "--inside", + ] + else: + command += ["/usr/bin/env", "-i", "PATH=/usr/bin:/bin", "LC_ALL=C.UTF-8"] + if name == "factory-health": + if notification_probe: + raise DeliveryError("unexpected factory probe override") + command += [ + *python_environment(payload).split(), + str(payload / "python/bin/python3.13"), + "-s", + "-P", + "-m", + "scripts.release.factory_health", + "--inside", + ] + elif name != "import-health": + command += [ + *python_environment(payload).split(), + str(payload / "python/bin/python3.13"), + "-s", + "-P", + "-m", + "scripts.release." + name.replace("-", "_"), + "--inside", + ] + if notification_probe and name == "node-health": + command.append("--probe-notification") + elif notification_probe: + raise DeliveryError("unexpected broker probe override") + try: + # Pipe output is bounded by the fixed helper and the service file/memory + # ceilings; helper tools below cap each pipe before serializing it. + result = subprocess.run( + command, + capture_output=True, + timeout=seconds + 15, + env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"}, + ) + if result.returncode or len(result.stderr) > 4096: + raise DeliveryError("operator helper unavailable") + value = decode(result.stdout) + if not isinstance(value, dict): + raise DeliveryError("operator helper object required") + finally: + actual = dict( + line.split("=", 1) + for line in service_install.systemctl( + "show", unit, "--property=LoadState,ActiveState,MainPID,Transient,Description" + ).splitlines() + if "=" in line + ) + if actual.get("LoadState") != "not-found": + if actual.get("Transient") != "yes" or actual.get("Description") != description: + raise DeliveryError("foreign operator helper; refused cleanup") + service_install.systemctl("stop", unit) + actual = dict( + line.split("=", 1) + for line in service_install.systemctl( + "show", unit, "--property=ActiveState,MainPID" + ).splitlines() + if "=" in line + ) + if actual.get("ActiveState") != "inactive" or actual.get("MainPID") != "0": + raise DeliveryError("operator helper cleanup unconfirmed") + return value