From ede3c303bf6300c7c42f28a3ba19d5144c47c16a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 3 Sep 2026 05:53:42 +0200 Subject: [PATCH] Adopt Idunn runtime presence authority --- .gitmodules | 6 + Cargo.lock | 62 +- Cargo.toml | 14 +- README.md | 72 +- deployment/codex-linux-x64.manifest | 4 - deployment/idunn/recipe.toml | 124 ++ src/daemon.rs | 498 +++-- src/idunn_health.rs | 2914 ++++++++++----------------- src/main.rs | 13 +- src/provider_backend.rs | 8 + vendor/cultcache-rs | 1 + vendor/cultnet-rs | 1 + 12 files changed, 1667 insertions(+), 2050 deletions(-) create mode 100644 .gitmodules delete mode 100644 deployment/codex-linux-x64.manifest create mode 100644 deployment/idunn/recipe.toml create mode 160000 vendor/cultcache-rs create mode 160000 vendor/cultnet-rs diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..a831628 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,6 @@ +[submodule "vendor/cultnet-rs"] + path = vendor/cultnet-rs + url = https://github.com/GameCult/cultnet-rs.git +[submodule "vendor/cultcache-rs"] + path = vendor/cultcache-rs + url = https://github.com/GameCult/cultcache-rs.git diff --git a/Cargo.lock b/Cargo.lock index ca26186..0adc2e0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -145,18 +145,17 @@ dependencies = [ "anyhow", "base64 0.23.1", "chrono", - "cultcache-rs 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=4b61f4df32d7ee1ba718596b03504d444489e7b4)", + "cultcache-rs", "cultnet-rs", - "ed25519-dalek", + "fs2", + "libc", "rand_core 0.6.4", "rmp-serde", "serde", - "serde_bytes", "serde_json", "sha2", "thiserror", "ureq", - "uuid", "zeroize", ] @@ -204,12 +203,12 @@ dependencies = [ [[package]] name = "cultcache-rs" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f" dependencies = [ "anyhow", "chrono", - "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f)", + "cultcache-rs-derive", "fs2", + "redb", "rmp-serde", "serde", "serde_bytes", @@ -218,36 +217,9 @@ dependencies = [ "windows-sys 0.59.0", ] -[[package]] -name = "cultcache-rs" -version = "0.1.0" -source = "git+https://github.com/GameCult/cultcache-rs.git?rev=4b61f4df32d7ee1ba718596b03504d444489e7b4#4b61f4df32d7ee1ba718596b03504d444489e7b4" -dependencies = [ - "anyhow", - "chrono", - "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=4b61f4df32d7ee1ba718596b03504d444489e7b4)", - "fs2", - "rmp-serde", - "serde", - "serde_bytes", - "serde_json", - "uuid", -] - [[package]] name = "cultcache-rs-derive" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - -[[package]] -name = "cultcache-rs-derive" -version = "0.1.0" -source = "git+https://github.com/GameCult/cultcache-rs.git?rev=4b61f4df32d7ee1ba718596b03504d444489e7b4#4b61f4df32d7ee1ba718596b03504d444489e7b4" dependencies = [ "proc-macro2", "quote", @@ -257,13 +229,13 @@ dependencies = [ [[package]] name = "cultnet-rs" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f" dependencies = [ "aes-gcm", "anyhow", "base64 0.22.1", "chrono", - "cultcache-rs 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f)", + "cultcache-rs", + "ed25519-dalek", "hmac", "rand", "rmp-serde", @@ -274,6 +246,8 @@ dependencies = [ "sha2", "socket2", "uuid", + "windows-sys 0.61.2", + "zeroize", ] [[package]] @@ -673,6 +647,15 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "redb" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de6c3b63e007e90ce536ec2ae4690826136a20ec8dbbbb400daef1bb999d2e36" +dependencies = [ + "libc", +] + [[package]] name = "ring" version = "0.17.14" @@ -1173,6 +1156,15 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-targets" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index cfedf20..05dcb96 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,10 +23,9 @@ daemon = [ "dep:chrono", "dep:cultcache-rs", "dep:cultnet-rs", - "dep:ed25519-dalek", - "dep:serde_bytes", + "dep:fs2", + "dep:libc", "dep:ureq", - "dep:uuid", ] [dependencies] @@ -34,16 +33,15 @@ aes-gcm = "0.10" anyhow = { version = "1", optional = true } base64 = "0.23" chrono = { version = "0.4", features = ["serde"], optional = true } -cultcache-rs = { git = "https://github.com/GameCult/cultcache-rs.git", rev = "4b61f4df32d7ee1ba718596b03504d444489e7b4", optional = true } -cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "75c180782aeba7cfd22d6412877397708a4ed28f", optional = true } -ed25519-dalek = { version = "2.2", optional = true } +cultcache-rs = { path = "vendor/cultcache-rs", optional = true } +cultnet-rs = { path = "vendor/cultnet-rs", optional = true } +fs2 = { version = "0.4", optional = true } +libc = { version = "0.2", optional = true } rand_core = { version = "0.6", features = ["getrandom"] } rmp-serde = "1" serde = { version = "1", features = ["derive"] } -serde_bytes = { version = "0.11", optional = true } serde_json = "1" sha2 = "0.10" thiserror = "2" ureq = { version = "3.4", default-features = false, features = ["rustls"], optional = true } -uuid = { version = "1", features = ["v4"], optional = true } zeroize = "1" diff --git a/README.md b/README.md index c7e5478..946c69f 100644 --- a/README.md +++ b/README.md @@ -54,33 +54,53 @@ cargo check --bin codex-connector --features daemon ``` The same production binary can initialize a typed single-caller `.cc` -configuration for first deployment, then serve only from that admitted state. -On Linux, serving requires Idunn. The daemon publishes one private-free signed -`warming / traffic-admission-pending` statement, then waits for Idunn's exact -root-owned typed traffic grant before it launches the provider child, opens -replay state, or binds a socket. -The grant also binds the root-observed systemd process ID and Linux process -start time, so another process with the same service identity cannot consume -it. The connector revalidates the grant before binding, after a complete -transport frame has decoded and immediately before service admission, and while -idle. That successful request check transfers one command to Connector-owned -execution; later revocation stops later commands but does not rewrite the -admitted result. The connector has read-only access to the grant and its -shared-snapshot lock and cannot create, repair, or replace either file. +configuration for first deployment, then serve from that state. Outside an +Idunn-managed launch the configuration's loopback bind remains available for +local operation. In a managed launch `GAMECULT_IDUNN_CANDIDATE_BIND` is the +effective private bind. Idunn admits the candidate and controls route promotion; +the Connector only binds and serves its private endpoint. -The health signing identity is created by Idunn's generic provisioning -authority. CodexConnector can use that identity to sign health but cannot enroll -it or export its public key. Signed health is derived from the active release: -the root-owned runtime activation request, the immutable adjacent `DEPLOYMENT` -witness, the compiled source commit, and the observed hashes of both the running -connector and configured Codex binary must agree exactly. Initial health setup -and the probation publication are startup requirements. Later periodic health -publication is observational and does not acquire traffic authority. The source -revision owns the exact Codex Linux package and binary hashes in -`deployment/codex-linux-x64.manifest`; that manifest is compiled into the daemon -and must agree with the immutable release witness. Activation timestamps live -in Idunn's mutable deployment receipt. Changing the provider toolchain -necessarily changes the connector release selected by Idunn. +The repo-visible Idunn declaration is +`deployment/idunn/recipe.toml`. A normal deployment is `idunn up +codex-connector`. The declaration pins the Codex package and extracted binary, +builds the one daemon, declares its process-bound replay and Codex credential +state, and requires a compatible Odin rendezvous capability before start. +`CODEX_CONNECTOR_CULTNET_RUDP` is only the current transport address for that +rendezvous. The selected managed dependency must carry an explicit `rudp://`, +`udp://`, or deliberately bare socket endpoint, and both addresses must agree. + +Idunn supplies a root-owned `GAMECULT_IDUNN_RUNTIME_BUNDLE` containing the exact +typed Expected and activation records. Systemd opens the root-owned activation +and stable-identity sources and passes exactly two named descriptors, +`gamecult-idunn-runtime-activation-key` and +`gamecult-runtime-presence-identity`, to the Connector parent. Connector marks +both close-on-exec, consumes and closes them before spawning the Codex +app-server, and strips systemd's descriptor environment from that child. +Connector publishes `gamecult.runtime_presence_health.v2` through CultNet with +both the stable provider proof and the activation-scoped proof. Target, plan, +incarnation, release, runtime, endpoint, health contract, state lineage, and +capability claims are bound to Expected rather than reconstructed from service +configuration. The binary pins the canonical two-slot state-contract digest, so +an Expected projection for a different recipe cannot be attested under the same +generation label. + +The managed command receives Idunn's admitted state root as an explicit launch +binding. Connector derives `codex-home` and `replay.cc` from that root and +rejects configuration that names a different layout. The managed configuration +and its CultCache lock must themselves be root-custodied read-only files; they +may describe caller admission and fixed runtime bounds, but cannot mutate those +inputs behind an admitted incarnation. + +The candidate socket is bound before the initial `warming` presence so the +reported endpoint is an observed service fact. Connector then waits for the +root-owned typed process-write lease that names that exact warming presence +before opening replay state or starting the official child that may refresh +`codex-home/auth.json`. It holds the lease's pre-created sibling lock shared for +the lifetime of both writers, so Idunn stops the process before moving write +authority to another incarnation. Periodic `active` or `degraded` presence is +factual provider observation and carries the exact current lease digest. Odin +derives Present and Ready by correlating these statements; Idunn decides route +admission and lifecycle actuation. ## Contract invariant diff --git a/deployment/codex-linux-x64.manifest b/deployment/codex-linux-x64.manifest deleted file mode 100644 index 53c3a0e..0000000 --- a/deployment/codex-linux-x64.manifest +++ /dev/null @@ -1,4 +0,0 @@ -schema_version=gamecult.codex_connector.codex_toolchain.v1 -package_url=https://registry.npmjs.org/@openai/codex/-/codex-0.150.0-alpha.7-linux-x64.tgz -package_sha256=8aca3112aef60127d4ffaf4c6f08e524fa7315279a1d08e7aa9c5e7f1a6943e0 -binary_sha256=aca268dc02dcef8b1ea9f528a10173d5071ae76d659a03cce6458bbebe228bee diff --git a/deployment/idunn/recipe.toml b/deployment/idunn/recipe.toml new file mode 100644 index 0000000..06bcce0 --- /dev/null +++ b/deployment/idunn/recipe.toml @@ -0,0 +1,124 @@ +schema = "gamecult.idunn.target_declaration.v1" +target = "codex-connector" +source_stamp_environment = "CODEX_CONNECTOR_BUILD_COMMIT" +required_gitlinks = ["vendor/cultcache-rs", "vendor/cultnet-rs"] + +[[steps]] +id = "extract-codex" +phase = "prepare" +runner = "rust-build" +argv = [ + "tar", + "--extract", + "--gzip", + "--file", + ".idunn-inputs/codex-linux-x64.tgz", + "--strip-components=4", + "package/vendor/x86_64-unknown-linux-musl/bin/codex", +] + +[[steps]] +id = "test-contract" +phase = "test" +runner = "rust-build" +argv = ["cargo", "test", "--locked"] + +[[steps]] +id = "test-daemon" +phase = "test" +runner = "rust-build" +argv = ["cargo", "test", "--locked", "--features", "daemon"] + +[[steps]] +id = "build-daemon" +phase = "build" +runner = "rust-build" +argv = [ + "cargo", + "build", + "--locked", + "--release", + "--features", + "daemon", + "--bin", + "codex-connector", +] + +[[external_inputs]] +id = "codex-linux-x64" +url = "https://registry.npmjs.org/@openai/codex/-/codex-0.150.0-alpha.7-linux-x64.tgz" +sha256 = "8aca3112aef60127d4ffaf4c6f08e524fa7315279a1d08e7aa9c5e7f1a6943e0" +runner = "rust-build" +destination = ".idunn-inputs/codex-linux-x64.tgz" + +[[artifacts]] +id = "codex-connector" +source_kind = "runner-output" +runner = "rust-build" +source = "target/release/codex-connector" +destination = "codex-connector" +executable = true + +[[artifacts]] +id = "codex" +source_kind = "runner-output" +runner = "rust-build" +source = "codex" +destination = "codex" +expected_sha256 = "aca268dc02dcef8b1ea9f528a10173d5071ae76d659a03cce6458bbebe228bee" +executable = true + +[service] +executable_artifact = "codex-connector" +required_adjacent_artifacts = ["codex"] +arguments = [ + { kind = "literal", value = "--config" }, + { kind = "binding", name = "config_path" }, + { kind = "literal", value = "--state-root" }, + { kind = "binding", name = "state_root" }, +] +transport = "tcp" +route_required = true +required_environment = [ + "CODEX_CONNECTOR_CULTNET_RUDP", + "GAMECULT_IDUNN_CANDIDATE_BIND", + "GAMECULT_IDUNN_PROCESS_WRITE_LEASE", + "GAMECULT_IDUNN_RUNTIME_BUNDLE", +] +optional_environment = ["RUST_LOG"] + +[service.health] +contract = "codex-connector.runtime-health.v1" + +[state] +schema_generation = "connector-v1" + +[[state.slots]] +id = "replay" +relative_path = "replay.cc" +kind = "cultcache-file" +schema = "gamecult.codex.replay_record.v1" +writer = "process-bound-single-writer" +recovery = "preserve" +startup = "create-or-open-after-write-lease" + +[[state.slots]] +id = "codex-auth" +relative_path = "codex-home/auth.json" +kind = "external-file" +schema = "openai.codex.auth-json.v1" +writer = "process-bound-single-writer" +recovery = "preserve" +startup = "create-or-open-after-write-lease" + +[[provides]] +capability = "gamecult.codex.subscription-inference" +schema = "gamecult.codex.transport_envelope.v2" +compatibility = "v2" + +[[dependencies]] +kind = "bootstrap" +capability = "odin.verse-rendezvous" +schema = "odin.verse-topology.v1" +compatibility = "v1" +startup = "before-start" diff --git a/src/daemon.rs b/src/daemon.rs index 4d7fc05..461c00f 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -8,11 +8,10 @@ use std::thread; use std::time::{Duration, SystemTime, UNIX_EPOCH}; #[cfg(target_os = "linux")] -use std::time::Instant; +use std::os::unix::fs::MetadataExt; use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; use serde::{Deserialize, Serialize}; -use sha2::Digest; use thiserror::Error; use zeroize::Zeroizing; @@ -166,12 +165,15 @@ impl CodexDaemonConfig { } fn required_connection_capacity(callers: &[CodexCallerConfig]) -> Result { + Ok(caller_request_capacity(callers)?.max(8)) +} + +fn caller_request_capacity(callers: &[CodexCallerConfig]) -> Result { callers .iter() .try_fold(0_usize, |total, caller| { total.checked_add(caller.max_concurrent_requests) }) - .map(|total| total.max(8)) .ok_or(CodexDaemonError::InvalidConfig("connection capacity")) } @@ -191,11 +193,9 @@ pub fn load_daemon_config(path: &Path) -> Result Result<(CodexDaemonConfig, Vec), CodexDaemonError> { +fn load_daemon_config_read_only(path: &Path) -> Result { let store = SingleFileMessagePackBackingStore::new(path); - let (config, snapshot) = store + let config = store .with_read_only_shared_snapshot(|entries| { let [envelope] = entries.as_slice() else { anyhow::bail!("connector configuration must contain exactly one record"); @@ -210,12 +210,11 @@ fn load_daemon_config_snapshot( if rmp_serde::to_vec(&config)? != envelope.payload { anyhow::bail!("connector configuration is not canonical positional MessagePack"); } - let snapshot = fs::read(path)?; - Ok((config, snapshot)) + Ok(config) }) .map_err(config_store_error)?; config.validate()?; - Ok((config, snapshot)) + Ok(config) } pub fn write_daemon_config( @@ -235,73 +234,106 @@ pub fn write_daemon_config( Ok(()) } -pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> { - let (config, config_bytes) = load_daemon_config_snapshot(config_path)?; - let bind = config.validate()?; - let (admissions, caller_key_bindings) = load_admissions(&config)?; +pub fn serve( + config_path: &Path, + managed_state_root: Option<&Path>, +) -> Result<(), CodexDaemonError> { #[cfg(target_os = "linux")] - let (traffic_admission, prepared_health_publisher) = - start_idunn_health_publisher(config_path, &config_bytes, &config, &caller_key_bindings)?; + if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_some() { + crate::idunn_health::require_managed_config_store(config_path) + .map_err(CodexDaemonError::Health)?; + } + let config = load_daemon_config_read_only(config_path)?; + let configured_bind = config.validate()?; + #[cfg(target_os = "linux")] + let bind = effective_bind_from_environment(configured_bind)?; + #[cfg(not(target_os = "linux"))] + let bind = configured_bind; + #[cfg(target_os = "linux")] + let (mut codex_executable, mut codex_home, mut replay_store) = + effective_runtime_paths_from_environment(&config, managed_state_root)?; + #[cfg(not(target_os = "linux"))] + let (codex_executable, codex_home, replay_store) = match managed_state_root { + None => ( + config.codex_executable.clone(), + config.codex_home.clone(), + config.replay_store.clone(), + ), + Some(_) => { + return Err(CodexDaemonError::InvalidConfig( + "Idunn state root on an unmanaged platform", + )); + } + }; + let admissions = load_admissions(&config)?; - // Probation performs only immutable activation verification and its one - // signed warming publication. Provider process, replay, and socket - // actuation begin only after Idunn grants that exact startup statement. + #[cfg(target_os = "linux")] + // Consume the systemd-owned signer descriptors before opening another + // process descriptor. Their numeric 3/4 contract is valid only at the + // managed process entry boundary. + let mut prepared_presence_publisher = prepare_runtime_presence_publisher(bind, &config)?; + + // Binding the candidate socket is service physiology. Idunn owns whether + // that candidate receives traffic, so Connector carries no second route + // or traffic-admission gate. + let listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?; + let bound = listener.local_addr().map_err(CodexDaemonError::Listen)?; + #[cfg(target_os = "linux")] + if let Some(publisher) = prepared_presence_publisher.as_mut() { + if bound != bind { + return Err(CodexDaemonError::InvalidConfig("Idunn candidate bind")); + } + let warming_sha256 = publisher + .publish("warming", "awaiting-process-write-lease") + .map_err(CodexDaemonError::Health)?; + publisher + .acquire_process_write_lease(&warming_sha256, Duration::from_secs(120)) + .map_err(CodexDaemonError::Health)?; + // The incumbent is fenced only when Idunn can replace the lease under + // our shared lock. Re-observe every write-capable path after that + // handoff so a pre-lease filesystem shape cannot survive into Active. + (codex_executable, codex_home, replay_store) = + effective_runtime_paths_from_environment(&config, managed_state_root)?; + } + #[cfg(target_os = "linux")] + // Keep the shared lease lock in the main service body as well as the + // publisher thread. A health-thread failure must not release write + // authority while replay or the official credential writer remains live. + let _process_write_lease_guard = prepared_presence_publisher + .as_ref() + .map(|publisher| publisher.process_write_lease_guard()) + .transpose() + .map_err(CodexDaemonError::Health)?; + + // Opening replay state is the first persistent write-capable actuation. + // Managed launches reach this point only after the exact warming process + // has received Idunn's typed process-write lease. let backend = Arc::new(CodexProviderBackend::start(CodexAppServerConfig { - executable: config.codex_executable.clone(), + executable: codex_executable, executable_sha256: config.codex_executable_sha256, - codex_home: config.codex_home.clone(), + codex_home, max_result_bytes: config.max_frame_bytes - FRAME_OVERHEAD_BUDGET, })?); let readiness = backend.readiness()?; let service = Arc::new(Mutex::new(CodexTransportService::open( - &config.replay_store, + &replay_store, admissions, config.max_expiry_skew_ms, )?)); #[cfg(target_os = "linux")] - traffic_admission - .require_current() - .map_err(CodexDaemonError::Health)?; - let listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?; - #[cfg(target_os = "linux")] - listener - .set_nonblocking(true) - .map_err(CodexDaemonError::Listen)?; - #[cfg(target_os = "linux")] - start_periodic_idunn_health_publisher(prepared_health_publisher, backend.clone())?; + if let Some(publisher) = prepared_presence_publisher { + start_periodic_runtime_presence_publisher(publisher, backend.clone())?; + } let active_connections = Arc::new(AtomicUsize::new(0)); - #[cfg(target_os = "linux")] - let mut next_traffic_admission_check = Instant::now(); eprintln!( "codex-connector ready at {} for {} callers using {:?}", - listener.local_addr().map_err(CodexDaemonError::Listen)?, + bound, config.callers.len(), readiness.auth_mode ); loop { - #[cfg(target_os = "linux")] - if Instant::now() >= next_traffic_admission_check { - traffic_admission - .require_current() - .map_err(CodexDaemonError::Health)?; - next_traffic_admission_check = Instant::now() + Duration::from_millis(100); - } - let stream = match listener.accept() { - Ok((stream, _)) => stream, - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { - thread::sleep(Duration::from_millis(100)); - continue; - } - Err(error) => return Err(CodexDaemonError::Listen(error)), - }; - #[cfg(target_os = "linux")] - // The polling check terminates an idle process after revocation. This - // second check is the traffic commit point: an accepted socket receives - // no request worker under a stale grant. - traffic_admission - .require_current() - .map_err(CodexDaemonError::Health)?; + let (stream, _) = listener.accept().map_err(CodexDaemonError::Listen)?; if !stream .peer_addr() .map_err(CodexDaemonError::Connection)? @@ -317,93 +349,43 @@ pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> { }; let service = service.clone(); let backend = backend.clone(); - #[cfg(target_os = "linux")] - let request_traffic_admission = traffic_admission.clone(); let max_frame_bytes = config.max_frame_bytes; let timeout = Duration::from_millis(config.socket_timeout_ms); thread::Builder::new() .name("codex-connector-request".to_string()) .spawn(move || { let _permit = permit; - let _ = serve_connection( - stream, - &service, - &backend, - #[cfg(target_os = "linux")] - &request_traffic_admission, - max_frame_bytes, - timeout, - ); + let _ = serve_connection(stream, &service, &backend, max_frame_bytes, timeout); }) .map_err(CodexDaemonError::Thread)?; } } #[cfg(target_os = "linux")] -fn start_idunn_health_publisher( - config_path: &Path, - config_bytes: &[u8], +fn prepare_runtime_presence_publisher( + candidate: SocketAddr, config: &CodexDaemonConfig, - caller_key_bindings: &[LoadedCallerKeyBinding], -) -> Result< - ( - crate::idunn_health::ConnectorTrafficAdmissionGate, - crate::idunn_health::ProviderHealthPublisher, - ), - CodexDaemonError, -> { - let endpoint = required_environment("CODEX_CONNECTOR_IDUNN_RUDP")? - .parse::() - .map_err(|_| CodexDaemonError::InvalidConfig("Idunn health endpoint"))?; - let daemon_id = required_environment("CODEX_CONNECTOR_IDUNN_DAEMON")?; - let runtime_id = required_environment("CODEX_CONNECTOR_RUNTIME_ID")?; - let contract = std::env::var("CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT") - .unwrap_or_else(|_| crate::idunn_health::CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT.to_string()); - let identity = PathBuf::from(required_environment( - "CODEX_CONNECTOR_IDUNN_HEALTH_IDENTITY", - )?); - let release = crate::idunn_health::active_release_binding( - &config.codex_executable, - &config.codex_executable_sha256, - ) - .map_err(CodexDaemonError::Health)?; - let activation = crate::idunn_health::active_activation_binding( - config_path, - config_bytes, - config, - caller_key_bindings, - ) - .map_err(CodexDaemonError::Health)?; - let mut publisher = crate::idunn_health::ProviderHealthPublisher::open( - endpoint, - daemon_id, - runtime_id, - contract, - &identity, - release.clone(), - activation.clone(), - ) - .map_err(CodexDaemonError::Health)?; - let published = publisher - .publish("warming", "traffic-admission-pending") - .map_err(CodexDaemonError::Health)?; - let traffic_admission = crate::idunn_health::ConnectorTrafficAdmissionGate::from_environment( - &release, - &activation, - &published, - ) - .map_err(CodexDaemonError::Health)?; - traffic_admission - .wait_until_granted(Duration::from_secs(120)) - .map_err(CodexDaemonError::Health)?; - Ok((traffic_admission, publisher)) +) -> Result, CodexDaemonError> { + if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_none() { + return Ok(None); + } + let capacity = u32::try_from(caller_request_capacity(&config.callers)?) + .map_err(|_| CodexDaemonError::InvalidConfig("runtime capability capacity"))?; + let publisher = + crate::idunn_health::RuntimePresencePublisher::open_from_environment(candidate, capacity) + .map_err(CodexDaemonError::Health)?; + Ok(Some(publisher)) } #[cfg(target_os = "linux")] -fn start_periodic_idunn_health_publisher( - mut publisher: crate::idunn_health::ProviderHealthPublisher, +fn start_periodic_runtime_presence_publisher( + mut publisher: crate::idunn_health::RuntimePresencePublisher, backend: Arc, ) -> Result<(), CodexDaemonError> { + let (state, detail) = provider_health_observation(&backend); + publisher + .publish(state, detail) + .map_err(CodexDaemonError::Health)?; thread::Builder::new() .name("codex-connector-health".to_string()) .spawn(move || { @@ -434,26 +416,131 @@ fn provider_health_state(ready: bool) -> (&'static str, &'static str) { } #[cfg(target_os = "linux")] -fn required_environment(name: &'static str) -> Result { - std::env::var(name) - .ok() - .filter(|value| !value.trim().is_empty() && value == value.trim()) - .ok_or(CodexDaemonError::InvalidConfig(name)) +fn effective_bind_from_environment(configured: SocketAddr) -> Result { + select_effective_bind( + configured, + std::env::var(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).ok(), + std::env::var(crate::idunn_health::IDUNN_CANDIDATE_BIND_ENVIRONMENT).ok(), + ) } -#[derive(Clone)] -pub(crate) struct LoadedCallerKeyBinding { - pub(crate) caller_runtime_id: String, - pub(crate) connection_key_file: PathBuf, - pub(crate) connection_key_epoch: u32, - pub(crate) raw_file_sha256: [u8; 32], +#[cfg(target_os = "linux")] +fn effective_runtime_paths_from_environment( + config: &CodexDaemonConfig, + managed_state_root: Option<&Path>, +) -> Result<(PathBuf, PathBuf, PathBuf), CodexDaemonError> { + if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_none() { + if managed_state_root.is_some() { + return Err(CodexDaemonError::InvalidConfig( + "Idunn state root outside a managed launch", + )); + } + return Ok(( + config.codex_executable.clone(), + config.codex_home.clone(), + config.replay_store.clone(), + )); + } + let managed_state_root = + managed_state_root.ok_or(CodexDaemonError::InvalidConfig("missing Idunn state root"))?; + + let running = std::env::current_exe() + .and_then(fs::canonicalize) + .map_err(|_| CodexDaemonError::InvalidConfig("managed Connector executable"))?; + managed_runtime_paths(config, &running, managed_state_root) +} + +#[cfg(target_os = "linux")] +fn managed_runtime_paths( + config: &CodexDaemonConfig, + running: &Path, + state_root: &Path, +) -> Result<(PathBuf, PathBuf, PathBuf), CodexDaemonError> { + let adjacent_codex = running + .parent() + .ok_or(CodexDaemonError::InvalidConfig( + "managed Connector executable", + ))? + .join("codex"); + let adjacent_codex = fs::canonicalize(adjacent_codex) + .map_err(|_| CodexDaemonError::InvalidConfig("managed Codex executable"))?; + let configured_codex = fs::canonicalize(&config.codex_executable) + .map_err(|_| CodexDaemonError::InvalidConfig("managed Codex executable"))?; + if configured_codex != adjacent_codex { + return Err(CodexDaemonError::InvalidConfig("managed Codex executable")); + } + + if !state_root.is_absolute() { + return Err(CodexDaemonError::InvalidConfig("managed state root")); + } + let canonical_state_root = fs::canonicalize(state_root) + .map_err(|_| CodexDaemonError::InvalidConfig("managed state root"))?; + let state_root_metadata = fs::symlink_metadata(state_root) + .map_err(|_| CodexDaemonError::InvalidConfig("managed state root"))?; + if canonical_state_root != state_root + || !state_root_metadata.is_dir() + || state_root_metadata.file_type().is_symlink() + { + return Err(CodexDaemonError::InvalidConfig("managed state root")); + } + + let codex_home = state_root.join("codex-home"); + if config.codex_home != codex_home { + return Err(CodexDaemonError::InvalidConfig("managed state layout")); + } + match fs::symlink_metadata(&codex_home) { + Ok(metadata) if metadata.is_dir() && !metadata.file_type().is_symlink() => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + _ => return Err(CodexDaemonError::InvalidConfig("managed state layout")), + } + + let replay_store = state_root.join("replay.cc"); + if config.replay_store != replay_store { + return Err(CodexDaemonError::InvalidConfig("managed state layout")); + } + match fs::symlink_metadata(&replay_store) { + Ok(metadata) + if metadata.is_file() + && !metadata.file_type().is_symlink() + && metadata.nlink() == 1 => {} + Ok(_) => return Err(CodexDaemonError::InvalidConfig("managed state layout")), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(_) => return Err(CodexDaemonError::InvalidConfig("managed state layout")), + } + Ok((adjacent_codex, codex_home, replay_store)) +} + +#[cfg(target_os = "linux")] +fn select_effective_bind( + configured: SocketAddr, + runtime_bundle: Option, + candidate_bind: Option, +) -> Result { + match (runtime_bundle, candidate_bind) { + (None, None) => Ok(configured), + (Some(bundle), Some(candidate)) + if !bundle.trim().is_empty() + && bundle.trim() == bundle + && candidate.trim() == candidate => + { + let bind = candidate + .parse::() + .map_err(|_| CodexDaemonError::InvalidConfig("Idunn candidate bind"))?; + if !bind.ip().is_loopback() { + return Err(CodexDaemonError::InvalidConfig("Idunn candidate bind")); + } + Ok(bind) + } + _ => Err(CodexDaemonError::InvalidConfig( + "partial Idunn managed runtime environment", + )), + } } fn load_admissions( config: &CodexDaemonConfig, -) -> Result<(Vec, Vec), CodexDaemonError> { +) -> Result, CodexDaemonError> { let mut admissions = Vec::with_capacity(config.callers.len()); - let mut bindings = Vec::with_capacity(config.callers.len()); for caller in &config.callers { let raw = Zeroizing::new( fs::read(&caller.connection_key_file).map_err(CodexDaemonError::ConnectionKey)?, @@ -473,22 +560,14 @@ fn load_admissions( ) .map_err(CodexDaemonError::Service)?, ); - bindings.push(LoadedCallerKeyBinding { - caller_runtime_id: caller.caller_runtime_id.clone(), - connection_key_file: caller.connection_key_file.clone(), - connection_key_epoch: caller.connection_key_epoch, - raw_file_sha256: sha2::Sha256::digest(raw.as_slice()).into(), - }); } - Ok((admissions, bindings)) + Ok(admissions) } fn serve_connection( mut stream: TcpStream, service: &Mutex, backend: &CodexProviderBackend, - #[cfg(target_os = "linux")] - traffic_admission: &crate::idunn_health::ConnectorTrafficAdmissionGate, max_frame_bytes: usize, timeout: Duration, ) -> Result<(), CodexDaemonError> { @@ -499,13 +578,6 @@ fn serve_connection( let request = read_transport_frame(&mut stream, max_frame_bytes).map_err(frame_error)?; let envelope: CodexTransportEnvelope = rmp_serde::from_slice(&request).map_err(|_| CodexDaemonError::FrameEncoding)?; - #[cfg(target_os = "linux")] - // This is the traffic-to-Connector authority transfer. The complete - // request is inert until the current root grant admits service.begin; - // backend consequence remains owned by the admitted request claim. - traffic_admission - .require_current() - .map_err(CodexDaemonError::Health)?; let admission = service .lock() .map_err(|_| CodexDaemonError::ServicePoisoned)? @@ -592,7 +664,7 @@ pub enum CodexDaemonError { #[error("connector request thread failed")] Thread(#[source] std::io::Error), #[cfg(target_os = "linux")] - #[error("connector signed health or traffic authority failed: {0}")] + #[error("connector runtime presence failed: {0}")] Health(#[source] anyhow::Error), #[cfg(target_os = "linux")] #[error("connector health thread could not start")] @@ -661,10 +733,8 @@ mod tests { let expected = config(&root); write_daemon_config(&path, &expected).unwrap(); assert_eq!(load_daemon_config(&path).unwrap(), expected); - let exact_bytes = fs::read(&path).unwrap(); - let (loaded, loaded_bytes) = load_daemon_config_snapshot(&path).unwrap(); + let loaded = load_daemon_config_read_only(&path).unwrap(); assert_eq!(loaded, expected); - assert_eq!(loaded_bytes, exact_bytes); let mut stale = expected; stale.epoch = 0; @@ -689,12 +759,60 @@ mod tests { let lock_path = path.with_file_name("connector.cc.lock"); fs::remove_file(&lock_path).unwrap(); - assert!(load_daemon_config_snapshot(&path).is_err()); + assert!(load_daemon_config_read_only(&path).is_err()); assert!(!lock_path.exists()); assert_eq!(fs::read(&path).unwrap(), before); fs::remove_dir_all(root).unwrap(); } + #[test] + #[cfg(target_os = "linux")] + fn managed_runtime_refuses_a_mutable_config_authority() { + let root = std::env::temp_dir().join(format!( + "codex-connector-managed-config-{}-{}", + std::process::id(), + unix_ms().unwrap() + )); + fs::create_dir_all(&root).unwrap(); + let path = root.join("connector.cc"); + write_daemon_config(&path, &config(&root)).unwrap(); + + assert!(crate::idunn_health::require_managed_config_store(&path).is_err()); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + #[cfg(target_os = "linux")] + fn idunn_candidate_bind_replaces_config_only_for_a_complete_managed_launch() { + let configured = "127.0.0.1:4103".parse().unwrap(); + let candidate = "127.0.0.1:18831".parse().unwrap(); + assert_eq!( + select_effective_bind(configured, None, None).unwrap(), + configured + ); + assert_eq!( + select_effective_bind( + configured, + Some("/run/idunn/runtime/instance".into()), + Some("127.0.0.1:18831".into()), + ) + .unwrap(), + candidate + ); + assert!( + select_effective_bind(configured, Some("/run/idunn/runtime/instance".into()), None,) + .is_err() + ); + assert!( + select_effective_bind( + configured, + Some("/run/idunn/runtime/instance".into()), + Some("0.0.0.0:18831".into()), + ) + .is_err() + ); + } + #[test] fn single_caller_initializer_derives_daemon_bounds_without_reowning_policy() { let root = Path::new("/srv/codex-connector"); @@ -782,6 +900,66 @@ mod tests { )); } + #[test] + fn provider_capacity_is_caller_execution_capacity_not_listener_capacity() { + let root = Path::new("/srv/codex-connector"); + let mut config = config(root); + config.max_connections = 64; + assert_eq!(caller_request_capacity(&config.callers).unwrap(), 4); + assert_eq!(required_connection_capacity(&config.callers).unwrap(), 8); + } + + #[test] + #[cfg(target_os = "linux")] + fn managed_runtime_uses_the_sealed_adjacent_codex_and_declared_state_layout() { + let root = std::env::temp_dir().join(format!( + "codex-connector-layout-{}-{}", + std::process::id(), + unix_ms().unwrap() + )); + let release = root.join("release"); + let state = root.join("state"); + fs::create_dir_all(release.clone()).unwrap(); + fs::create_dir_all(&state).unwrap(); + fs::write(release.join("codex-connector"), b"connector").unwrap(); + fs::write(release.join("codex"), b"codex").unwrap(); + let mut config = config(&root); + config.codex_executable = release.join("codex"); + config.codex_home = state.join("codex-home"); + config.replay_store = state.join("replay.cc"); + + let (_, codex_home, replay_store) = + managed_runtime_paths(&config, &release.join("codex-connector"), &state).unwrap(); + assert_eq!(codex_home, state.join("codex-home")); + assert_eq!( + replay_store, + fs::canonicalize(&state).unwrap().join("replay.cc") + ); + + fs::write(&replay_store, b"replay").unwrap(); + let replay_alias = state.join("replay-alias.cc"); + fs::hard_link(&replay_store, &replay_alias).unwrap(); + assert!(managed_runtime_paths(&config, &release.join("codex-connector"), &state).is_err()); + fs::remove_file(&replay_alias).unwrap(); + fs::remove_file(&replay_store).unwrap(); + std::os::unix::fs::symlink(state.join("missing-replay.cc"), &replay_store).unwrap(); + assert!(managed_runtime_paths(&config, &release.join("codex-connector"), &state).is_err()); + fs::remove_file(&replay_store).unwrap(); + + config.codex_executable = root.join("other-codex"); + fs::write(&config.codex_executable, b"other").unwrap(); + assert!(managed_runtime_paths(&config, &release.join("codex-connector"), &state).is_err()); + + config.codex_executable = release.join("codex"); + let other_state = root.join("other-state"); + fs::create_dir_all(other_state.join("codex-home")).unwrap(); + assert!( + managed_runtime_paths(&config, &release.join("codex-connector"), &other_state,) + .is_err() + ); + fs::remove_dir_all(root).unwrap(); + } + #[test] fn direct_pipe_frame_is_big_endian_and_refuses_oversize_before_allocation() { let mut encoded = Vec::new(); diff --git a/src/idunn_health.rs b/src/idunn_health.rs index aea11fa..f871b68 100644 --- a/src/idunn_health.rs +++ b/src/idunn_health.rs @@ -1,1359 +1,645 @@ +use std::fs::{File, OpenOptions}; use std::net::{SocketAddr, UdpSocket}; +use std::os::fd::{AsRawFd, FromRawFd, RawFd}; use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; -use std::time::{Duration, Instant}; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use anyhow::{Context, Result, anyhow, bail}; -use cultcache_rs::{CacheBackingStore, SingleFileMessagePackBackingStore}; +use anyhow::{Context, Result, anyhow, bail, ensure}; +use cultcache_rs::{DatabaseEntry, SingleFileMessagePackBackingStore}; use cultnet_rs::{ CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, - CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract, - encode_cultnet_message_to_vec, + CultNetRudpReliableSendStatus, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, CultNetWireContract, + GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, GameCultProviderHealthIdentity, + GameCultRuntimeCapability, GameCultRuntimePresenceHealthPurpose, + GameCultRuntimePresenceHealthRecord, IDUNN_EXPECTED_INCARNATION_SCHEMA, + IDUNN_PROCESS_WRITE_LEASE_SCHEMA, IDUNN_RUNTIME_ACTIVATION_CREDENTIAL_NAME, + IDUNN_RUNTIME_ACTIVATION_SCHEMA, IdunnExpectedIncarnationRecord, IdunnProcessWriteLeaseRecord, + IdunnRuntimeActivationRecord, IdunnRuntimeActivationSigner, ServiceIdentitySigner, + encode_cultnet_message_to_vec, open_service_identity_credential_reader, }; -use ed25519_dalek::{Signer, SigningKey}; -use serde::{Deserialize, Serialize}; -use sha2::{Digest, Sha256}; -pub const CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT: &str = - "codex-connector.cultnet-rudp-service-health"; -const HEALTH_SCHEMA: &str = "idunn.signed_daemon_health.v1"; -const IDENTITY_SCHEMA: &str = "gamecult.provider_health_identity.private.v1"; -const IDENTITY_TYPE: &str = "gamecult.provider_health_identity.private.v1"; -const IDENTITY_KEY: &str = "gamecult-provider-health-identity"; -const ID_DOMAIN: &[u8] = b"gamecult.provider-health.identity.v1\0"; -const SIGNATURE_DOMAIN: &[u8] = b"gamecult.provider-health.signature.v1\0"; -const SIGNATURE_PURPOSE: &[u8] = b"idunn.signed_daemon_health.v1"; -const PROTECTOR_CONTEXT: &str = "gamecult-provider-health-identity-v1"; +pub(crate) const IDUNN_RUNTIME_BUNDLE_ENVIRONMENT: &str = "GAMECULT_IDUNN_RUNTIME_BUNDLE"; +pub(crate) const IDUNN_CANDIDATE_BIND_ENVIRONMENT: &str = "GAMECULT_IDUNN_CANDIDATE_BIND"; +const IDUNN_PROCESS_WRITE_LEASE_ENVIRONMENT: &str = "GAMECULT_IDUNN_PROCESS_WRITE_LEASE"; +const CULTNET_RUDP_ENVIRONMENT: &str = "CODEX_CONNECTOR_CULTNET_RUDP"; +const SYSTEMD_LISTEN_PID_ENVIRONMENT: &str = "LISTEN_PID"; +const SYSTEMD_LISTEN_FDS_ENVIRONMENT: &str = "LISTEN_FDS"; +const SYSTEMD_LISTEN_FDNAMES_ENVIRONMENT: &str = "LISTEN_FDNAMES"; +const SYSTEMD_LISTEN_FDS_START: RawFd = 3; +const ACTIVATION_SIGNER_FD_NAME: &str = IDUNN_RUNTIME_ACTIVATION_CREDENTIAL_NAME; +const PROVIDER_SIGNER_FD_NAME: &str = "gamecult-runtime-presence-identity"; + +const CONNECTOR_TARGET: &str = "codex-connector"; +const CONNECTOR_CAPABILITY: &str = "gamecult.codex.subscription-inference"; +const CONNECTOR_COMPATIBILITY: &str = "v2"; +pub(crate) const CONNECTOR_HEALTH_CONTRACT: &str = "codex-connector.runtime-health.v1"; +pub(crate) const CONNECTOR_STATE_SCHEMA_GENERATION: &str = "connector-v1"; +// Odin's canonical rmp-serde digest of deployment/idunn/recipe.toml [state]. +pub(crate) const CONNECTOR_STATE_CONTRACT_SHA256: &str = + "sha256-61da5fa6ee0ed2971137a724d6d880e3def273c43e160c5001ea56f57bf11a44"; +const ODIN_RENDEZVOUS_CAPABILITY: &str = "odin.verse-rendezvous"; +const ODIN_RENDEZVOUS_SCHEMA: &str = "odin.verse-topology.v1"; +const ODIN_RENDEZVOUS_COMPATIBILITY: &str = "v1"; const RUDP_PROTOCOL_ID: &str = "cultnet.transport.rudp.v0"; -const DEPLOYMENT_SCHEMA: &str = "gamecult.codex_connector.deployment.v2"; -const ACTIVATION_SCHEMA: &str = "gamecult.codex_connector.activation.v2"; -const TRAFFIC_ADMISSION_PATH: &str = "/etc/gamecult/codex-connector/runtime/traffic-admission.cc"; -const TRAFFIC_ADMISSION_TYPE: &str = "idunn.runtime_traffic_admission"; -const TRAFFIC_ADMISSION_SCHEMA: &str = "idunn.runtime_traffic_admission.v2"; -const TRAFFIC_ADMISSION_KEY: &str = "yggdrasil-codex-connector"; -const TOOLCHAIN_SCHEMA: &str = "gamecult.codex_connector.codex_toolchain.v1"; -const SOURCE_TOOLCHAIN_MANIFEST: &str = include_str!("../deployment/codex-linux-x64.manifest"); -const BUILD_COMMIT: &str = match option_env!("CODEX_CONNECTOR_BUILD_COMMIT") { - Some(value) => value, - None => "development", -}; -// Idunn's daemon-health ingress owns one shared RUDP connection contract. -// Publisher identity is carried by the signed record, not by transport IDs. -const RUDP_CONNECTION_ID: u32 = 0x1d0d_0001; +const ODIN_CULTMESH_DOCUMENT_CATALOG_CONNECTION_ID: u32 = 0x0d1d_0002; -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -struct ProviderHealthIdentity { - schema_version: String, - identity_id: String, - #[serde(with = "serde_bytes")] - public_key: Vec, - #[serde(with = "serde_bytes")] - protected_private_seed: Vec, - protector_kind: String, - protector_binding: String, - protector_version: String, - assurance: String, - created_at: String, - #[serde(with = "serde_bytes")] - enrollment_nonce: Vec, +pub(crate) fn require_managed_config_store(path: &Path) -> Result<()> { + require_root_read_only_file(path, "Connector managed configuration")?; + require_root_read_only_file( + &sibling_lock_path(path)?, + "Connector managed configuration lock", + )?; + require_root_controlled_directory_chain( + path.parent() + .context("managed configuration has no parent")?, + "Connector managed configuration parent", + ) } -struct ProviderHealthSigner { - entry: ProviderHealthIdentity, - key: SigningKey, +struct RuntimeAuthorityMaterial { + expected: IdunnExpectedIncarnationRecord, + expected_sha256: String, + activation: IdunnRuntimeActivationRecord, + activation_sha256: String, + activation_signer: IdunnRuntimeActivationSigner, + provider_signer: ServiceIdentitySigner, } -#[derive(Clone, Debug, PartialEq, Eq)] -pub(crate) struct ConnectorReleaseBinding { - release_id: String, - release_witness_sha256: String, - source_commit: String, - deployment_id: String, +struct RuntimeSignerDescriptors { + activation: File, + provider: File, } -#[derive(Clone, Debug, PartialEq, Eq)] -pub(crate) struct ConnectorActivationBinding { - activation_witness_sha256: String, -} - -// Idunn owns and writes this CultCache schema. Connector carries only the -// exact read-side projection needed to consume that root admission. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -struct TrafficAdmissionConsumerProjection { - schema_version: String, - daemon_id: String, - release_id: String, - release_witness_sha256: String, - source_commit: String, - deployment_id: String, - activation_witness_sha256: String, - signed_health_sha256: String, - publisher_incarnation_id: String, - publisher_sequence: u64, - signer_identity_id: String, - runtime_process_id: u32, - runtime_process_starttime_ticks: u64, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -struct RuntimeProcessInstanceBinding { - process_id: u32, - starttime_ticks: u64, +pub(crate) struct RuntimePresencePublisher { + endpoint: SocketAddr, + authority: RuntimeAuthorityMaterial, + bound_endpoint: String, + capabilities: Vec, + write_lease_path: PathBuf, + write_lease: Option, + sequence: u64, } #[derive(Clone)] -pub(crate) struct ConnectorTrafficAdmissionGate { +pub(crate) struct ProcessWriteLeaseGuard { + held: Arc, +} + +struct HeldProcessWriteLease { + // Idunn uses this pre-created sibling lock for write-lease replacement. + // Keeping it shared for the whole process fences both Connector's replay + // commits and the official app-server's credential refresh writes. + _lock: File, path: PathBuf, - expected: TrafficAdmissionConsumerProjection, + expected: IdunnExpectedIncarnationRecord, + expected_sha256: String, + activation: IdunnRuntimeActivationRecord, + activation_sha256: String, + warming_presence_sha256: String, + lease_sha256: String, } -pub(crate) struct PublishedHealthStatementIdentity { - daemon_id: String, - signed_health_sha256: String, - publisher_incarnation_id: String, - publisher_sequence: u64, - signer_identity_id: String, +impl ProcessWriteLeaseGuard { + fn sha256(&self) -> &str { + &self.held.lease_sha256 + } + + fn require_current(&self) -> Result<()> { + let observed = load_process_write_lease( + &self.held.path, + &self.held.expected, + &self.held.expected_sha256, + &self.held.activation, + &self.held.activation_sha256, + &self.held.warming_presence_sha256, + )?; + ensure!( + observed.as_deref() == Some(self.sha256()), + "process-write lease is no longer the admitted lifetime grant" + ); + Ok(()) + } } -impl ConnectorTrafficAdmissionGate { - pub(crate) fn from_environment( - release: &ConnectorReleaseBinding, - activation: &ConnectorActivationBinding, - published: &PublishedHealthStatementIdentity, - ) -> Result { - let raw_path = std::env::var("CODEX_CONNECTOR_TRAFFIC_ADMISSION") - .context("CODEX_CONNECTOR_TRAFFIC_ADMISSION is required for signed health")?; - let path = fixed_traffic_admission_path(&raw_path)?; +impl RuntimePresencePublisher { + pub(crate) fn open_from_environment(bound: SocketAddr, capacity: u32) -> Result { + ensure!( + bound.ip().is_loopback() && bound.port() != 0, + "Connector candidate bind is not a fixed loopback endpoint" + ); + ensure!(capacity > 0, "Connector capability capacity is zero"); + let bundle = PathBuf::from(required_environment(IDUNN_RUNTIME_BUNDLE_ENVIRONMENT)?); + let authority = load_runtime_authority(&bundle)?; + let candidate: SocketAddr = required_environment(IDUNN_CANDIDATE_BIND_ENVIRONMENT)? + .parse() + .context("parsing Idunn candidate bind")?; + ensure!( + candidate == bound, + "bound endpoint differs from Idunn candidate bind" + ); + let bound_endpoint = format!("tcp://{bound}"); + let route = authority + .expected + .route + .as_ref() + .context("Connector Expected projection has no admitted route")?; + ensure!( + route.transport == "tcp" && route.candidate_endpoint == bound_endpoint, + "Connector bound endpoint differs from Expected" + ); + + let endpoint: SocketAddr = required_environment(CULTNET_RUDP_ENVIRONMENT)? + .parse() + .context("parsing Connector CultNet RUDP endpoint")?; + require_odin_rendezvous(&authority.expected, endpoint)?; + + let capability = GameCultRuntimeCapability { + capability: CONNECTOR_CAPABILITY.into(), + schema: crate::ENVELOPE_SCHEMA_ID.into(), + compatibility: CONNECTOR_COMPATIBILITY.into(), + capacity, + }; + require_connector_expected_contract(&authority.expected, &capability)?; + let write_lease_path = + PathBuf::from(required_environment(IDUNN_PROCESS_WRITE_LEASE_ENVIRONMENT)?); + Ok(Self { - path, - expected: traffic_admission_expectation(release, activation, published)?, + endpoint, + authority, + bound_endpoint, + capabilities: vec![capability], + write_lease_path, + write_lease: None, + sequence: 0, }) } - pub(crate) fn wait_until_granted(&self, timeout: Duration) -> Result<()> { + pub(crate) fn acquire_process_write_lease( + &mut self, + warming_presence_sha256: &str, + timeout: Duration, + ) -> Result<()> { + let path = &self.write_lease_path; let started = Instant::now(); - let mut last_invalid_observation = None; loop { - match self.is_current() { - Ok(true) => return Ok(()), - Ok(false) => last_invalid_observation = None, - Err(error) => last_invalid_observation = Some(format!("{error:#}")), - } + let last_error = match acquire_process_write_lease_guard( + path, + &self.authority.expected, + &self.authority.expected_sha256, + &self.authority.activation, + &self.authority.activation_sha256, + warming_presence_sha256, + ) { + Ok(Some(guard)) => { + self.write_lease = Some(guard); + return Ok(()); + } + Ok(None) => None, + Err(error) => Some(format!("{error:#}")), + }; if started.elapsed() >= timeout { - let detail = last_invalid_observation + let detail = last_error .as_deref() - .unwrap_or("the fixed admission record remained absent"); + .unwrap_or("the process-write-lease record remained absent"); bail!( - "timed out waiting for exact sealed root traffic admission {}: {}", - self.path.display(), - detail, + "timed out waiting for Idunn process-write lease {}: {detail}", + path.display() ); } std::thread::sleep(Duration::from_millis(100)); } } - pub(crate) fn require_current(&self) -> Result<()> { - if !self.is_current()? { - bail!( - "root traffic admission disappeared from {}", - self.path.display() - ); - } - Ok(()) + pub(crate) fn process_write_lease_guard(&self) -> Result { + self.write_lease + .clone() + .context("Connector has not acquired its process-write lease") } - fn is_current(&self) -> Result { - match std::fs::symlink_metadata(&self.path) { - Ok(_) => {} - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), - Err(error) => { - return Err(error).with_context(|| { - format!( - "inspecting connector traffic admission {}", - self.path.display() - ) - }); - } + pub(crate) fn publish(&mut self, state: &str, detail: &str) -> Result { + if let Some(guard) = &self.write_lease { + guard.require_current()?; } - let lock_path = sibling_lock_path(&self.path)?; - require_runtime_capability_file(&self.path, "connector traffic admission")?; - require_runtime_capability_file(&lock_path, "connector traffic admission lock")?; - // This is a pull-only capability read. Connector never registers, - // creates, migrates, or writes Idunn's traffic-admission store. - SingleFileMessagePackBackingStore::new(&self.path) - .with_read_only_shared_snapshot(|entries| { - let [envelope] = entries.as_slice() else { - bail!("connector traffic admission store must contain exactly one record"); - }; - if envelope.key != TRAFFIC_ADMISSION_KEY - || envelope.r#type != TRAFFIC_ADMISSION_TYPE - || envelope.schema_id.as_deref() != Some(TRAFFIC_ADMISSION_SCHEMA) - { - bail!("connector traffic admission store has the wrong typed envelope"); - } - let admitted = decode_traffic_admission_payload(&envelope.payload)?; - require_exact_traffic_admission(&admitted, &self.expected)?; - Ok(true) - }) - .with_context(|| { - format!( - "reading connector traffic admission {}", - self.path.display() - ) - }) - } -} - -fn fixed_traffic_admission_path(raw_path: &str) -> Result { - if raw_path != TRAFFIC_ADMISSION_PATH { - bail!("connector traffic admission path is not the fixed root policy path"); - } - Ok(PathBuf::from(raw_path)) -} - -impl ConnectorActivationBinding { - fn validate(&self) -> Result<()> { - if !self - .activation_witness_sha256 - .strip_prefix("sha256-") - .is_some_and(|digest| is_lower_hex(digest, 64)) - { - bail!("connector activation binding is malformed"); - } - Ok(()) - } -} - -impl ConnectorReleaseBinding { - fn validate(&self) -> Result<()> { - require_id(&self.release_id, "release id")?; - require_deployment_id(&self.deployment_id)?; - if !self - .release_witness_sha256 - .strip_prefix("sha256-") - .is_some_and(|digest| is_lower_hex(digest, 64)) - || !is_lower_hex(&self.source_commit, 40) - { - bail!("connector release binding is malformed"); - } - Ok(()) - } -} - -#[derive(Debug, PartialEq, Eq)] -struct DeploymentWitness { - release_id: String, - source_commit: String, - connector_binary_sha256: String, - codex_package_url: String, - codex_package_sha256: String, - codex_binary_sha256: String, -} - -#[derive(Clone, Debug, PartialEq, Eq)] -struct ToolchainManifest { - package_url: String, - package_sha256: String, - binary_sha256: String, -} - -#[derive(Clone, Debug, PartialEq, Eq)] -struct ActivationWitness { - activation_id: String, - config_path: String, - config_sha256: String, - config_epoch: u32, - bind: String, - codex_executable_path: String, - codex_executable_sha256: String, - codex_home_path: String, - replay_store_path: String, - max_frame_bytes: usize, - max_connections: usize, - socket_timeout_ms: u64, - max_expiry_skew_ms: u64, - callers: Vec, -} - -#[derive(Clone, Debug, PartialEq, Eq)] -struct ActivationCallerWitness { - runtime_id: String, - connection_key_path: String, - connection_key_sha256: String, - connection_key_epoch: u32, - allowed_models: Vec, - max_concurrent_requests: usize, - max_payload_bytes: usize, - max_output_tokens: u32, -} - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -struct SignedDaemonHealth { - schema_version: String, - daemon_id: String, - health_contract: String, - source_runtime_id: String, - state: String, - detail: String, - signer_identity_id: String, - publisher_incarnation_id: String, - publisher_sequence: u64, - observed_at_unix_millis: u64, - release_id: Option, - release_witness_sha256: Option, - source_commit: Option, - deployment_id: Option, - signature_algorithm: String, - #[serde(with = "serde_bytes")] - signature: Vec, - private_state_exposed: bool, - activation_witness_sha256: Option, -} - -impl SignedDaemonHealth { - fn validate(&self) -> Result<()> { - for (value, label) in [ - (&self.daemon_id, "daemon id"), - (&self.health_contract, "health contract"), - (&self.source_runtime_id, "runtime id"), - (&self.signer_identity_id, "signer identity"), - (&self.publisher_incarnation_id, "publisher incarnation"), - ] { - require_id(value, label)?; - } - if self.schema_version != HEALTH_SCHEMA - || !matches!( - self.state.as_str(), - "active" | "warming" | "degraded" | "failed" - ) - || self.detail.len() > 512 - || self.detail.chars().any(char::is_control) - || self.publisher_sequence == 0 - || self.observed_at_unix_millis == 0 - || self.signature_algorithm != "ed25519" - || self.signature.len() != 64 - || self.private_state_exposed - { - bail!("signed daemon health shape is invalid"); - } - ConnectorReleaseBinding { - release_id: self - .release_id - .clone() - .ok_or_else(|| anyhow!("signed health release id is absent"))?, - release_witness_sha256: self - .release_witness_sha256 - .clone() - .ok_or_else(|| anyhow!("signed health release witness is absent"))?, - source_commit: self - .source_commit - .clone() - .ok_or_else(|| anyhow!("signed health source commit is absent"))?, - deployment_id: self - .deployment_id - .clone() - .ok_or_else(|| anyhow!("signed health deployment id is absent"))?, - } - .validate()?; - ConnectorActivationBinding { - activation_witness_sha256: self - .activation_witness_sha256 - .clone() - .ok_or_else(|| anyhow!("signed health activation witness is absent"))?, - } - .validate()?; - Ok(()) - } -} - -pub(crate) struct ProviderHealthPublisher { - endpoint: SocketAddr, - daemon_id: String, - runtime_id: String, - contract: String, - signer: ProviderHealthSigner, - release: ConnectorReleaseBinding, - activation: ConnectorActivationBinding, - incarnation: String, - sequence: u64, -} - -impl ProviderHealthPublisher { - pub(crate) fn open( - endpoint: SocketAddr, - daemon_id: impl Into, - runtime_id: impl Into, - contract: impl Into, - identity_store: &Path, - release: ConnectorReleaseBinding, - activation: ConnectorActivationBinding, - ) -> Result { - let daemon_id = daemon_id.into(); - let runtime_id = runtime_id.into(); - let contract = contract.into(); - require_id(&daemon_id, "daemon id")?; - require_id(&runtime_id, "runtime id")?; - require_id(&contract, "health contract")?; - release.validate()?; - activation.validate()?; - Ok(Self { - endpoint, - daemon_id, - runtime_id, - contract, - signer: open_identity(identity_store)?, - release, - activation, - incarnation: uuid::Uuid::new_v4().to_string(), - sequence: 0, - }) + let record = self.signed_record(state, detail, unix_millis()?)?; + let payload = rmp_serde::to_vec(&record)?; + ensure!( + rmp_serde::from_slice::(&payload)? == record, + "runtime presence did not round-trip canonically" + ); + let sha256 = record.canonical_sha256()?; + publish_presence( + self.endpoint, + &self.authority.expected.target, + &self.authority.expected.runtime_id, + &record, + payload, + )?; + Ok(sha256) } - pub(crate) fn publish( + fn signed_record( &mut self, state: &str, detail: &str, - ) -> Result { + observed_at_unix_millis: u64, + ) -> Result { + if state == "warming" { + ensure!( + self.write_lease.is_none(), + "warming presence cannot carry a process-write lease" + ); + } else if state == "active" && self.authority.expected.write_lease_required { + ensure!( + self.write_lease.is_some(), + "active stateful presence lacks its process-write lease" + ); + } self.sequence = self .sequence .checked_add(1) - .ok_or_else(|| anyhow!("health publisher sequence overflow"))?; - let mut record = SignedDaemonHealth { - schema_version: HEALTH_SCHEMA.into(), - daemon_id: self.daemon_id.clone(), - health_contract: self.contract.clone(), - source_runtime_id: self.runtime_id.clone(), + .ok_or_else(|| anyhow!("runtime presence publisher sequence overflow"))?; + let expected = &self.authority.expected; + let activation = &self.authority.activation; + let mut record = GameCultRuntimePresenceHealthRecord { + schema_version: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(), + target: expected.target.clone(), + expected_projection_sha256: self.authority.expected_sha256.clone(), + plan_id: expected.plan_id.clone(), + incarnation_id: expected.incarnation_id.clone(), + sealed_release_id: expected.sealed_release_id.clone(), + activation_witness_sha256: self.authority.activation_sha256.clone(), + state_schema_generation: expected.state_schema_generation.clone(), + state_contract_sha256: expected.state_contract_sha256.clone(), + runtime_id: expected.runtime_id.clone(), + runtime_instance_id: activation.runtime_instance_id.clone(), + bound_endpoint: Some(self.bound_endpoint.clone()), + capabilities: self.capabilities.clone(), + health_contract: expected.health_contract.clone(), state: state.into(), detail: detail.into(), - signer_identity_id: self.signer.entry.identity_id.clone(), - publisher_incarnation_id: self.incarnation.clone(), + write_lease_sha256: self + .write_lease + .as_ref() + .map(|guard| guard.sha256().to_string()), + signer_identity_id: self.authority.provider_signer.entry().identity_id.clone(), publisher_sequence: self.sequence, - observed_at_unix_millis: chrono::Utc::now().timestamp_millis().try_into()?, - release_id: Some(self.release.release_id.clone()), - release_witness_sha256: Some(self.release.release_witness_sha256.clone()), - source_commit: Some(self.release.source_commit.clone()), - deployment_id: Some(self.release.deployment_id.clone()), + observed_at_unix_millis, signature_algorithm: "ed25519".into(), signature: Vec::new(), - private_state_exposed: false, - activation_witness_sha256: Some(self.activation.activation_witness_sha256.clone()), + activation_signer_identity_id: activation.activation_signer_identity_id.clone(), + activation_signature: Vec::new(), }; - let unsigned = unsigned_record(&record)?; - record.signature = self - .signer - .key - .sign(&signing_message(&unsigned)) - .to_bytes() - .to_vec(); + let proof_payload = record.canonical_proof_payload()?; + let provider_proof = self + .authority + .provider_signer + .sign::(&proof_payload); + ensure!( + provider_proof.identity_id == record.signer_identity_id, + "provider signer identity changed after runtime admission" + ); + record.signature = provider_proof.signature; + record.activation_signature = self + .authority + .activation_signer + .sign_presence_proof(&record)?; record.validate()?; - let signed_health_sha256 = publish(self.endpoint, &self.runtime_id, &record)?; - Ok(PublishedHealthStatementIdentity { - daemon_id: record.daemon_id, - signed_health_sha256, - publisher_incarnation_id: record.publisher_incarnation_id, - publisher_sequence: record.publisher_sequence, - signer_identity_id: record.signer_identity_id, + Ok(record) + } +} + +fn load_runtime_authority(bundle: &Path) -> Result { + require_runtime_bundle(bundle)?; + let (expected_key, expected_payload) = read_immutable_runtime_record( + &bundle.join("expected.cc"), + IdunnExpectedIncarnationRecord::TYPE, + IDUNN_EXPECTED_INCARNATION_SCHEMA, + )?; + let expected = IdunnExpectedIncarnationRecord::decode_canonical(&expected_payload)?; + ensure!( + expected_key == expected.target, + "Expected record key is not its target" + ); + let (activation_key, activation_payload) = read_immutable_runtime_record( + &bundle.join("activation.cc"), + IdunnRuntimeActivationRecord::TYPE, + IDUNN_RUNTIME_ACTIVATION_SCHEMA, + )?; + let activation = IdunnRuntimeActivationRecord::decode_canonical(&activation_payload)?; + ensure!( + activation_key == expected.target, + "activation record key is not the Expected target" + ); + let expected_sha256 = expected.canonical_sha256()?; + ensure!( + activation.expected_projection_sha256 == expected_sha256 + && activation.runtime_id == expected.runtime_id, + "runtime activation does not bind the bundled Expected projection" + ); + + let RuntimeSignerDescriptors { + activation: activation_credential, + provider: provider_identity, + } = take_runtime_signer_descriptors_from_environment()?; + let activation_signer = + IdunnRuntimeActivationSigner::from_credential_reader(activation_credential)?; + ensure!( + activation_signer.identity_id() == activation.activation_signer_identity_id + && activation_signer.public_key() == activation.activation_signer_public_key, + "runtime activation credential does not belong to the Idunn activation record" + ); + + // The shared signer owns profile validation, platform binding, seed + // unprotection, and domain-separated signing. Connector never parses or + // exports stable private key material. + let provider_signer = open_service_identity_credential_reader::( + provider_identity, + )?; + ensure!( + provider_signer.entry().identity_id == expected.expected_signer_identity_id, + "provider signer is not the identity selected by Expected" + ); + + Ok(RuntimeAuthorityMaterial { + expected, + expected_sha256, + activation_sha256: activation.canonical_sha256()?, + activation, + activation_signer, + provider_signer, + }) +} + +fn read_immutable_runtime_record( + path: &Path, + expected_type: &str, + expected_schema: &str, +) -> Result<(String, Vec)> { + require_root_read_only_file(path, "runtime authority record")?; + require_root_read_only_file(&sibling_lock_path(path)?, "runtime authority lock")?; + SingleFileMessagePackBackingStore::new(path).with_read_only_shared_snapshot(|entries| { + let [envelope] = entries.as_slice() else { + bail!("runtime authority store must contain exactly one record"); + }; + ensure!( + envelope.r#type == expected_type + && envelope.schema_id.as_deref() == Some(expected_schema), + "runtime authority store has the wrong typed envelope" + ); + Ok((envelope.key.clone(), envelope.payload.clone())) + }) +} + +fn require_odin_rendezvous( + expected: &IdunnExpectedIncarnationRecord, + endpoint: SocketAddr, +) -> Result<()> { + let dependency = expected + .dependencies + .iter() + .find(|dependency| { + dependency.capability == ODIN_RENDEZVOUS_CAPABILITY + && dependency.schema == ODIN_RENDEZVOUS_SCHEMA + && dependency.compatibility == ODIN_RENDEZVOUS_COMPATIBILITY }) - } -} - -pub(crate) fn active_release_binding( - codex_executable: &Path, - configured_codex_sha256: &[u8; 32], -) -> Result { - let connector_executable = std::env::current_exe().context("locating active connector")?; - let release_root = connector_executable - .parent() - .ok_or_else(|| anyhow!("active connector has no release directory"))?; - let witness_path = release_root.join("DEPLOYMENT"); - let expected_codex = release_root.join("codex"); - for (path, label, directory) in [ - (release_root, "connector release directory", true), - ( - connector_executable.as_path(), - "active connector executable", - false, - ), - (witness_path.as_path(), "connector release witness", false), - (expected_codex.as_path(), "active Codex executable", false), - ] { - require_root_sealed(path, label, directory)?; - } - let source_toolchain = parse_toolchain_manifest(SOURCE_TOOLCHAIN_MANIFEST)?; - release_binding_from_artifacts( - &connector_executable, - codex_executable, - configured_codex_sha256, - BUILD_COMMIT, - &required_runtime_deployment_id()?, - &source_toolchain, - ) -} - -pub(crate) fn active_activation_binding( - config_path: &Path, - config_bytes: &[u8], - config: &crate::daemon::CodexDaemonConfig, - callers: &[crate::daemon::LoadedCallerKeyBinding], -) -> Result { - let witness_path = PathBuf::from( - std::env::var("CODEX_CONNECTOR_ACTIVATION_WITNESS") - .context("CODEX_CONNECTOR_ACTIVATION_WITNESS is required for signed health")?, + .context("Expected projection has no Odin rendezvous dependency")?; + ensure!( + dependency.kind == "bootstrap" + && dependency.startup == "before-start" + && dependency.provider_id.is_some() + && dependency.provider_authority.as_deref() == Some("managed-incarnation") + && dependency.provider_expected_projection_sha256.is_some(), + "Odin rendezvous dependency is not a resolved managed bootstrap" ); - if !witness_path.is_absolute() { - bail!("connector activation witness path is not absolute"); - } - require_root_controlled_runtime_file(config_path, "connector configuration")?; - require_root_controlled_runtime_file(&witness_path, "connector activation witness")?; - for caller in callers { - require_root_controlled_runtime_file(&caller.connection_key_file, "connector caller key")?; - } - activation_binding_from_material(config_path, config_bytes, config, callers, &witness_path) -} - -fn activation_binding_from_material( - config_path: &Path, - config_bytes: &[u8], - config: &crate::daemon::CodexDaemonConfig, - callers: &[crate::daemon::LoadedCallerKeyBinding], - witness_path: &Path, -) -> Result { - let witness_bytes = std::fs::read(witness_path) - .with_context(|| format!("reading activation witness {}", witness_path.display()))?; - let witness = parse_activation_witness(&witness_bytes)?; - let expected = activation_expectation( - witness.activation_id.clone(), - config_path, - config_bytes, - config, - callers, - )?; - if witness != expected { - bail!("activation witness does not bind the complete loaded connector policy"); - } - let binding = ConnectorActivationBinding { - activation_witness_sha256: format!("sha256-{}", hex(&Sha256::digest(&witness_bytes))), - }; - binding.validate()?; - Ok(binding) -} - -fn activation_expectation( - activation_id: String, - config_path: &Path, - config_bytes: &[u8], - config: &crate::daemon::CodexDaemonConfig, - callers: &[crate::daemon::LoadedCallerKeyBinding], -) -> Result { - require_deployment_id(&activation_id) - .context("activation witness launch identity is malformed")?; - config - .validate() - .map_err(|error| anyhow!("loaded connector policy is invalid: {error}"))?; - if config.callers.len() != callers.len() { - bail!("loaded connector caller material does not match typed policy"); - } - let mut expected_callers = Vec::with_capacity(callers.len()); - for (policy, loaded) in config.callers.iter().zip(callers) { - if policy.caller_runtime_id != loaded.caller_runtime_id - || policy.connection_key_file != loaded.connection_key_file - || policy.connection_key_epoch != loaded.connection_key_epoch - { - bail!("loaded connector caller material does not match typed policy"); - } - expected_callers.push(ActivationCallerWitness { - runtime_id: policy.caller_runtime_id.clone(), - connection_key_path: absolute_utf8_path( - &policy.connection_key_file, - "connector caller key", - )?, - connection_key_sha256: hex(&loaded.raw_file_sha256), - connection_key_epoch: policy.connection_key_epoch, - allowed_models: policy.allowed_models.clone(), - max_concurrent_requests: policy.max_concurrent_requests, - max_payload_bytes: policy.max_payload_bytes, - max_output_tokens: policy.max_output_tokens, - }); - } - Ok(ActivationWitness { - activation_id, - config_path: absolute_utf8_path(config_path, "connector configuration")?, - config_sha256: hex(&Sha256::digest(config_bytes)), - config_epoch: config.epoch, - bind: config.bind.clone(), - codex_executable_path: absolute_utf8_path(&config.codex_executable, "Codex executable")?, - codex_executable_sha256: hex(&config.codex_executable_sha256), - codex_home_path: absolute_utf8_path(&config.codex_home, "Codex home")?, - replay_store_path: absolute_utf8_path(&config.replay_store, "connector replay store")?, - max_frame_bytes: config.max_frame_bytes, - max_connections: config.max_connections, - socket_timeout_ms: config.socket_timeout_ms, - max_expiry_skew_ms: config.max_expiry_skew_ms, - callers: expected_callers, - }) -} - -fn absolute_utf8_path(path: &Path, label: &str) -> Result { - if !path.is_absolute() { - bail!("{label} path is not absolute"); - } - path.to_str() - .map(str::to_owned) - .ok_or_else(|| anyhow!("{label} path is not UTF-8")) -} - -fn release_binding_from_artifacts( - connector_executable: &Path, - codex_executable: &Path, - configured_codex_sha256: &[u8; 32], - compiled_commit: &str, - deployment_id: &str, - source_toolchain: &ToolchainManifest, -) -> Result { - require_deployment_id(deployment_id)?; - if !is_lower_hex(compiled_commit, 40) { - bail!("compiled connector source commit is unavailable or malformed"); - } - let connector_executable = canonical_regular_file(connector_executable, "connector binary")?; - let release_root = connector_executable - .parent() - .ok_or_else(|| anyhow!("active connector has no release directory"))?; - if connector_executable - .file_name() - .and_then(|name| name.to_str()) - != Some("codex-connector") - { - bail!("active connector executable has an unexpected name"); - } - let expected_codex_path = release_root.join("codex"); - if codex_executable != expected_codex_path.as_path() { - bail!("configured Codex binary is not the active release's direct Codex artifact"); - } - let expected_codex = canonical_regular_file(&expected_codex_path, "Codex binary")?; - let configured_codex = canonical_regular_file(codex_executable, "configured Codex binary")?; - if configured_codex != expected_codex { - bail!("configured Codex binary is outside the active connector release"); - } - let witness_path = release_root.join("DEPLOYMENT"); - let witness_bytes = std::fs::read(&witness_path) - .with_context(|| format!("reading release witness {}", witness_path.display()))?; - let witness = parse_deployment_witness(&witness_bytes)?; - if witness.source_commit != compiled_commit { - bail!("release witness does not bind the compiled connector source commit"); - } - if witness.codex_package_url != source_toolchain.package_url - || witness.codex_package_sha256 != source_toolchain.package_sha256 - || witness.codex_binary_sha256 != source_toolchain.binary_sha256 - { - bail!("release witness does not bind the source-owned Codex toolchain manifest"); - } - let release_directory = release_root - .file_name() - .and_then(|name| name.to_str()) - .ok_or_else(|| anyhow!("connector release directory is not UTF-8"))?; - if witness.release_id != release_directory - || witness.release_id - != format!( - "{}-{}", - witness.source_commit, - &witness.codex_binary_sha256[..12] - ) - { - bail!("release witness does not bind its release directory"); - } - let connector_sha256 = sha256_file(&connector_executable)?; - if witness.connector_binary_sha256 != connector_sha256 { - bail!("release witness does not bind the active connector executable"); - } - let codex_sha256 = sha256_file(&configured_codex)?; - if witness.codex_binary_sha256 != codex_sha256 - || witness.codex_binary_sha256 != hex(configured_codex_sha256) - { - bail!("release witness does not bind the active configured Codex executable"); - } - let release = ConnectorReleaseBinding { - release_id: witness.release_id, - release_witness_sha256: format!("sha256-{}", hex(&Sha256::digest(&witness_bytes))), - source_commit: witness.source_commit, - deployment_id: deployment_id.to_owned(), - }; - release.validate()?; - Ok(release) -} - -fn parse_deployment_witness(bytes: &[u8]) -> Result { - let text = std::str::from_utf8(bytes).context("release witness is not UTF-8")?; - let body = text - .strip_suffix('\n') - .ok_or_else(|| anyhow!("release witness has no canonical final newline"))?; - if body.contains('\r') { - bail!("release witness contains noncanonical line endings"); - } - let lines = body.split('\n').collect::>(); - let [ - schema, - release_id, - source_commit, - connector_binary_sha256, - codex_package_url, - codex_package_sha256, - codex_binary_sha256, - ] = lines.as_slice() - else { - bail!("release witness has an unexpected field set"); - }; - if exact_manifest_value(schema, "schema_version")? != DEPLOYMENT_SCHEMA { - bail!("release witness schema is not admitted"); - } - let witness = DeploymentWitness { - release_id: exact_manifest_value(release_id, "release_id")?.to_owned(), - source_commit: exact_manifest_value(source_commit, "source_commit")?.to_owned(), - connector_binary_sha256: exact_manifest_value(connector_binary_sha256, "binary_sha256")? - .to_owned(), - codex_package_url: exact_manifest_value(codex_package_url, "codex_package_url")?.to_owned(), - codex_package_sha256: exact_manifest_value(codex_package_sha256, "codex_package_sha256")? - .to_owned(), - codex_binary_sha256: exact_manifest_value(codex_binary_sha256, "codex_binary_sha256")? - .to_owned(), - }; - require_id(&witness.release_id, "release id")?; - if !is_lower_hex(&witness.source_commit, 40) - || !is_lower_hex(&witness.connector_binary_sha256, 64) - { - bail!("release witness contains malformed artifact identity"); - } - validate_toolchain_artifacts( - &witness.codex_package_url, - &witness.codex_package_sha256, - &witness.codex_binary_sha256, - )?; - Ok(witness) -} - -fn parse_activation_witness(bytes: &[u8]) -> Result { - let text = std::str::from_utf8(bytes).context("activation witness is not UTF-8")?; - let body = text - .strip_suffix('\n') - .ok_or_else(|| anyhow!("activation witness has no canonical final newline"))?; - if body.contains('\r') { - bail!("activation witness contains noncanonical line endings"); - } - let lines = body.split('\n').collect::>(); - let mut cursor = 0; - if next_manifest_value(&lines, &mut cursor, "schema_version")? != ACTIVATION_SCHEMA { - bail!("activation witness schema is not admitted"); - } - let activation_id = next_manifest_value(&lines, &mut cursor, "activation_id")?.to_owned(); - let config_path = next_manifest_value(&lines, &mut cursor, "config_path")?.to_owned(); - let config_sha256 = next_manifest_value(&lines, &mut cursor, "config_sha256")?.to_owned(); - let config_epoch = - parse_canonical_u32(next_manifest_value(&lines, &mut cursor, "config_epoch")?)?; - let bind = next_manifest_value(&lines, &mut cursor, "bind")?.to_owned(); - let codex_executable_path = - next_manifest_value(&lines, &mut cursor, "codex_executable_path")?.to_owned(); - let codex_executable_sha256 = - next_manifest_value(&lines, &mut cursor, "codex_executable_sha256")?.to_owned(); - let codex_home_path = next_manifest_value(&lines, &mut cursor, "codex_home_path")?.to_owned(); - let replay_store_path = - next_manifest_value(&lines, &mut cursor, "replay_store_path")?.to_owned(); - let max_frame_bytes = - parse_canonical_usize(next_manifest_value(&lines, &mut cursor, "max_frame_bytes")?)?; - let max_connections = - parse_canonical_usize(next_manifest_value(&lines, &mut cursor, "max_connections")?)?; - let socket_timeout_ms = parse_canonical_u64(next_manifest_value( - &lines, - &mut cursor, - "socket_timeout_ms", - )?)?; - let max_expiry_skew_ms = parse_canonical_u64(next_manifest_value( - &lines, - &mut cursor, - "max_expiry_skew_ms", - )?)?; - let caller_count = - parse_canonical_usize(next_manifest_value(&lines, &mut cursor, "caller_count")?)?; - require_deployment_id(&activation_id) - .context("activation witness launch identity is malformed")?; - if config_epoch == 0 - || max_frame_bytes == 0 - || max_connections == 0 - || socket_timeout_ms == 0 - || max_expiry_skew_ms == 0 - || caller_count == 0 - || caller_count > lines.len() - || !Path::new(&config_path).is_absolute() - || !Path::new(&codex_executable_path).is_absolute() - || !Path::new(&codex_home_path).is_absolute() - || !Path::new(&replay_store_path).is_absolute() - || !is_lower_hex(&config_sha256, 64) - || !is_lower_hex(&codex_executable_sha256, 64) - { - bail!("activation witness configuration identity is malformed"); - } - let mut callers = Vec::with_capacity(caller_count); - for index in 0..caller_count { - let runtime_id = - next_manifest_value(&lines, &mut cursor, &format!("caller.{index}.runtime_id"))? - .to_owned(); - let connection_key_path = next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.connection_key_path"), - )? - .to_owned(); - let connection_key_sha256 = next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.connection_key_sha256"), - )? - .to_owned(); - let connection_key_epoch = parse_canonical_u32(next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.connection_key_epoch"), - )?)?; - let allowed_model_count = parse_canonical_usize(next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.allowed_model_count"), - )?)?; - if allowed_model_count == 0 || allowed_model_count > lines.len().saturating_sub(cursor) { - bail!("activation witness caller model set is malformed"); - } - let mut allowed_models = Vec::with_capacity(allowed_model_count); - for model_index in 0..allowed_model_count { - let model = next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.allowed_model.{model_index}"), - )? - .to_owned(); - require_id(&model, "activation caller model")?; - allowed_models.push(model); - } - let max_concurrent_requests = parse_canonical_usize(next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.max_concurrent_requests"), - )?)?; - let max_payload_bytes = parse_canonical_usize(next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.max_payload_bytes"), - )?)?; - let max_output_tokens = parse_canonical_u32(next_manifest_value( - &lines, - &mut cursor, - &format!("caller.{index}.max_output_tokens"), - )?)?; - require_id(&runtime_id, "activation caller runtime")?; - if connection_key_epoch == 0 - || max_concurrent_requests == 0 - || max_payload_bytes == 0 - || max_output_tokens == 0 - || !Path::new(&connection_key_path).is_absolute() - || !is_lower_hex(&connection_key_sha256, 64) - { - bail!("activation witness caller identity is malformed"); - } - callers.push(ActivationCallerWitness { - runtime_id, - connection_key_path, - connection_key_sha256, - connection_key_epoch, - allowed_models, - max_concurrent_requests, - max_payload_bytes, - max_output_tokens, - }); - } - if cursor != lines.len() { - bail!("activation witness has an unexpected field set"); - } - Ok(ActivationWitness { - activation_id, - config_path, - config_sha256, - config_epoch, - bind, - codex_executable_path, - codex_executable_sha256, - codex_home_path, - replay_store_path, - max_frame_bytes, - max_connections, - socket_timeout_ms, - max_expiry_skew_ms, - callers, - }) -} - -fn next_manifest_value<'a>( - lines: &'a [&'a str], - cursor: &mut usize, - name: &str, -) -> Result<&'a str> { - let line = lines - .get(*cursor) - .ok_or_else(|| anyhow!("manifest field {name} is absent"))?; - *cursor += 1; - exact_manifest_value(line, name) -} - -fn parse_canonical_u32(value: &str) -> Result { - let parsed = value - .parse::() - .context("activation witness integer is malformed")?; - if parsed.to_string() != value { - bail!("activation witness integer is noncanonical"); - } - Ok(parsed) -} - -fn parse_canonical_u64(value: &str) -> Result { - let parsed = value - .parse::() - .context("activation witness integer is malformed")?; - if parsed.to_string() != value { - bail!("activation witness integer is noncanonical"); - } - Ok(parsed) -} - -fn parse_canonical_usize(value: &str) -> Result { - let parsed = value - .parse::() - .context("activation witness integer is malformed")?; - if parsed.to_string() != value { - bail!("activation witness integer is noncanonical"); - } - Ok(parsed) -} - -fn current_runtime_process_instance() -> Result { - let process_id = std::process::id(); - if process_id == 0 { - bail!("current connector process id is invalid"); - } - let stat = std::fs::read_to_string("/proc/self/stat") - .context("reading current connector /proc starttime")?; - Ok(RuntimeProcessInstanceBinding { - process_id, - starttime_ticks: parse_proc_stat_starttime(&stat, process_id)?, - }) -} - -fn parse_proc_stat_starttime(stat: &str, expected_process_id: u32) -> Result { - let stat = stat.strip_suffix('\n').unwrap_or(stat); - if stat.contains('\r') || stat.contains('\n') { - bail!("current connector /proc stat has noncanonical line endings"); - } - let prefix = format!("{expected_process_id} ("); - if !stat.starts_with(&prefix) { - bail!("current connector /proc stat has the wrong process id"); - } - let command_end = stat - .rfind(") ") - .context("current connector /proc stat has no command terminator")?; - let fields = stat[command_end + 2..] - .split_ascii_whitespace() - .collect::>(); - if fields.len() <= 19 || fields[0].len() != 1 { - bail!("current connector /proc stat is missing starttime"); - } - let value = fields[19]; - let starttime_ticks = value - .parse::() - .context("current connector /proc starttime is malformed")?; - if starttime_ticks == 0 || starttime_ticks.to_string() != value { - bail!("current connector /proc starttime is noncanonical"); - } - Ok(starttime_ticks) -} - -fn traffic_admission_expectation( - release: &ConnectorReleaseBinding, - activation: &ConnectorActivationBinding, - published: &PublishedHealthStatementIdentity, -) -> Result { - release.validate()?; - activation.validate()?; - require_id(&published.daemon_id, "traffic admission daemon id")?; - require_id( - &published.publisher_incarnation_id, - "traffic admission publisher incarnation", - )?; - require_id( - &published.signer_identity_id, - "traffic admission signer identity", - )?; - if published.daemon_id != TRAFFIC_ADMISSION_KEY - || published.publisher_sequence == 0 - || !published - .signed_health_sha256 - .strip_prefix("sha256-") - .is_some_and(|digest| is_lower_hex(digest, 64)) - { - bail!("signed health digest for traffic admission is malformed"); - } - let process = current_runtime_process_instance()?; - let expected = TrafficAdmissionConsumerProjection { - schema_version: TRAFFIC_ADMISSION_SCHEMA.into(), - daemon_id: published.daemon_id.clone(), - release_id: release.release_id.clone(), - release_witness_sha256: release.release_witness_sha256.clone(), - source_commit: release.source_commit.clone(), - deployment_id: release.deployment_id.clone(), - activation_witness_sha256: activation.activation_witness_sha256.clone(), - signed_health_sha256: published.signed_health_sha256.clone(), - publisher_incarnation_id: published.publisher_incarnation_id.clone(), - publisher_sequence: published.publisher_sequence, - signer_identity_id: published.signer_identity_id.clone(), - runtime_process_id: process.process_id, - runtime_process_starttime_ticks: process.starttime_ticks, - }; - validate_traffic_admission_projection(&expected)?; - Ok(expected) -} - -fn decode_traffic_admission_payload(bytes: &[u8]) -> Result { - let admission: TrafficAdmissionConsumerProjection = - rmp_serde::from_slice(bytes).context("decoding typed connector traffic admission")?; - if rmp_serde::to_vec(&admission)? != bytes { - bail!("connector traffic admission payload is not canonical positional MessagePack"); - } - validate_traffic_admission_projection(&admission)?; - Ok(admission) -} - -fn validate_traffic_admission_projection( - admission: &TrafficAdmissionConsumerProjection, -) -> Result<()> { - if admission.schema_version != TRAFFIC_ADMISSION_SCHEMA - || admission.daemon_id != TRAFFIC_ADMISSION_KEY - || admission.publisher_sequence == 0 - || admission.runtime_process_id == 0 - || admission.runtime_process_starttime_ticks == 0 - { - bail!("connector traffic admission typed identity is invalid"); - } - require_id(&admission.release_id, "traffic admission release id")?; - require_deployment_id(&admission.deployment_id)?; - require_id( - &admission.signer_identity_id, - "traffic admission signer identity", - )?; - uuid::Uuid::parse_str(&admission.publisher_incarnation_id) - .context("traffic admission publisher incarnation is malformed")?; - if !admission - .release_witness_sha256 - .strip_prefix("sha256-") - .is_some_and(|digest| is_lower_hex(digest, 64)) - || !is_lower_hex(&admission.source_commit, 40) - || !admission - .activation_witness_sha256 - .strip_prefix("sha256-") - .is_some_and(|digest| is_lower_hex(digest, 64)) - || !admission - .signed_health_sha256 - .strip_prefix("sha256-") - .is_some_and(|digest| is_lower_hex(digest, 64)) - { - bail!("connector traffic admission authority identity is malformed"); - } - Ok(()) -} - -fn require_exact_traffic_admission( - admitted: &TrafficAdmissionConsumerProjection, - expected: &TrafficAdmissionConsumerProjection, -) -> Result<()> { - if admitted != expected { - bail!("root traffic admission does not match the exact startup statement"); - } - Ok(()) -} - -fn parse_toolchain_manifest(text: &str) -> Result { - let body = text - .strip_suffix('\n') - .ok_or_else(|| anyhow!("source toolchain manifest has no canonical final newline"))?; - if body.contains('\r') { - bail!("source toolchain manifest contains noncanonical line endings"); - } - let lines = body.split('\n').collect::>(); - let [schema, package_url, package_sha256, binary_sha256] = lines.as_slice() else { - bail!("source toolchain manifest has an unexpected field set"); - }; - if exact_manifest_value(schema, "schema_version")? != TOOLCHAIN_SCHEMA { - bail!("source toolchain manifest schema is not admitted"); - } - let manifest = ToolchainManifest { - package_url: exact_manifest_value(package_url, "package_url")?.to_owned(), - package_sha256: exact_manifest_value(package_sha256, "package_sha256")?.to_owned(), - binary_sha256: exact_manifest_value(binary_sha256, "binary_sha256")?.to_owned(), - }; - validate_toolchain_artifacts( - &manifest.package_url, - &manifest.package_sha256, - &manifest.binary_sha256, - )?; - Ok(manifest) -} - -fn validate_toolchain_artifacts( - package_url: &str, - package_sha256: &str, - binary_sha256: &str, -) -> Result<()> { - if !is_lower_hex(package_sha256, 64) - || !is_lower_hex(binary_sha256, 64) - || !package_url.starts_with("https://registry.npmjs.org/@openai/codex/-/codex-") - || !package_url.ends_with("-linux-x64.tgz") - { - bail!("Codex toolchain artifact identity is malformed"); - } - Ok(()) -} - -fn exact_manifest_value<'a>(line: &'a str, name: &str) -> Result<&'a str> { - let value = line - .strip_prefix(&format!("{name}=")) - .ok_or_else(|| anyhow!("manifest field {name} is misplaced or absent"))?; - if value.is_empty() || value.trim() != value || value.chars().any(char::is_control) { - bail!("manifest field {name} is malformed"); - } - Ok(value) -} - -fn required_runtime_deployment_id() -> Result { - let deployment_id = std::env::var("CODEX_CONNECTOR_IDUNN_DEPLOYMENT_REQUEST_ID") - .context("CODEX_CONNECTOR_IDUNN_DEPLOYMENT_REQUEST_ID is required for signed health")?; - require_deployment_id(&deployment_id)?; - Ok(deployment_id) -} - -fn require_deployment_id(value: &str) -> Result<()> { - if value.is_empty() - || value.len() > 256 - || !value - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b':' | b'-')) - { - bail!("Idunn deployment request identity is malformed"); - } - Ok(()) -} - -fn canonical_regular_file(path: &Path, label: &str) -> Result { - let path = std::fs::canonicalize(path) - .with_context(|| format!("locating {label} {}", path.display()))?; - if !std::fs::metadata(&path)?.is_file() { - bail!("{label} is not a regular file"); - } - Ok(path) -} - -fn require_root_sealed(path: &Path, label: &str, directory: bool) -> Result<()> { - let metadata = std::fs::symlink_metadata(path) - .with_context(|| format!("inspecting {label} {}", path.display()))?; - let wrong_kind = if directory { - !metadata.is_dir() - } else { - !metadata.is_file() - }; - if metadata.file_type().is_symlink() - || wrong_kind - || metadata.uid() != 0 - || metadata.mode() & 0o222 != 0 - { - bail!("{label} is not a root-owned immutable direct artifact"); - } - Ok(()) -} - -fn require_root_controlled_runtime_file(path: &Path, label: &str) -> Result<()> { - if !path.is_absolute() { - bail!("{label} path is not absolute"); - } - - let mut current = Some(path); - let mut leaf = true; - while let Some(component) = current { - let metadata = std::fs::symlink_metadata(component) - .with_context(|| format!("inspecting {label} path {}", component.display()))?; - if metadata.file_type().is_symlink() - || (leaf && !metadata.is_file()) - || (!leaf && !metadata.is_dir()) - || metadata.uid() != 0 - || metadata.mode() & 0o022 != 0 - { - bail!("{label} path is not directly root-controlled"); - } - - leaf = false; - current = component.parent().filter(|parent| *parent != component); - } - - Ok(()) -} - -fn sibling_lock_path(path: &Path) -> Result { - let mut name = path - .file_name() - .ok_or_else(|| anyhow!("runtime capability path has no file name"))? - .to_os_string(); - name.push(".lock"); - Ok(path.with_file_name(name)) -} - -fn require_runtime_capability_file(path: &Path, label: &str) -> Result<()> { - require_root_controlled_runtime_file(path, label)?; - let parent = path - .parent() - .ok_or_else(|| anyhow!("{label} has no parent directory"))?; - let parent_metadata = std::fs::symlink_metadata(parent) - .with_context(|| format!("inspecting {label} parent {}", parent.display()))?; - let leaf_metadata = std::fs::symlink_metadata(path) - .with_context(|| format!("inspecting {label} {}", path.display()))?; - if parent_metadata.uid() != 0 - || parent_metadata.mode() & 0o7777 != 0o750 - || leaf_metadata.uid() != 0 - || leaf_metadata.gid() != parent_metadata.gid() - || leaf_metadata.mode() & 0o7777 != 0o640 - { - bail!("{label} custody is not the exact root-to-runtime-group read boundary"); - } - Ok(()) -} - -fn sha256_file(path: &Path) -> Result { - Ok(hex(&Sha256::digest( - std::fs::read(path).with_context(|| format!("hashing {}", path.display()))?, - ))) -} - -fn open_identity(path: &Path) -> Result { - let binding = machine_binding()?; - open_identity_with_binding(path, &binding) -} - -fn open_identity_with_binding(path: &Path, binding: &str) -> Result { - let entries = SingleFileMessagePackBackingStore::new(path).pull_all()?; - let [envelope] = entries.as_slice() else { - bail!("provider health identity store must contain exactly one record"); - }; - if envelope.key != IDENTITY_KEY - || envelope.r#type != IDENTITY_TYPE - || envelope.schema_id.as_deref() != Some(IDENTITY_SCHEMA) - { - bail!("provider health identity store has the wrong type"); - } - let entry: ProviderHealthIdentity = rmp_serde::from_slice(&envelope.payload)?; - validate_identity(&entry)?; - if entry.protector_binding != binding { - bail!("provider health identity belongs to another machine"); - } - let seed: [u8; 32] = mask_seed(&entry.protected_private_seed, binding) - .try_into() - .map_err(|_| anyhow!("provider health identity seed has invalid length"))?; - let key = SigningKey::from_bytes(&seed); - if key.verifying_key().to_bytes().as_slice() != entry.public_key.as_slice() { - bail!("provider health identity seed does not match its public key"); - } - Ok(ProviderHealthSigner { entry, key }) -} - -fn validate_identity(entry: &ProviderHealthIdentity) -> Result<()> { - if entry.schema_version != IDENTITY_SCHEMA - || entry.public_key.len() != 32 - || entry.protected_private_seed.len() != 32 - || entry.enrollment_nonce.len() != 32 - || entry.protector_kind != "linux_file_mode_machine_id_binding" - || entry.protector_version != "v1" - || entry.assurance != "os_installation_file_bound_cloneable_baseline" - || entry.identity_id != identity_id(&entry.public_key) - { - bail!("provider health identity is invalid"); - } - chrono::DateTime::parse_from_rfc3339(&entry.created_at)?; - Ok(()) -} - -fn machine_binding() -> Result { - let raw = std::fs::read_to_string("/etc/machine-id") - .or_else(|_| std::fs::read_to_string("/var/lib/dbus/machine-id"))?; - let id = raw.trim(); - if id.is_empty() { - bail!("Linux machine-id is empty"); - } - Ok(format!( - "{PROTECTOR_CONTEXT}:machine-id-sha256:{}", - hex(&Sha256::digest(id.as_bytes())) - )) -} - -fn mask_seed(seed: &[u8], binding: &str) -> Vec { - let mask = Sha256::digest( - [ - b"gamecult-linux-service-seed-v1\0".as_slice(), - PROTECTOR_CONTEXT.as_bytes(), - binding.as_bytes(), - ] - .concat(), + let provider_endpoint = dependency + .provider_endpoint + .as_deref() + .context("Expected Odin provider has no RUDP endpoint")?; + let expected_endpoint = parse_expected_rudp_endpoint(provider_endpoint)?; + ensure!( + expected_endpoint == endpoint, + "CultNet RUDP endpoint differs from Expected Odin provider endpoint" ); - seed.iter() - .zip(mask) - .map(|(left, right)| left ^ right) - .collect() + Ok(()) } -fn identity_id(public_key: &[u8]) -> String { - hex(&Sha256::digest([ID_DOMAIN, public_key].concat())) +fn parse_expected_rudp_endpoint(value: &str) -> Result { + if let Some(candidate) = value + .strip_prefix("rudp://") + .or_else(|| value.strip_prefix("udp://")) + { + return candidate + .parse() + .context("parsing Expected Odin RUDP endpoint"); + } + ensure!( + !value.contains("://"), + "Expected Odin provider endpoint uses an unsupported transport" + ); + value + .parse() + .context("parsing bare Expected Odin RUDP endpoint") } -fn unsigned_record(record: &SignedDaemonHealth) -> Result> { - let mut unsigned = record.clone(); - unsigned.signature.clear(); - uuid::Uuid::parse_str(&unsigned.publisher_incarnation_id)?; - Ok(rmp_serde::to_vec(&unsigned)?) +fn require_expected_capability( + expected: &IdunnExpectedIncarnationRecord, + actual: &GameCultRuntimeCapability, +) -> Result<()> { + let required = expected + .capabilities + .iter() + .find(|required| { + required.capability == actual.capability + && required.schema == actual.schema + && required.compatibility == actual.compatibility + }) + .context("Expected projection does not admit Connector's runtime capability")?; + ensure!( + actual.capacity >= required.minimum_capacity, + "Connector capacity is below Expected" + ); + Ok(()) } -fn signing_message(payload: &[u8]) -> Vec { - let mut out = - Vec::with_capacity(SIGNATURE_DOMAIN.len() + SIGNATURE_PURPOSE.len() + payload.len() + 16); - out.extend_from_slice(SIGNATURE_DOMAIN); - out.extend_from_slice(&(SIGNATURE_PURPOSE.len() as u64).to_be_bytes()); - out.extend_from_slice(SIGNATURE_PURPOSE); - out.extend_from_slice(&(payload.len() as u64).to_be_bytes()); - out.extend_from_slice(payload); - out +fn require_connector_expected_contract( + expected: &IdunnExpectedIncarnationRecord, + actual: &GameCultRuntimeCapability, +) -> Result<()> { + require_expected_capability(expected, actual)?; + ensure!( + expected.target == CONNECTOR_TARGET, + "Connector Expected projection names a different target" + ); + ensure!( + expected.health_contract == CONNECTOR_HEALTH_CONTRACT, + "Connector Expected projection names an unimplemented health contract" + ); + ensure!( + expected.state_schema_generation.as_deref() == Some(CONNECTOR_STATE_SCHEMA_GENERATION) + && expected.state_contract_sha256.as_deref() == Some(CONNECTOR_STATE_CONTRACT_SHA256) + && expected.write_lease_required, + "Connector Expected projection does not match its compiled state contract" + ); + Ok(()) } -fn publish(endpoint: SocketAddr, runtime_id: &str, record: &SignedDaemonHealth) -> Result { - let (payload, signed_health_sha256) = canonical_signed_health_payload(record)?; +fn load_process_write_lease( + path: &Path, + expected: &IdunnExpectedIncarnationRecord, + expected_sha256: &str, + activation: &IdunnRuntimeActivationRecord, + activation_sha256: &str, + warming_presence_sha256: &str, +) -> Result> { + match std::fs::symlink_metadata(path) { + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + } + require_root_controlled_file(path, "process-write-lease record")?; + require_root_controlled_file(&sibling_lock_path(path)?, "process-write-lease lock")?; + SingleFileMessagePackBackingStore::new(path).with_read_only_shared_snapshot(|entries| { + let [envelope] = entries.as_slice() else { + if entries.is_empty() { + return Ok(None); + } + bail!("process-write-lease store is ambiguous"); + }; + ensure!( + envelope.r#type == IdunnProcessWriteLeaseRecord::TYPE + && envelope.schema_id.as_deref() == Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA), + "process-write-lease store has the wrong typed envelope" + ); + let lease = IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload)?; + ensure!( + envelope.key == lease.target, + "process-write-lease key is not its target" + ); + exact_process_write_lease_sha256( + &lease, + expected, + expected_sha256, + activation, + activation_sha256, + warming_presence_sha256, + ) + .map(Some) + }) +} + +fn acquire_process_write_lease_guard( + path: &Path, + expected: &IdunnExpectedIncarnationRecord, + expected_sha256: &str, + activation: &IdunnRuntimeActivationRecord, + activation_sha256: &str, + warming_presence_sha256: &str, +) -> Result> { + match std::fs::symlink_metadata(path) { + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + } + require_root_controlled_file(path, "process-write-lease record")?; + let lock_path = sibling_lock_path(path)?; + require_root_controlled_file(&lock_path, "process-write-lease lock")?; + let lock = OpenOptions::new() + .read(true) + .open(&lock_path) + .with_context(|| format!("opening process-write-lease lock {}", lock_path.display()))?; + fs2::FileExt::lock_shared(&lock) + .with_context(|| format!("locking process-write-lease lock {}", lock_path.display()))?; + + // The first shared lock remains held while this exact snapshot is decoded. + // The backing-store read takes its normal nested shared lock so Connector + // continues to use CultCache's canonical snapshot decoder. + let lease_sha256 = load_process_write_lease( + path, + expected, + expected_sha256, + activation, + activation_sha256, + warming_presence_sha256, + )?; + let Some(lease_sha256) = lease_sha256 else { + return Ok(None); + }; + Ok(Some(ProcessWriteLeaseGuard { + held: Arc::new(HeldProcessWriteLease { + _lock: lock, + path: path.to_path_buf(), + expected: expected.clone(), + expected_sha256: expected_sha256.into(), + activation: activation.clone(), + activation_sha256: activation_sha256.into(), + warming_presence_sha256: warming_presence_sha256.into(), + lease_sha256, + }), + })) +} + +fn exact_process_write_lease_sha256( + lease: &IdunnProcessWriteLeaseRecord, + expected: &IdunnExpectedIncarnationRecord, + expected_sha256: &str, + activation: &IdunnRuntimeActivationRecord, + activation_sha256: &str, + warming_presence_sha256: &str, +) -> Result { + lease.validate()?; + ensure!( + lease.target == expected.target + && lease.expected_projection_sha256 == expected_sha256 + && lease.plan_id == expected.plan_id + && lease.incarnation_id == expected.incarnation_id + && lease.sealed_release_id == expected.sealed_release_id + && lease.activation_witness_sha256 == activation_sha256 + && lease.state_schema_generation + == expected + .state_schema_generation + .as_deref() + .context("Expected has no state schema generation")? + && lease.state_contract_sha256 + == expected + .state_contract_sha256 + .as_deref() + .context("Expected has no state contract")? + && lease.runtime_id == expected.runtime_id + && lease.runtime_instance_id == activation.runtime_instance_id + && lease.warming_presence_sha256 == warming_presence_sha256, + "process-write lease does not bind this exact warming incarnation" + ); + lease.canonical_sha256() +} + +fn publish_presence( + endpoint: SocketAddr, + target: &str, + runtime_id: &str, + record: &GameCultRuntimePresenceHealthRecord, + payload: Vec, +) -> Result<()> { let message = CultNetMessage::DocumentPutRaw { message_id: format!( - "codex-connector-health:{}:{}:{}", - record.daemon_id, record.publisher_incarnation_id, record.publisher_sequence + "codex-connector-presence:{}:{}:{}", + target, record.runtime_instance_id, record.publisher_sequence ), document: CultNetRawDocumentRecord { - schema_id: HEALTH_SCHEMA.into(), - record_key: record.daemon_id.clone(), + schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(), + record_key: target.into(), stored_at: chrono::DateTime::from_timestamp_millis( record.observed_at_unix_millis.try_into()?, ) - .context("health observation time is invalid")? + .context("runtime presence observation time is invalid")? .to_rfc3339(), payload_encoding: CultNetRawPayloadEncoding::Messagepack, payload, source_runtime_id: Some(runtime_id.into()), source_agent_id: Some(record.signer_identity_id.clone()), - source_role: Some("daemon-health-publisher".into()), + source_role: Some("runtime-presence-health-publisher".into()), tags: Some(vec![RUDP_PROTOCOL_ID.into()]), }, }; @@ -1363,590 +649,592 @@ fn publish(endpoint: SocketAddr, runtime_id: &str, record: &SignedDaemonHealth) "[::]:0" })?; socket.set_read_timeout(Some(Duration::from_millis(100)))?; - let mut transport = CultNetRudpSocketTransportConnection::new( - CultNetRudpSocketTransportOptions::client(runtime_id, socket, endpoint, RUDP_CONNECTION_ID), - )?; + let mut transport = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions::client( + runtime_id, + socket, + endpoint, + ODIN_CULTMESH_DOCUMENT_CATALOG_CONNECTION_ID, + ))?; transport.connect(Vec::new())?; let deadline = Instant::now() + Duration::from_millis(500); while !transport.connected() { let _ = transport.receive_once()?; transport.poll_resends()?; if Instant::now() >= deadline { - bail!("timed out connecting provider health to {endpoint}"); + bail!("timed out connecting runtime presence to {endpoint}"); } } - transport.send( + let receipt = transport.send_reliable( "schema", encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?, )?; - Ok(signed_health_sha256) -} - -fn canonical_signed_health_payload(record: &SignedDaemonHealth) -> Result<(Vec, String)> { - let payload = rmp_serde::to_vec(record)?; - if rmp_serde::from_slice::(&payload)? != *record { - bail!("signed health MessagePack did not round trip"); - } - let signed_health_sha256 = format!("sha256-{}", hex(&Sha256::digest(&payload))); - Ok((payload, signed_health_sha256)) -} - -fn require_id(value: &str, label: &str) -> Result<()> { - if value.is_empty() - || value != value.trim() - || value.len() > 256 - || value.chars().any(char::is_control) - { - bail!("{label} is invalid"); + let ack_deadline = Instant::now() + Duration::from_millis(500); + loop { + match transport.reliable_send_status(&receipt) { + CultNetRudpReliableSendStatus::Acknowledged => break, + CultNetRudpReliableSendStatus::Invalidated => { + bail!("runtime presence send was invalidated before acknowledgement") + } + CultNetRudpReliableSendStatus::Pending => { + let _ = transport.receive_once()?; + transport.poll_resends()?; + if Instant::now() >= ack_deadline { + bail!("timed out awaiting runtime presence acknowledgement from {endpoint}"); + } + } + } } Ok(()) } -fn is_lower_hex(value: &str, length: usize) -> bool { - value.len() == length - && value - .bytes() - .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) +fn require_runtime_bundle(bundle: &Path) -> Result<()> { + ensure!( + bundle.is_absolute(), + "Idunn runtime bundle path is not absolute" + ); + let metadata = std::fs::symlink_metadata(bundle) + .with_context(|| format!("inspecting runtime bundle {}", bundle.display()))?; + ensure!( + metadata.is_dir() + && !metadata.file_type().is_symlink() + && metadata.uid() == 0 + && metadata.mode() & 0o222 == 0, + "Idunn runtime bundle is not a root-owned read-only directory" + ); + let parent = bundle.parent().context("runtime bundle has no parent")?; + require_root_controlled_directory_chain(parent, "Idunn runtime bundle parent") } -fn hex(bytes: &[u8]) -> String { - const DIGITS: &[u8; 16] = b"0123456789abcdef"; - let mut output = String::with_capacity(bytes.len() * 2); - for byte in bytes { - output.push(DIGITS[(byte >> 4) as usize] as char); - output.push(DIGITS[(byte & 15) as usize] as char); +fn take_runtime_signer_descriptors_from_environment() -> Result { + let listen_pid = required_environment(SYSTEMD_LISTEN_PID_ENVIRONMENT)?; + let listen_fds = required_environment(SYSTEMD_LISTEN_FDS_ENVIRONMENT)?; + let listen_fd_names = required_environment(SYSTEMD_LISTEN_FDNAMES_ENVIRONMENT)?; + let (activation_fd, provider_fd) = parse_runtime_signer_descriptor_contract( + &listen_pid, + &listen_fds, + &listen_fd_names, + std::process::id(), + )?; + ensure_descriptor_is_open(activation_fd, "runtime activation signer")?; + ensure_descriptor_is_open(provider_fd, "stable provider signer")?; + + // SAFETY: systemd's LISTEN_* contract assigns the two verified-open, + // distinct descriptors starting at SD_LISTEN_FDS_START to this exact PID. + // This is the first descriptor-consuming step in the managed process and + // this function takes their sole ownership. + let activation = unsafe { File::from_raw_fd(activation_fd) }; + // SAFETY: same ownership proof as above; the descriptor indices are + // distinct because the required names are distinct and occur once. + let provider = unsafe { File::from_raw_fd(provider_fd) }; + protect_signer_descriptor(&activation, "runtime activation signer")?; + protect_signer_descriptor(&provider, "stable provider signer")?; + Ok(RuntimeSignerDescriptors { + activation, + provider, + }) +} + +fn parse_runtime_signer_descriptor_contract( + listen_pid: &str, + listen_fds: &str, + listen_fd_names: &str, + process_id: u32, +) -> Result<(RawFd, RawFd)> { + let declared_pid = listen_pid + .parse::() + .context("parsing systemd LISTEN_PID")?; + ensure!( + declared_pid.to_string() == listen_pid && declared_pid == process_id, + "systemd signer descriptors belong to a different process" + ); + let descriptor_count = listen_fds + .parse::() + .context("parsing systemd LISTEN_FDS")?; + ensure!( + descriptor_count.to_string() == listen_fds && descriptor_count == 2, + "Connector requires exactly two systemd signer descriptors" + ); + let names = listen_fd_names.split(':').collect::>(); + ensure!( + names.as_slice() == [ACTIVATION_SIGNER_FD_NAME, PROVIDER_SIGNER_FD_NAME], + "systemd signer descriptor names or order differ from Idunn's contract" + ); + Ok((SYSTEMD_LISTEN_FDS_START, SYSTEMD_LISTEN_FDS_START + 1)) +} + +fn ensure_descriptor_is_open(descriptor: RawFd, label: &str) -> Result<()> { + // SAFETY: F_GETFD accepts an integer descriptor and performs no pointer + // dereference. A failure leaves the descriptor untouched. + if unsafe { libc::fcntl(descriptor, libc::F_GETFD) } == -1 { + return Err(std::io::Error::last_os_error()) + .with_context(|| format!("opening inherited {label} descriptor")); } - output + Ok(()) +} + +fn protect_signer_descriptor(file: &File, label: &str) -> Result<()> { + let metadata = file + .metadata() + .with_context(|| format!("inspecting {label} descriptor"))?; + ensure!( + metadata.is_file() + && metadata.uid() == 0 + && metadata.gid() == 0 + && metadata.mode() & 0o777 == 0o400 + && metadata.nlink() == 1 + && metadata.len() > 0, + "{label} descriptor is not a root-owned 0400 singly-linked regular file" + ); + // SAFETY: F_GETFL observes integer flags on the live descriptor owned by + // `file` and does not transfer ownership. + let status_flags = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_GETFL) }; + if status_flags == -1 { + return Err(std::io::Error::last_os_error()) + .with_context(|| format!("reading {label} access mode")); + } + ensure!( + status_flags & libc::O_ACCMODE == libc::O_RDONLY, + "{label} descriptor is not read-only" + ); + mark_descriptor_close_on_exec(file, label) +} + +fn mark_descriptor_close_on_exec(file: &File, label: &str) -> Result<()> { + let descriptor = file.as_raw_fd(); + // SAFETY: fcntl observes and mutates only the flags of the valid descriptor + // owned by `file`; it does not dereference a pointer or transfer ownership. + let flags = unsafe { libc::fcntl(descriptor, libc::F_GETFD) }; + if flags == -1 { + return Err(std::io::Error::last_os_error()) + .with_context(|| format!("reading {label} descriptor flags")); + } + // Defense in depth: even before the readers consume and close the files, + // an accidentally spawned child cannot inherit either signing authority. + // SAFETY: same valid descriptor and integer-only fcntl contract as above. + if unsafe { libc::fcntl(descriptor, libc::F_SETFD, flags | libc::FD_CLOEXEC) } == -1 { + return Err(std::io::Error::last_os_error()) + .with_context(|| format!("protecting {label} descriptor from child inheritance")); + } + Ok(()) +} + +fn require_root_read_only_file(path: &Path, label: &str) -> Result<()> { + ensure!(path.is_absolute(), "{label} path is not absolute"); + let metadata = std::fs::symlink_metadata(path) + .with_context(|| format!("inspecting {label} {}", path.display()))?; + ensure!( + metadata.is_file() + && !metadata.file_type().is_symlink() + && metadata.uid() == 0 + && metadata.mode() & 0o222 == 0, + "{label} is not a root-owned read-only regular file" + ); + Ok(()) +} + +fn require_root_controlled_file(path: &Path, label: &str) -> Result<()> { + ensure!(path.is_absolute(), "{label} path is not absolute"); + let parent = path + .parent() + .context("root-controlled path has no parent")?; + let parent_metadata = std::fs::symlink_metadata(parent) + .with_context(|| format!("inspecting {label} parent {}", parent.display()))?; + let metadata = std::fs::symlink_metadata(path) + .with_context(|| format!("inspecting {label} {}", path.display()))?; + ensure!( + metadata.is_file() + && !metadata.file_type().is_symlink() + && metadata.uid() == 0 + && metadata.mode() & 0o022 == 0 + && parent_metadata.is_dir() + && !parent_metadata.file_type().is_symlink() + && parent_metadata.uid() == 0 + && parent_metadata.mode() & 0o022 == 0, + "{label} is not a root-controlled regular file" + ); + Ok(()) +} + +fn require_root_controlled_directory_chain(path: &Path, label: &str) -> Result<()> { + ensure!(path.is_absolute(), "{label} path is not absolute"); + for directory in path.ancestors() { + let metadata = std::fs::symlink_metadata(directory) + .with_context(|| format!("inspecting {label} {}", directory.display()))?; + ensure!( + metadata.is_dir() + && !metadata.file_type().is_symlink() + && metadata.uid() == 0 + && metadata.mode() & 0o022 == 0, + "{label} is not entirely root-controlled" + ); + } + Ok(()) +} + +fn sibling_lock_path(path: &Path) -> Result { + let mut name = path + .file_name() + .context("CultCache authority path has no file name")? + .to_os_string(); + name.push(".lock"); + Ok(path.with_file_name(name)) +} + +fn required_environment(name: &'static str) -> Result { + std::env::var(name) + .ok() + .filter(|value| !value.is_empty() && value.trim() == value) + .ok_or_else(|| anyhow!("{name} is required for an Idunn-managed Connector")) +} + +fn unix_millis() -> Result { + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .context("system clock is before the Unix epoch")? + .as_millis(); + millis.try_into().context("Unix time exceeds u64") } #[cfg(test)] mod tests { use super::*; - use ed25519_dalek::{Signature, Verifier, VerifyingKey}; + use cultnet_rs::{ + IdunnExpectedCapability, IdunnRuntimeActivationLaunch, IdunnServiceIdentity, + RuntimePresenceAuthenticationContext, authenticate_runtime_presence_claim, + derive_idunn_runtime_activation_identity_id, enroll_service_identity_at, + verify_runtime_authority, + }; + use std::io::{Cursor, Write}; + use std::process::{Command, Stdio}; - const SOURCE_COMMIT: &str = "0123456789abcdef0123456789abcdef01234567"; - const DEPLOYMENT_ID: &str = "deploy-codex-connector-42"; - - struct ReleaseFixture { - root: PathBuf, - connector: PathBuf, - codex: PathBuf, - witness: PathBuf, - codex_sha256: [u8; 32], - witness_bytes: Vec, - release_id: String, - source_toolchain: ToolchainManifest, - } - - impl ReleaseFixture { - fn new() -> Self { - let root = std::env::temp_dir().join(format!( - "codex-connector-release-{}-{}", - std::process::id(), - uuid::Uuid::new_v4() - )); - let connector_bytes = b"exact connector fixture"; - let codex_bytes = b"exact codex fixture"; - let connector_sha256 = hex(&Sha256::digest(connector_bytes)); - let codex_sha256: [u8; 32] = Sha256::digest(codex_bytes).into(); - let codex_sha256_hex = hex(&codex_sha256); - let release_id = format!("{SOURCE_COMMIT}-{}", &codex_sha256_hex[..12]); - let release = root.join(&release_id); - std::fs::create_dir_all(&release).unwrap(); - let connector = release.join("codex-connector"); - let codex = release.join("codex"); - let witness = release.join("DEPLOYMENT"); - std::fs::write(&connector, connector_bytes).unwrap(); - std::fs::write(&codex, codex_bytes).unwrap(); - let package_url = - "https://registry.npmjs.org/@openai/codex/-/codex-0.150.0-alpha.7-linux-x64.tgz"; - let package_sha256 = "a".repeat(64); - let witness_bytes = format!( - "schema_version={DEPLOYMENT_SCHEMA}\nrelease_id={release_id}\nsource_commit={SOURCE_COMMIT}\nbinary_sha256={connector_sha256}\ncodex_package_url={package_url}\ncodex_package_sha256={package_sha256}\ncodex_binary_sha256={codex_sha256_hex}\n", - ) - .into_bytes(); - std::fs::write(&witness, &witness_bytes).unwrap(); - Self { - root, - connector, - codex, - witness, - codex_sha256, - witness_bytes, - release_id, - source_toolchain: ToolchainManifest { - package_url: package_url.into(), - package_sha256, - binary_sha256: codex_sha256_hex, - }, - } - } - - fn binding(&self) -> Result { - release_binding_from_artifacts( - &self.connector, - &self.codex, - &self.codex_sha256, - SOURCE_COMMIT, - DEPLOYMENT_ID, - &self.source_toolchain, - ) - } - } - - impl Drop for ReleaseFixture { - fn drop(&mut self) { - let _ = std::fs::remove_dir_all(&self.root); - } + fn sha(digit: char) -> String { + format!("sha256-{}", digit.to_string().repeat(64)) } #[test] - fn exact_release_binding_covers_artifact_witness_and_both_binaries() { - let fixture = ReleaseFixture::new(); - let binding = fixture.binding().unwrap(); - assert_eq!(binding.release_id, fixture.release_id); - assert_eq!(binding.source_commit, SOURCE_COMMIT); - assert_eq!(binding.deployment_id, DEPLOYMENT_ID); + fn systemd_signer_descriptor_contract_is_exact_and_process_bound() -> Result<()> { + let pid = 42; assert_eq!( - binding.release_witness_sha256, - format!("sha256-{}", hex(&Sha256::digest(&fixture.witness_bytes))) + parse_runtime_signer_descriptor_contract( + "42", + "2", + &format!("{ACTIVATION_SIGNER_FD_NAME}:{PROVIDER_SIGNER_FD_NAME}"), + pid, + )?, + (3, 4) ); + for (declared_pid, count, names) in [ + ( + "41", + "2", + format!("{ACTIVATION_SIGNER_FD_NAME}:{PROVIDER_SIGNER_FD_NAME}"), + ), + ( + "42", + "2", + format!("{PROVIDER_SIGNER_FD_NAME}:{ACTIVATION_SIGNER_FD_NAME}"), + ), + ( + "42", + "3", + format!("{ACTIVATION_SIGNER_FD_NAME}:{PROVIDER_SIGNER_FD_NAME}:unexpected"), + ), + ("42", "2", "unexpected:also-unexpected".into()), + ( + "42", + "2", + format!("{ACTIVATION_SIGNER_FD_NAME}:{ACTIVATION_SIGNER_FD_NAME}"), + ), + ] { + assert!( + parse_runtime_signer_descriptor_contract(&declared_pid, count, &names, pid) + .is_err() + ); + } + Ok(()) } #[test] - fn activation_binding_covers_exact_loaded_config_and_caller_key_material() { + fn expected_odin_endpoint_is_explicit_rudp_and_exact() -> Result<()> { + let expected: SocketAddr = "127.0.0.1:17871".parse()?; + for value in [ + "rudp://127.0.0.1:17871", + "udp://127.0.0.1:17871", + "127.0.0.1:17871", + ] { + assert_eq!(parse_expected_rudp_endpoint(value)?, expected); + } + for value in [ + "tcp://127.0.0.1:17871", + "http://127.0.0.1:17871", + "odin.internal:17871", + "", + ] { + assert!(parse_expected_rudp_endpoint(value).is_err()); + } + Ok(()) + } + + #[test] + fn signer_descriptors_and_systemd_environment_do_not_reach_a_child() -> Result<()> { let root = std::env::temp_dir().join(format!( - "codex-connector-activation-{}-{}", + "codex-connector-signer-fds-{}-{}", std::process::id(), - uuid::Uuid::new_v4() + unix_millis()? )); - std::fs::create_dir_all(&root).unwrap(); - let config_path = root.join("connector.cc"); - let witness_path = root.join("ACTIVATION"); - let first_key = root.join("ghostlight.key"); - let second_key = root.join("epiphany.key"); - let config_bytes = b"exact CultCache config bytes"; - let first_bytes = b"ghostlight secret\n"; - let second_bytes = b"epiphany secret\n"; - std::fs::write(&config_path, config_bytes).unwrap(); - std::fs::write(&first_key, first_bytes).unwrap(); - std::fs::write(&second_key, second_bytes).unwrap(); - let callers = vec![ - crate::daemon::LoadedCallerKeyBinding { - caller_runtime_id: "ghostlight-dungeon-yggdrasil".into(), - connection_key_file: first_key.clone(), - connection_key_epoch: 3, - raw_file_sha256: Sha256::digest(first_bytes).into(), - }, - crate::daemon::LoadedCallerKeyBinding { - caller_runtime_id: "epiphany-model-runtime".into(), - connection_key_file: second_key.clone(), - connection_key_epoch: 7, - raw_file_sha256: Sha256::digest(second_bytes).into(), - }, - ]; - let config = crate::daemon::CodexDaemonConfig { - epoch: 2, - bind: "127.0.0.1:4103".into(), - codex_executable: root.join("codex"), - codex_executable_sha256: [9; 32], - codex_home: root.join("codex-home"), - max_frame_bytes: 1_048_576, - max_connections: 12, - socket_timeout_ms: 10_000, - max_expiry_skew_ms: 300_000, - callers: vec![ - crate::daemon::CodexCallerConfig { - caller_runtime_id: "ghostlight-dungeon-yggdrasil".into(), - connection_key_file: first_key.clone(), - connection_key_epoch: 3, - allowed_models: vec!["gpt-5.6-luna".into(), "gpt-5.6-terra".into()], - max_concurrent_requests: 4, - max_payload_bytes: 524_288, - max_output_tokens: 16_384, - }, - crate::daemon::CodexCallerConfig { - caller_runtime_id: "epiphany-model-runtime".into(), - connection_key_file: second_key.clone(), - connection_key_epoch: 7, - allowed_models: vec!["gpt-5.6-luna".into()], - max_concurrent_requests: 8, - max_payload_bytes: 262_144, - max_output_tokens: 8_192, - }, - ], - replay_store: root.join("replay.cc"), - }; - let witness_bytes = format!( - "schema_version={ACTIVATION_SCHEMA}\nactivation_id=deploy-codex-connector-42:canonical:00000000-0000-4000-8000-000000000042\nconfig_path={}\nconfig_sha256={}\nconfig_epoch=2\nbind=127.0.0.1:4103\ncodex_executable_path={}\ncodex_executable_sha256={}\ncodex_home_path={}\nreplay_store_path={}\nmax_frame_bytes=1048576\nmax_connections=12\nsocket_timeout_ms=10000\nmax_expiry_skew_ms=300000\ncaller_count=2\ncaller.0.runtime_id=ghostlight-dungeon-yggdrasil\ncaller.0.connection_key_path={}\ncaller.0.connection_key_sha256={}\ncaller.0.connection_key_epoch=3\ncaller.0.allowed_model_count=2\ncaller.0.allowed_model.0=gpt-5.6-luna\ncaller.0.allowed_model.1=gpt-5.6-terra\ncaller.0.max_concurrent_requests=4\ncaller.0.max_payload_bytes=524288\ncaller.0.max_output_tokens=16384\ncaller.1.runtime_id=epiphany-model-runtime\ncaller.1.connection_key_path={}\ncaller.1.connection_key_sha256={}\ncaller.1.connection_key_epoch=7\ncaller.1.allowed_model_count=1\ncaller.1.allowed_model.0=gpt-5.6-luna\ncaller.1.max_concurrent_requests=8\ncaller.1.max_payload_bytes=262144\ncaller.1.max_output_tokens=8192\n", - config_path.display(), - hex(&Sha256::digest(config_bytes)), - config.codex_executable.display(), - hex(&config.codex_executable_sha256), - config.codex_home.display(), - config.replay_store.display(), - first_key.display(), - hex(&Sha256::digest(first_bytes)), - second_key.display(), - hex(&Sha256::digest(second_bytes)), - ) - .into_bytes(); - std::fs::write(&witness_path, &witness_bytes).unwrap(); - let binding = activation_binding_from_material( - &config_path, - config_bytes, - &config, - &callers, - &witness_path, - ) - .unwrap(); - assert_eq!( - binding.activation_witness_sha256, - format!("sha256-{}", hex(&Sha256::digest(&witness_bytes))) - ); - let next_launch = String::from_utf8(witness_bytes.clone()) - .unwrap() - .replace("000000000042", "000000000043") - .into_bytes(); - std::fs::write(&witness_path, &next_launch).unwrap(); - let next_binding = activation_binding_from_material( - &config_path, - config_bytes, - &config, - &callers, - &witness_path, - ) - .unwrap(); - assert_ne!( - next_binding.activation_witness_sha256, binding.activation_witness_sha256, - "a new root-issued launch admission must invalidate stale health" - ); - std::fs::write(&witness_path, &witness_bytes).unwrap(); + std::fs::create_dir_all(&root)?; + let mut activation = File::create(root.join("activation"))?; + activation.write_all(&[7; 32])?; + let mut provider = File::create(root.join("provider"))?; + provider.write_all(&[8; 32])?; + mark_descriptor_close_on_exec(&activation, "activation test signer")?; + mark_descriptor_close_on_exec(&provider, "provider test signer")?; + let activation_fd = activation.as_raw_fd(); + let provider_fd = provider.as_raw_fd(); - for substituted in [ - witness_bytes - .iter() - .copied() - .chain(b"extra=true\n".iter().copied()) - .collect::>(), - String::from_utf8(witness_bytes.clone()) - .unwrap() - .replace(&hex(&Sha256::digest(first_bytes)), &"f".repeat(64)) - .into_bytes(), - ] { - std::fs::write(&witness_path, substituted).unwrap(); - assert!( - activation_binding_from_material( - &config_path, - config_bytes, - &config, - &callers, - &witness_path, - ) - .is_err() - ); - } - std::fs::write(&witness_path, &witness_bytes).unwrap(); - assert!( - activation_binding_from_material( - &config_path, - b"substituted config bytes", - &config, - &callers, - &witness_path, + let mut child = Command::new("/bin/sh"); + child + .arg("-c") + .arg(format!( + "test ! -e /proc/self/fd/{activation_fd} && \ + test ! -e /proc/self/fd/{provider_fd} && \ + test -z \"${{LISTEN_PID+x}}${{LISTEN_FDS+x}}${{LISTEN_FDNAMES+x}}\"" + )) + .env( + SYSTEMD_LISTEN_PID_ENVIRONMENT, + std::process::id().to_string(), ) - .is_err() - ); - let mut substituted_callers = callers.clone(); - substituted_callers[1].connection_key_epoch = 8; - assert!( - activation_binding_from_material( - &config_path, - config_bytes, - &config, - &substituted_callers, - &witness_path, + .env(SYSTEMD_LISTEN_FDS_ENVIRONMENT, "2") + .env( + SYSTEMD_LISTEN_FDNAMES_ENVIRONMENT, + format!("{ACTIVATION_SIGNER_FD_NAME}:{PROVIDER_SIGNER_FD_NAME}"), ) - .is_err() - ); - let mut substituted_config = config.clone(); - substituted_config.max_connections += 1; - assert!( - activation_binding_from_material( - &config_path, - config_bytes, - &substituted_config, - &callers, - &witness_path, - ) - .is_err() - ); - let mut substituted_caller_policy = config; - substituted_caller_policy.callers[0] - .allowed_models - .reverse(); - assert!( - activation_binding_from_material( - &config_path, - config_bytes, - &substituted_caller_policy, - &callers, - &witness_path, - ) - .is_err() - ); - std::fs::remove_dir_all(root).unwrap(); + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + crate::provider_backend::strip_systemd_descriptor_environment(&mut child); + ensure!(child.status()?.success(), "signer authority reached child"); + + drop((activation, provider)); + std::fs::remove_dir_all(root)?; + Ok(()) } - #[test] - fn typed_traffic_admission_binds_the_exact_startup_statement() { - assert_eq!( - fixed_traffic_admission_path(TRAFFIC_ADMISSION_PATH).unwrap(), - PathBuf::from(TRAFFIC_ADMISSION_PATH) - ); - assert!(fixed_traffic_admission_path("/tmp/traffic-admission.cc").is_err()); - let release = ConnectorReleaseBinding { - release_id: format!("{SOURCE_COMMIT}-aaaaaaaaaaaa"), - release_witness_sha256: format!("sha256-{}", "b".repeat(64)), - source_commit: SOURCE_COMMIT.into(), - deployment_id: DEPLOYMENT_ID.into(), - }; - let activation = ConnectorActivationBinding { - activation_witness_sha256: format!("sha256-{}", "c".repeat(64)), - }; - let published = PublishedHealthStatementIdentity { - daemon_id: TRAFFIC_ADMISSION_KEY.into(), - signed_health_sha256: format!("sha256-{}", "d".repeat(64)), - publisher_incarnation_id: "00000000-0000-4000-8000-000000000042".into(), - publisher_sequence: 1, - signer_identity_id: "e".repeat(64), - }; - let expected = traffic_admission_expectation(&release, &activation, &published).unwrap(); - let payload = rmp_serde::to_vec(&expected).unwrap(); - assert_eq!(payload.first().copied(), Some(0x9d)); - assert_eq!(expected.runtime_process_id, std::process::id()); - assert!(expected.runtime_process_starttime_ticks > 0); - assert_eq!( - decode_traffic_admission_payload(&payload).unwrap(), - expected - ); - - let named = rmp_serde::to_vec_named(&expected).unwrap(); - assert!(decode_traffic_admission_payload(&named).is_err()); - for mut substituted in [ - { - let mut value = expected.clone(); - value.signed_health_sha256 = format!("sha256-{}", "f".repeat(64)); - value - }, - { - let mut value = expected.clone(); - value.publisher_incarnation_id = "00000000-0000-4000-8000-000000000043".into(); - value - }, - { - let mut value = expected.clone(); - value.publisher_sequence = 2; - value - }, - { - let mut value = expected.clone(); - value.activation_witness_sha256 = format!("sha256-{}", "f".repeat(64)); - value - }, - { - let mut value = expected.clone(); - value.runtime_process_id = value.runtime_process_id.checked_add(1).unwrap(); - value - }, - { - let mut value = expected.clone(); - value.runtime_process_starttime_ticks = value - .runtime_process_starttime_ticks - .checked_add(1) - .unwrap(); - value - }, - ] { - validate_traffic_admission_projection(&substituted).unwrap(); - assert!(require_exact_traffic_admission(&substituted, &expected).is_err()); - substituted.schema_version = "idunn.runtime_traffic_admission.v1".into(); - assert!(validate_traffic_admission_projection(&substituted).is_err()); + fn expected(provider_identity_id: String) -> IdunnExpectedIncarnationRecord { + IdunnExpectedIncarnationRecord { + schema_version: IDUNN_EXPECTED_INCARNATION_SCHEMA.into(), + target: CONNECTOR_TARGET.into(), + plan_id: sha('1'), + incarnation_id: "codex-connector-test".into(), + sealed_release_id: sha('2'), + source_repository: "github.com/GameCult/CodexConnector".into(), + source_revision: "3".repeat(40), + recipe_sha256: sha('4'), + runtime_id: "codex-connector-yggdrasil".into(), + expected_signer_identity_id: provider_identity_id, + health_contract: CONNECTOR_HEALTH_CONTRACT.into(), + artifact_sha256: sha('5'), + state_schema_generation: Some(CONNECTOR_STATE_SCHEMA_GENERATION.into()), + state_contract_sha256: Some(CONNECTOR_STATE_CONTRACT_SHA256.into()), + write_lease_required: true, + route: Some(cultnet_rs::IdunnExpectedRoute { + route_id: "codex-connector-private".into(), + transport: "tcp".into(), + stable_endpoint: "tcp://127.0.0.1:4103".into(), + candidate_endpoint: "tcp://127.0.0.1:18831".into(), + }), + capabilities: vec![IdunnExpectedCapability { + capability: CONNECTOR_CAPABILITY.into(), + schema: crate::ENVELOPE_SCHEMA_ID.into(), + compatibility: CONNECTOR_COMPATIBILITY.into(), + minimum_capacity: 1, + }], + dependencies: Vec::new(), } } #[test] - fn proc_stat_starttime_parser_binds_pid_and_kernel_tick_identity() { - let stat = - "4242 (connector ) worker) S 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 98765 20\n"; - assert_eq!(parse_proc_stat_starttime(stat, 4242).unwrap(), 98_765); - assert!(parse_proc_stat_starttime(stat, 4243).is_err()); - assert!( - parse_proc_stat_starttime( - "4242 (connector) S 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 0", - 4242, - ) - .is_err() - ); - } - - #[test] - fn embedded_toolchain_manifest_is_the_exact_source_contract() { - let manifest = parse_toolchain_manifest(SOURCE_TOOLCHAIN_MANIFEST).unwrap(); - assert_eq!( - manifest.package_url, - "https://registry.npmjs.org/@openai/codex/-/codex-0.150.0-alpha.7-linux-x64.tgz" - ); - assert_eq!( - manifest.package_sha256, - "8aca3112aef60127d4ffaf4c6f08e524fa7315279a1d08e7aa9c5e7f1a6943e0" - ); - assert_eq!( - manifest.binary_sha256, - "aca268dc02dcef8b1ea9f528a10173d5071ae76d659a03cce6458bbebe228bee" - ); - } - - #[test] - fn malformed_or_mismatched_release_witness_never_authenticates() { - let fixture = ReleaseFixture::new(); - let malformed = [fixture.witness_bytes.as_slice(), b"deployment_id=stale\n"].concat(); - std::fs::write(&fixture.witness, malformed).unwrap(); - assert!(fixture.binding().is_err()); - - std::fs::write(&fixture.witness, &fixture.witness_bytes).unwrap(); - let witness_text = std::str::from_utf8(&fixture.witness_bytes).unwrap(); - for substituted in [ - witness_text.replace( - fixture.source_toolchain.package_url.as_str(), - "https://registry.npmjs.org/@openai/codex/-/codex-0.149.0-linux-x64.tgz", - ), - witness_text.replace( - fixture.source_toolchain.package_sha256.as_str(), - &"b".repeat(64), - ), - ] { - std::fs::write(&fixture.witness, substituted).unwrap(); - assert!( - fixture - .binding() - .unwrap_err() - .to_string() - .contains("source-owned Codex toolchain") - ); - } - std::fs::write(&fixture.witness, &fixture.witness_bytes).unwrap(); - assert!( - release_binding_from_artifacts( - &fixture.connector, - &fixture.codex, - &fixture.codex_sha256, - "ffffffffffffffffffffffffffffffffffffffff", - DEPLOYMENT_ID, - &fixture.source_toolchain, - ) - .is_err() - ); - - std::fs::write(&fixture.connector, b"substituted connector").unwrap(); - assert!(fixture.binding().is_err()); - std::fs::write(&fixture.connector, b"exact connector fixture").unwrap(); - - std::fs::write(&fixture.codex, b"substituted codex").unwrap(); - assert!(fixture.binding().is_err()); - std::fs::write(&fixture.codex, b"exact codex fixture").unwrap(); - - assert!( - release_binding_from_artifacts( - &fixture.connector, - &fixture.codex, - &[7; 32], - SOURCE_COMMIT, - DEPLOYMENT_ID, - &fixture.source_toolchain, - ) - .is_err() - ); - assert!( - release_binding_from_artifacts( - &fixture.connector, - &fixture.codex, - &fixture.codex_sha256, - SOURCE_COMMIT, - "not a current/request", - &fixture.source_toolchain, - ) - .is_err() - ); - } - - #[test] - fn signed_health_requires_the_complete_release_binding() { - let key = SigningKey::from_bytes(&[7; 32]); - let public_key = key.verifying_key().to_bytes(); - let signer = ProviderHealthSigner { - entry: ProviderHealthIdentity { - schema_version: IDENTITY_SCHEMA.into(), - identity_id: identity_id(&public_key), - public_key: public_key.to_vec(), - protected_private_seed: vec![0; 32], - protector_kind: "linux_file_mode_machine_id_binding".into(), - protector_binding: "fixture".into(), - protector_version: "v1".into(), - assurance: "os_installation_file_bound_cloneable_baseline".into(), - created_at: "2026-09-02T12:00:00Z".into(), - enrollment_nonce: vec![1; 32], + fn presence_is_dual_proved_and_warming_never_claims_the_write_lease() -> Result<()> { + let root = std::env::temp_dir().join(format!( + "codex-connector-presence-{}-{}", + std::process::id(), + unix_millis()? + )); + std::fs::create_dir_all(&root)?; + let provider = enroll_service_identity_at::( + &root.join("provider.cc"), + )?; + let provider_public_key = provider.entry().public_key.clone(); + let idunn = enroll_service_identity_at::(&root.join("idunn.cc"))?; + let expected = expected(provider.entry().identity_id.clone()); + let issued_at = unix_millis()?; + let launch = IdunnRuntimeActivationLaunch::issue(&expected, sha('7'), issued_at, &idunn)?; + let activation = launch.activation().clone(); + let mut credential = Vec::new(); + assert_eq!(launch.write_credential(&mut credential)?, activation); + let activation_signer = + IdunnRuntimeActivationSigner::from_credential_reader(Cursor::new(credential))?; + let mut publisher = RuntimePresencePublisher { + endpoint: "127.0.0.1:9".parse()?, + authority: RuntimeAuthorityMaterial { + expected_sha256: expected.canonical_sha256()?, + activation_sha256: activation.canonical_sha256()?, + expected: expected.clone(), + activation: activation.clone(), + activation_signer, + provider_signer: provider, }, - key, + bound_endpoint: "tcp://127.0.0.1:18831".into(), + capabilities: vec![GameCultRuntimeCapability { + capability: CONNECTOR_CAPABILITY.into(), + schema: crate::ENVELOPE_SCHEMA_ID.into(), + compatibility: CONNECTOR_COMPATIBILITY.into(), + capacity: 8, + }], + write_lease_path: root.join("lease.cc"), + write_lease: None, + sequence: 0, }; - let mut record = SignedDaemonHealth { - schema_version: HEALTH_SCHEMA.into(), - daemon_id: "yggdrasil-codex-connector".into(), - health_contract: CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT.into(), - source_runtime_id: "codex-connector-yggdrasil".into(), - state: "active".into(), - detail: "credential-isolated-transport-ready".into(), - signer_identity_id: signer.entry.identity_id.clone(), - publisher_incarnation_id: uuid::Uuid::new_v4().to_string(), - publisher_sequence: 1, - observed_at_unix_millis: 1_787_315_696_789, - release_id: Some(format!("{SOURCE_COMMIT}-aaaaaaaaaaaa")), - release_witness_sha256: Some(format!("sha256-{}", "b".repeat(64))), - source_commit: Some(SOURCE_COMMIT.into()), - deployment_id: Some(DEPLOYMENT_ID.into()), + + let warming = publisher.signed_record("warming", "waiting", issued_at)?; + assert_eq!(warming.publisher_sequence, 1); + assert_eq!(warming.write_lease_sha256, None); + let lock_path = root.join("lease.cc.lock"); + File::create(&lock_path)?; + let lock = OpenOptions::new().read(true).open(lock_path)?; + fs2::FileExt::lock_shared(&lock)?; + publisher.write_lease = Some(ProcessWriteLeaseGuard { + held: Arc::new(HeldProcessWriteLease { + _lock: lock, + path: root.join("lease.cc"), + expected: expected.clone(), + expected_sha256: expected.canonical_sha256()?, + activation: activation.clone(), + activation_sha256: activation.canonical_sha256()?, + warming_presence_sha256: warming.canonical_sha256()?, + lease_sha256: sha('8'), + }), + }); + let active = publisher.signed_record("active", "ready", issued_at)?; + assert_eq!(active.publisher_sequence, 2); + assert_eq!(active.signature.len(), 64); + assert_eq!(active.activation_signature.len(), 64); + + let authority = verify_runtime_authority( + &expected, + &activation, + &idunn.trust_anchor()?, + &provider_public_key, + )?; + let payload = rmp_serde::to_vec(&active)?; + let claim = authenticate_runtime_presence_claim( + &payload, + &authority, + RuntimePresenceAuthenticationContext { + trusted_received_at_unix_millis: issued_at, + maximum_age_millis: 1_000, + maximum_future_skew_millis: 1_000, + }, + )?; + assert_eq!(claim.record(), &active); + let contender = OpenOptions::new() + .read(true) + .write(true) + .open(root.join("lease.cc.lock"))?; + assert!(fs2::FileExt::try_lock_exclusive(&contender).is_err()); + drop(publisher); + fs2::FileExt::try_lock_exclusive(&contender)?; + fs2::FileExt::unlock(&contender)?; + std::fs::remove_dir_all(root)?; + Ok(()) + } + + #[test] + fn expected_must_name_the_connector_implemented_health_and_state_contract() -> Result<()> { + let mut expected = expected("provider".into()); + let actual = GameCultRuntimeCapability { + capability: CONNECTOR_CAPABILITY.into(), + schema: crate::ENVELOPE_SCHEMA_ID.into(), + compatibility: CONNECTOR_COMPATIBILITY.into(), + capacity: 1, + }; + require_connector_expected_contract(&expected, &actual)?; + + expected.health_contract = "configured.health.v1".into(); + assert!(require_connector_expected_contract(&expected, &actual).is_err()); + expected.health_contract = CONNECTOR_HEALTH_CONTRACT.into(); + expected.target = "another-target".into(); + assert!(require_connector_expected_contract(&expected, &actual).is_err()); + expected.target = CONNECTOR_TARGET.into(); + expected.state_contract_sha256 = Some(sha('6')); + assert!(require_connector_expected_contract(&expected, &actual).is_err()); + expected.state_contract_sha256 = Some(CONNECTOR_STATE_CONTRACT_SHA256.into()); + expected.state_schema_generation = Some("configured-state-v1".into()); + assert!(require_connector_expected_contract(&expected, &actual).is_err()); + Ok(()) + } + + #[test] + fn write_lease_must_bind_the_exact_warming_incarnation() -> Result<()> { + let expected = expected("provider".into()); + let expected_sha256 = expected.canonical_sha256()?; + let activation_public_key = vec![0; 32]; + let activation = IdunnRuntimeActivationRecord { + schema_version: IDUNN_RUNTIME_ACTIVATION_SCHEMA.into(), + expected_projection_sha256: expected_sha256.clone(), + runtime_id: expected.runtime_id.clone(), + runtime_instance_id: sha('7'), + activation_signer_identity_id: derive_idunn_runtime_activation_identity_id( + &activation_public_key, + )?, + activation_signer_public_key: activation_public_key, + issued_at_unix_millis: 1, + idunn_signer_identity_id: "idunn".into(), signature_algorithm: "ed25519".into(), - signature: Vec::new(), - private_state_exposed: false, - activation_witness_sha256: Some(format!("sha256-{}", "c".repeat(64))), + signature: vec![0; 64], }; - let unsigned = unsigned_record(&record).unwrap(); - record.signature = signer - .key - .sign(&signing_message(&unsigned)) - .to_bytes() - .to_vec(); - record.validate().unwrap(); - let (encoded, digest) = canonical_signed_health_payload(&record).unwrap(); - assert_eq!(&encoded[..3], &[0xdc, 0, 18]); - assert_eq!(digest, format!("sha256-{}", hex(&Sha256::digest(&encoded)))); - let public: [u8; 32] = signer.entry.public_key.try_into().unwrap(); - let signature = Signature::from_slice(&record.signature).unwrap(); - VerifyingKey::from_bytes(&public) - .unwrap() - .verify(&signing_message(&unsigned), &signature) - .unwrap(); - assert!(!record.private_state_exposed); - - for missing in 0..5 { - let mut incomplete = record.clone(); - match missing { - 0 => incomplete.release_id = None, - 1 => incomplete.release_witness_sha256 = None, - 2 => incomplete.source_commit = None, - 3 => incomplete.deployment_id = None, - _ => incomplete.activation_witness_sha256 = None, - } - assert!(incomplete.validate().is_err()); - } + let activation_sha256 = activation.canonical_sha256()?; + let warming_sha256 = sha('8'); + let mut lease = IdunnProcessWriteLeaseRecord { + schema_version: IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into(), + target: expected.target.clone(), + expected_projection_sha256: expected_sha256.clone(), + plan_id: expected.plan_id.clone(), + incarnation_id: expected.incarnation_id.clone(), + sealed_release_id: expected.sealed_release_id.clone(), + activation_witness_sha256: activation_sha256.clone(), + state_schema_generation: expected.state_schema_generation.clone().unwrap(), + state_contract_sha256: expected.state_contract_sha256.clone().unwrap(), + runtime_id: expected.runtime_id.clone(), + runtime_instance_id: activation.runtime_instance_id.clone(), + warming_presence_sha256: warming_sha256.clone(), + lease_epoch: 1, + issued_at_unix_millis: 1, + }; + assert_eq!( + exact_process_write_lease_sha256( + &lease, + &expected, + &expected_sha256, + &activation, + &activation_sha256, + &warming_sha256, + )?, + lease.canonical_sha256()? + ); + lease.warming_presence_sha256 = sha('9'); + assert!( + exact_process_write_lease_sha256( + &lease, + &expected, + &expected_sha256, + &activation, + &activation_sha256, + &warming_sha256, + ) + .is_err() + ); + Ok(()) } } diff --git a/src/main.rs b/src/main.rs index f57bd81..f3b54e8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,7 +4,7 @@ use std::path::PathBuf; use codex_connector::{CodexCallerConfig, CodexDaemonConfig}; -const USAGE: &str = "usage:\n codex-connector --config PATH.cc\n codex-connector --initialize-single-caller-config PATH.cc BIND CODEX_EXECUTABLE CODEX_SHA256 CODEX_HOME REPLAY_STORE CALLER_RUNTIME_ID CONNECTION_KEY_FILE CONNECTION_KEY_EPOCH ALLOWED_MODELS_CSV MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS\n codex-connector --admit-caller-config SOURCE.cc DESTINATION.cc CALLER_RUNTIME_ID CONNECTION_KEY_FILE CONNECTION_KEY_EPOCH ALLOWED_MODELS_CSV MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS"; +const USAGE: &str = "usage:\n codex-connector --config PATH.cc\n codex-connector --config PATH.cc --state-root PATH\n codex-connector --initialize-single-caller-config PATH.cc BIND CODEX_EXECUTABLE CODEX_SHA256 CODEX_HOME REPLAY_STORE CALLER_RUNTIME_ID CONNECTION_KEY_FILE CONNECTION_KEY_EPOCH ALLOWED_MODELS_CSV MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS\n codex-connector --admit-caller-config SOURCE.cc DESTINATION.cc CALLER_RUNTIME_ID CONNECTION_KEY_FILE CONNECTION_KEY_EPOCH ALLOWED_MODELS_CSV MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS"; fn main() { let args = std::env::args_os().skip(1).collect::>(); @@ -29,14 +29,19 @@ fn main() { } return; } - let config = match args.as_slice() { - [flag, path] if flag == "--config" => PathBuf::from(path), + let (config, state_root) = match args.as_slice() { + [flag, path] if flag == "--config" => (PathBuf::from(path), None), + [config_flag, config, state_flag, state_root] + if config_flag == "--config" && state_flag == "--state-root" => + { + (PathBuf::from(config), Some(PathBuf::from(state_root))) + } _ => { eprintln!("{USAGE}"); std::process::exit(2); } }; - if let Err(error) = codex_connector::serve(&config) { + if let Err(error) = codex_connector::serve(&config, state_root.as_deref()) { eprintln!("codex-connector stopped: {error}"); std::process::exit(1); } diff --git a/src/provider_backend.rs b/src/provider_backend.rs index c78c19a..8873708 100644 --- a/src/provider_backend.rs +++ b/src/provider_backend.rs @@ -635,6 +635,7 @@ impl CodexAppServerAuthority { .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::inherit()); + strip_systemd_descriptor_environment(&mut command); let mut child = command.spawn().map_err(CodexProviderBackendError::Spawn)?; let stdin = match child.stdin.take() { Some(stdin) => stdin, @@ -697,6 +698,13 @@ impl CodexAppServerAuthority { } } +pub(crate) fn strip_systemd_descriptor_environment(command: &mut Command) { + command + .env_remove("LISTEN_PID") + .env_remove("LISTEN_FDS") + .env_remove("LISTEN_FDNAMES"); +} + fn codex_cli_version_from_user_agent( user_agent: &str, ) -> Result { diff --git a/vendor/cultcache-rs b/vendor/cultcache-rs new file mode 160000 index 0000000..73b2555 --- /dev/null +++ b/vendor/cultcache-rs @@ -0,0 +1 @@ +Subproject commit 73b2555bbcd5f05f091d2cf17fcb86ac30695cae diff --git a/vendor/cultnet-rs b/vendor/cultnet-rs new file mode 160000 index 0000000..9dcb88d --- /dev/null +++ b/vendor/cultnet-rs @@ -0,0 +1 @@ +Subproject commit 9dcb88df1361b157a48502f43093fe7a25ca22b7