diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..640c1a1 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,3 @@ +Repository doctrine for this project lives in AGENTS.md, shared with Codex. + +@AGENTS.md diff --git a/Cargo.lock b/Cargo.lock index b872432..100cc6c 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=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c)", + "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,28 +203,11 @@ dependencies = [ [[package]] name = "cultcache-rs" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f" +source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" dependencies = [ "anyhow", "chrono", - "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f)", - "fs2", - "rmp-serde", - "serde", - "serde_bytes", - "serde_json", - "uuid", - "windows-sys 0.59.0", -] - -[[package]] -name = "cultcache-rs" -version = "0.1.0" -source = "git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c#fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c" -dependencies = [ - "anyhow", - "chrono", - "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c)", + "cultcache-rs-derive", "fs2", "redb", "rmp-serde", @@ -239,17 +221,7 @@ dependencies = [ [[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=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c#fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c" +source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" dependencies = [ "proc-macro2", "quote", @@ -259,13 +231,14 @@ dependencies = [ [[package]] name = "cultnet-rs" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f" +source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" 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", @@ -276,6 +249,8 @@ dependencies = [ "sha2", "socket2", "uuid", + "windows-sys 0.61.2", + "zeroize", ] [[package]] @@ -1184,6 +1159,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 ba6a00e..e4087ba 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 = "fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c", optional = true } -cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "75c180782aeba7cfd22d6412877397708a4ed28f", optional = true } -ed25519-dalek = { version = "2.2", optional = true } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultcache-rs", optional = true } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "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 7410ed3..550db9c 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,14 @@ subscription inference. Epiphany and Ghostlight remain separate minds. They derive their own exact provider requests, execute their own tools, interpret their own results, and admit their own state. +**Valuable here:** the consumer ABI in `src/lib.rs`. It holds the v2 typed contracts (`CodexProviderRequest`, `CodexTransportInvocation`, `CodexTransportReceipt`), exact Responses lowering (`canonical_responses_body`), AES-256-GCM envelopes (`encrypt_invocation`, `decrypt_result`) and the one-request client (`CodexConnectorClient`). The daemon behind it (`src/daemon.rs`, `src/provider_backend.rs`) is the only process that ever holds the Codex credential. + +**Problem it solves:** several minds (Epiphany, Ghostlight) need Codex-compatible subscription inference, but must not each hold or refresh the credential, copy wire law, or hand cognition to a transport service. The connector isolates the credential and transports exact provider requests with receipts, and decides nothing about prompts or tools. + +**Entry points:** library `src/lib.rs` (no daemon code, safe to depend on); daemon feature `--features daemon`, binary `codex-connector` (`src/main.rs`); Idunn declaration `deployment/idunn/recipe.toml`, deployed with `idunn up codex-connector`; verification commands are listed below. + +**Status:** live as the library plus the daemon and its Idunn recipe, last changed 2026-09-13. This repo records no planned work; changes to the contract below are the open surface. + The service owns authenticated caller admission, bounded transport physiology, exact provider-request receipts, and redacted CultMesh health. A private pinned official `codex app-server` child is the sole writer and refresh authority for @@ -54,11 +62,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. -The daemon publishes private-free signed health to Idunn from its own runtime -identity. Its source revision also owns the exact Codex Linux package and -binary hashes in `deployment/codex-linux-x64.manifest`, so changing the provider -toolchain necessarily changes the connector release selected by Idunn. +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 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..f46a4b5 --- /dev/null +++ b/deployment/idunn/recipe.toml @@ -0,0 +1,122 @@ +schema = "gamecult.idunn.target_declaration.v1" +target = "codex-connector" +source_stamp_environment = "CODEX_CONNECTOR_BUILD_COMMIT" + +[[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", +] + +[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 982a86e..c7287bc 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -7,11 +7,20 @@ use std::sync::{Arc, Mutex}; use std::thread; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +#[cfg(target_os = "linux")] +use std::os::unix::fs::MetadataExt; + use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; use serde::{Deserialize, Serialize}; use thiserror::Error; use zeroize::Zeroizing; +#[cfg(target_os = "linux")] +use cultnet_rs::{ + CultNetMessage, CultNetWireContract, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, + decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, +}; + use crate::{ CodexAppServerConfig, CodexCallerAdmission, CodexProviderBackend, CodexProviderBackendError, CodexTransportAdmission, CodexTransportEnvelope, CodexTransportService, ServiceError, @@ -110,18 +119,24 @@ impl CodexDaemonConfig { { return Err(CodexDaemonError::InvalidConfig("runtime bounds")); } + let required_connections = required_connection_capacity(&self.callers)?; + if self.max_connections < required_connections { + return Err(CodexDaemonError::InvalidConfig("connection capacity")); + } let mut caller_ids = HashSet::new(); for caller in &self.callers { + let mut allowed_models = HashSet::new(); if caller.caller_runtime_id.trim().is_empty() || caller.caller_runtime_id.trim() != caller.caller_runtime_id || !caller_ids.insert(caller.caller_runtime_id.as_str()) || caller.connection_key_file.as_os_str().is_empty() || caller.connection_key_epoch == 0 || caller.allowed_models.is_empty() - || caller - .allowed_models - .iter() - .any(|model| model.trim().is_empty() || model.trim() != model) + || caller.allowed_models.iter().any(|model| { + model.trim().is_empty() + || model.trim() != model + || !allowed_models.insert(model.as_str()) + }) || caller.max_concurrent_requests == 0 || caller.max_payload_bytes < 4096 || caller.max_payload_bytes > self.max_frame_bytes - FRAME_OVERHEAD_BUDGET @@ -132,6 +147,40 @@ impl CodexDaemonConfig { } Ok(bind) } + + pub fn admit_caller(&mut self, caller: CodexCallerConfig) -> Result<(), CodexDaemonError> { + let mut candidate = self.clone(); + if candidate + .callers + .iter() + .any(|existing| existing.caller_runtime_id == caller.caller_runtime_id) + { + return Err(CodexDaemonError::InvalidConfig("duplicate caller")); + } + candidate.max_frame_bytes = candidate.max_frame_bytes.max( + caller + .max_payload_bytes + .saturating_add(FRAME_OVERHEAD_BUDGET), + ); + candidate.callers.push(caller); + candidate.max_connections = required_connection_capacity(&candidate.callers)?; + candidate.validate()?; + *self = candidate; + Ok(()) + } +} + +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) + }) + .ok_or(CodexDaemonError::InvalidConfig("connection capacity")) } pub fn load_daemon_config(path: &Path) -> Result { @@ -150,6 +199,30 @@ pub fn load_daemon_config(path: &Path) -> Result Result { + let store = SingleFileMessagePackBackingStore::new(path); + let config = store + .with_read_only_shared_snapshot(|entries| { + let [envelope] = entries.as_slice() else { + anyhow::bail!("connector configuration must contain exactly one record"); + }; + if envelope.key != CONFIG_KEY + || envelope.r#type != CodexDaemonConfig::TYPE + || envelope.schema_id.as_deref() != Some(CodexDaemonConfig::TYPE) + { + anyhow::bail!("connector configuration has the wrong typed envelope"); + } + let config: CodexDaemonConfig = rmp_serde::from_slice(&envelope.payload)?; + if rmp_serde::to_vec(&config)? != envelope.payload { + anyhow::bail!("connector configuration is not canonical positional MessagePack"); + } + Ok(config) + }) + .map_err(config_store_error)?; + config.validate()?; + Ok(config) +} + pub fn write_daemon_config( path: &Path, config: &CodexDaemonConfig, @@ -167,37 +240,109 @@ pub fn write_daemon_config( Ok(()) } -pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> { - let config = load_daemon_config(config_path)?; - let bind = config.validate()?; - let service = Arc::new(Mutex::new(CodexTransportService::open( - &config.replay_store, - load_admissions(&config)?, - config.max_expiry_skew_ms, - )?)); +pub fn serve( + config_path: &Path, + managed_state_root: Option<&Path>, +) -> Result<(), CodexDaemonError> { + #[cfg(target_os = "linux")] + 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)?; + + #[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 listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?; + let service = Arc::new(Mutex::new(CodexTransportService::open( + &replay_store, + admissions, + config.max_expiry_skew_ms, + )?)); + #[cfg(target_os = "linux")] + let runtime_presence_publisher = + prepared_presence_publisher.map(|publisher| Arc::new(Mutex::new(publisher))); + #[cfg(target_os = "linux")] + if let Some(publisher) = runtime_presence_publisher.as_ref() { + start_periodic_runtime_presence_publisher(publisher.clone(), backend.clone())?; + } let active_connections = Arc::new(AtomicUsize::new(0)); eprintln!( "codex-connector ready at {} for {} callers using {:?}", - listener.local_addr().map_err(CodexDaemonError::Listen)?, + bound, config.callers.len(), readiness.auth_mode ); - #[cfg(target_os = "linux")] - start_idunn_health_publisher()?; - for accepted in listener.incoming() { - let stream = match accepted { - Ok(stream) => stream, - Err(error) => return Err(CodexDaemonError::Listen(error)), - }; + loop { + let (stream, _) = listener.accept().map_err(CodexDaemonError::Listen)?; if !stream .peer_addr() .map_err(CodexDaemonError::Connection)? @@ -213,49 +358,66 @@ pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> { }; let service = service.clone(); let backend = backend.clone(); + #[cfg(target_os = "linux")] + let runtime_presence_publisher = runtime_presence_publisher.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, max_frame_bytes, timeout); + let _ = serve_connection( + stream, + &service, + &backend, + #[cfg(target_os = "linux")] + runtime_presence_publisher.as_deref(), + max_frame_bytes, + timeout, + ); }) .map_err(CodexDaemonError::Thread)?; } - Err(CodexDaemonError::ListenerClosed) } #[cfg(target_os = "linux")] -fn start_idunn_health_publisher() -> Result<(), CodexDaemonError> { - let Some(endpoint) = std::env::var_os("CODEX_CONNECTOR_IDUNN_RUDP") else { - return Ok(()); - }; - let endpoint = endpoint - .to_string_lossy() - .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::CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT.to_string()); - let identity = PathBuf::from(required_environment( - "CODEX_CONNECTOR_IDUNN_HEALTH_IDENTITY", - )?); - let mut publisher = - crate::ProviderHealthPublisher::open(endpoint, daemon_id, runtime_id, contract, &identity) +fn prepare_runtime_presence_publisher( + candidate: SocketAddr, + config: &CodexDaemonConfig, +) -> 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_runtime_presence_publisher( + publisher: Arc>, + backend: Arc, +) -> Result<(), CodexDaemonError> { + let (state, detail) = provider_health_observation(&backend); publisher - .publish("active", "credential-isolated-transport-ready") + .lock() + .map_err(|_| CodexDaemonError::RuntimePresencePoisoned)? + .publish(state, detail) .map_err(CodexDaemonError::Health)?; thread::Builder::new() .name("codex-connector-health".to_string()) .spawn(move || { loop { thread::sleep(Duration::from_secs(10)); - if let Err(error) = - publisher.publish("active", "credential-isolated-transport-ready") - { + let (state, detail) = provider_health_observation(&backend); + let publication = publisher + .lock() + .map_err(|_| anyhow::anyhow!("runtime presence publisher lock was poisoned")) + .and_then(|mut publisher| publisher.publish(state, detail)); + if let Err(error) = publication { eprintln!("codex-connector health publication failed: {error}"); } } @@ -265,25 +427,153 @@ fn start_idunn_health_publisher() -> Result<(), CodexDaemonError> { } #[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 provider_health_observation(backend: &CodexProviderBackend) -> (&'static str, &'static str) { + provider_health_state(backend.readiness().is_ok()) +} + +#[cfg(target_os = "linux")] +fn provider_health_state(ready: bool) -> (&'static str, &'static str) { + if ready { + ("active", "credential-isolated-transport-ready") + } else { + ("degraded", "provider-backend-unready") + } +} + +#[cfg(target_os = "linux")] +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(), + ) +} + +#[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, CodexDaemonError> { - config - .callers - .iter() - .map(|caller| { - let secret = Zeroizing::new( - fs::read_to_string(&caller.connection_key_file) - .map_err(CodexDaemonError::ConnectionKey)?, - ); - let secret = secret.trim_end_matches(['\r', '\n']); + let mut admissions = 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)?, + ); + let secret = std::str::from_utf8(raw.as_slice()) + .map_err(|_| CodexDaemonError::InvalidConfig("caller connection key encoding"))? + .trim_end_matches(['\r', '\n']); + admissions.push( CodexCallerAdmission::new( caller.caller_runtime_id.clone(), secret.to_string(), @@ -293,15 +583,19 @@ fn load_admissions( caller.max_payload_bytes, caller.max_output_tokens, ) - .map_err(CodexDaemonError::Service) - }) - .collect() + .map_err(CodexDaemonError::Service)?, + ); + } + Ok(admissions) } fn serve_connection( mut stream: TcpStream, service: &Mutex, backend: &CodexProviderBackend, + #[cfg(target_os = "linux")] runtime_presence_publisher: Option< + &Mutex, + >, max_frame_bytes: usize, timeout: Duration, ) -> Result<(), CodexDaemonError> { @@ -310,8 +604,37 @@ fn serve_connection( .and_then(|()| stream.set_write_timeout(Some(timeout))) .map_err(CodexDaemonError::Connection)?; let request = read_transport_frame(&mut stream, max_frame_bytes).map_err(frame_error)?; + #[cfg(target_os = "linux")] + let response = dispatch_native_frame( + &request, + runtime_presence_publisher.is_some(), + |message_id| { + let publisher = + runtime_presence_publisher.ok_or(CodexDaemonError::RouteObservationRefused)?; + let mut publisher = publisher + .lock() + .map_err(|_| CodexDaemonError::RuntimePresencePoisoned)?; + let backend_ready = backend.readiness().is_ok(); + let response = publisher + .route_observation(message_id, backend_ready) + .map_err(CodexDaemonError::Health)?; + encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0) + .map_err(|_| CodexDaemonError::FrameEncoding) + }, + |request| execute_inference_frame(request, service, backend), + )?; + #[cfg(not(target_os = "linux"))] + let response = execute_inference_frame(&request, service, backend)?; + write_transport_frame(&mut stream, &response, max_frame_bytes).map_err(frame_error) +} + +fn execute_inference_frame( + request: &[u8], + service: &Mutex, + backend: &CodexProviderBackend, +) -> Result, CodexDaemonError> { let envelope: CodexTransportEnvelope = - rmp_serde::from_slice(&request).map_err(|_| CodexDaemonError::FrameEncoding)?; + rmp_serde::from_slice(request).map_err(|_| CodexDaemonError::FrameEncoding)?; let admission = service .lock() .map_err(|_| CodexDaemonError::ServicePoisoned)? @@ -326,8 +649,45 @@ fn serve_connection( .complete(*claim, result)? } }; - let response = rmp_serde::to_vec(&response).map_err(|_| CodexDaemonError::FrameEncoding)?; - write_transport_frame(&mut stream, &response, max_frame_bytes).map_err(frame_error) + rmp_serde::to_vec(&response).map_err(|_| CodexDaemonError::FrameEncoding) +} + +#[cfg(target_os = "linux")] +fn dispatch_native_frame( + request: &[u8], + managed: bool, + route_observation: RouteObservation, + inference: Inference, +) -> Result, CodexDaemonError> +where + RouteObservation: FnOnce(&str) -> Result, CodexDaemonError>, + Inference: FnOnce(&[u8]) -> Result, CodexDaemonError>, +{ + let Ok(message) = + decode_cultnet_message_from_slice(request, CultNetWireContract::CultNetSchemaV0) + else { + return inference(request); + }; + let CultNetMessage::SnapshotRequest { + message_id, + schema_ids, + record_keys, + } = message + else { + return inference(request); + }; + let exact_schema = matches!( + schema_ids.as_deref(), + Some([schema_id]) if schema_id == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA + ); + let exact_record = matches!( + record_keys.as_deref(), + Some([record_key]) if record_key == crate::idunn_health::CONNECTOR_TARGET + ); + if !managed || !exact_schema || !exact_record { + return Err(CodexDaemonError::RouteObservationRefused); + } + route_observation(&message_id) } fn frame_error(error: TransportFrameError) -> CodexDaemonError { @@ -398,11 +758,17 @@ pub enum CodexDaemonError { #[error("connector request thread failed")] Thread(#[source] std::io::Error), #[cfg(target_os = "linux")] - #[error("connector health publication could not start: {0}")] + #[error("connector runtime presence failed: {0}")] Health(#[source] anyhow::Error), #[cfg(target_os = "linux")] #[error("connector health thread could not start")] HealthThread(#[source] std::io::Error), + #[cfg(target_os = "linux")] + #[error("connector runtime presence publisher lock was poisoned")] + RuntimePresencePoisoned, + #[cfg(target_os = "linux")] + #[error("connector route observation request was refused")] + RouteObservationRefused, #[error("connector connection failed")] Connection(#[source] std::io::Error), #[error("connector frame exceeded its bound")] @@ -411,13 +777,13 @@ pub enum CodexDaemonError { FrameEncoding, #[error("system clock is before the Unix epoch")] Clock, - #[error("connector listener closed")] - ListenerClosed, } #[cfg(test)] mod tests { use super::*; + #[cfg(target_os = "linux")] + use std::cell::Cell; use std::io::Cursor; fn config(root: &Path) -> CodexDaemonConfig { @@ -444,6 +810,118 @@ mod tests { } } + #[test] + #[cfg(target_os = "linux")] + fn signed_health_never_reports_an_unready_provider_active() { + assert_eq!( + provider_health_state(true), + ("active", "credential-isolated-transport-ready") + ); + assert_eq!( + provider_health_state(false), + ("degraded", "provider-backend-unready") + ); + } + + #[test] + #[cfg(target_os = "linux")] + fn exact_managed_route_probe_bypasses_inference_execution() { + let request = encode_cultnet_message_to_vec( + &CultNetMessage::SnapshotRequest { + message_id: "route-challenge-17".into(), + schema_ids: Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]), + record_keys: Some(vec![crate::idunn_health::CONNECTOR_TARGET.into()]), + }, + CultNetWireContract::CultNetSchemaV0, + ) + .unwrap(); + let inference_calls = Cell::new(0); + let route_calls = Cell::new(0); + + let response = dispatch_native_frame( + &request, + true, + |message_id| { + route_calls.set(route_calls.get() + 1); + assert_eq!(message_id, "route-challenge-17"); + Ok(b"signed-route-observation".to_vec()) + }, + |_| { + inference_calls.set(inference_calls.get() + 1); + Ok(b"provider-response".to_vec()) + }, + ) + .unwrap(); + + assert_eq!(response, b"signed-route-observation"); + assert_eq!(route_calls.get(), 1); + assert_eq!(inference_calls.get(), 0); + } + + #[test] + #[cfg(target_os = "linux")] + fn route_probe_refuses_broad_foreign_and_unmanaged_requests() { + let requests = [ + ( + CultNetMessage::SnapshotRequest { + message_id: "broad".into(), + schema_ids: None, + record_keys: None, + }, + true, + ), + ( + CultNetMessage::SnapshotRequest { + message_id: "foreign-schema".into(), + schema_ids: Some(vec!["gamecult.other.v1".into()]), + record_keys: Some(vec![crate::idunn_health::CONNECTOR_TARGET.into()]), + }, + true, + ), + ( + CultNetMessage::SnapshotRequest { + message_id: "foreign-record".into(), + schema_ids: Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]), + record_keys: Some(vec!["other-target".into()]), + }, + true, + ), + ( + CultNetMessage::SnapshotRequest { + message_id: "unmanaged".into(), + schema_ids: Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]), + record_keys: Some(vec![crate::idunn_health::CONNECTOR_TARGET.into()]), + }, + false, + ), + ]; + + for (request, managed) in requests { + let request = + encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0) + .unwrap(); + let route_calls = Cell::new(0); + let inference_calls = Cell::new(0); + assert!(matches!( + dispatch_native_frame( + &request, + managed, + |_| { + route_calls.set(route_calls.get() + 1); + Ok(Vec::new()) + }, + |_| { + inference_calls.set(inference_calls.get() + 1); + Ok(Vec::new()) + }, + ), + Err(CodexDaemonError::RouteObservationRefused) + )); + assert_eq!(route_calls.get(), 0); + assert_eq!(inference_calls.get(), 0); + } + } + #[test] fn typed_cultcache_config_round_trips_and_refuses_epoch_substitution() { let root = std::env::temp_dir().join(format!( @@ -456,6 +934,8 @@ mod tests { let expected = config(&root); write_daemon_config(&path, &expected).unwrap(); assert_eq!(load_daemon_config(&path).unwrap(), expected); + let loaded = load_daemon_config_read_only(&path).unwrap(); + assert_eq!(loaded, expected); let mut stale = expected; stale.epoch = 0; @@ -466,6 +946,74 @@ mod tests { fs::remove_dir_all(root).unwrap(); } + #[test] + fn production_config_snapshot_never_creates_its_authority_lock() { + let root = std::env::temp_dir().join(format!( + "codex-connector-read-only-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(); + let before = fs::read(&path).unwrap(); + let lock_path = path.with_file_name("connector.cc.lock"); + fs::remove_file(&lock_path).unwrap(); + + 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"); @@ -494,6 +1042,125 @@ mod tests { config.validate().unwrap(); } + #[test] + fn typed_config_preserves_a_unique_multi_model_caller_allowlist() { + let root = Path::new("/srv/codex-connector"); + let mut config = config(root); + config.callers[0].allowed_models = + vec!["gpt-5.6-luna".to_string(), "gpt-5.6-terra".to_string()]; + config.validate().unwrap(); + + config.callers[0] + .allowed_models + .push("gpt-5.6-luna".to_string()); + assert!(matches!( + config.validate(), + Err(CodexDaemonError::InvalidConfig("caller admission")) + )); + } + + #[test] + fn caller_admission_reserves_every_independently_admitted_request_slot() { + let root = Path::new("/srv/codex-connector"); + let first = CodexCallerConfig { + caller_runtime_id: "ghostlight-dungeon-yggdrasil".to_string(), + connection_key_file: root.join("ghostlight.key"), + connection_key_epoch: 1, + allowed_models: vec!["gpt-5.6-luna".to_string()], + max_concurrent_requests: 8, + max_payload_bytes: 1_048_576, + max_output_tokens: 16_384, + }; + let mut config = CodexDaemonConfig::single_caller( + "127.0.0.1:4103", + root.join("codex"), + [7; 32], + root.join("codex-home"), + root.join("replay.cc"), + first, + ); + config + .admit_caller(CodexCallerConfig { + caller_runtime_id: "epiphany-model-runtime".to_string(), + connection_key_file: root.join("epiphany.key"), + connection_key_epoch: 1, + allowed_models: vec!["gpt-5.6-luna".to_string()], + max_concurrent_requests: 8, + max_payload_bytes: 1_048_576, + max_output_tokens: 16_384, + }) + .unwrap(); + + assert_eq!(config.max_connections, 16); + config.validate().unwrap(); + + config.max_connections = 15; + assert!(matches!( + config.validate(), + Err(CodexDaemonError::InvalidConfig("connection capacity")) + )); + } + + #[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 c5e06c7..73c81ff 100644 --- a/src/idunn_health.rs +++ b/src/idunn_health.rs @@ -1,364 +1,706 @@ -use std::fs::OpenOptions; -use std::io::Write; +use std::fs::{File, OpenOptions}; use std::net::{SocketAddr, UdpSocket}; -use std::os::unix::fs::PermissionsExt; -use std::path::Path; -use std::time::{Duration, Instant}; +use std::os::fd::{AsRawFd, FromRawFd, RawFd}; +use std::os::unix::fs::MetadataExt; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use anyhow::{Context, Result, anyhow, bail}; -use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, 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 rand_core::{OsRng, RngCore}; -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 WARMING_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); +const MAX_RECENT_WARMING_PROOFS: usize = 64; + +pub(crate) 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"; -// 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)] -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, Serialize, Deserialize)] -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, +struct RuntimeSignerDescriptors { + activation: File, + provider: File, } -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.release_id.is_some() - || self.release_witness_sha256.is_some() - || self.source_commit.is_some() - || self.deployment_id.is_some() - || self.signature_algorithm != "ed25519" - || self.signature.len() != 64 - || self.private_state_exposed - { - bail!("signed daemon health shape is invalid"); - } +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 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: IdunnExpectedIncarnationRecord, + expected_sha256: String, + activation: IdunnRuntimeActivationRecord, + activation_sha256: String, + warming_presence_sha256es: Vec, + lease_sha256: 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_sha256es, + )?; + ensure!( + observed.as_deref() == Some(self.sha256()), + "process-write lease is no longer the admitted lifetime grant" + ); Ok(()) } } -pub struct ProviderHealthPublisher { - endpoint: SocketAddr, - daemon_id: String, - runtime_id: String, - contract: String, - signer: ProviderHealthSigner, - incarnation: String, - sequence: u64, -} +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)?); -impl ProviderHealthPublisher { - pub fn open( - endpoint: SocketAddr, - daemon_id: impl Into, - runtime_id: impl Into, - contract: impl Into, - identity_store: &Path, - ) -> 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")?; Ok(Self { endpoint, - daemon_id, - runtime_id, - contract, - signer: open_identity(identity_store)?, - incarnation: uuid::Uuid::new_v4().to_string(), + authority, + bound_endpoint, + capabilities: vec![capability], + write_lease_path, + write_lease: None, sequence: 0, }) } - pub fn publish(&mut self, state: &str, detail: &str) -> Result<()> { + pub(crate) fn acquire_process_write_lease( + &mut self, + warming_presence_sha256: &str, + timeout: Duration, + ) -> Result<()> { + let path = self.write_lease_path.clone(); + let started = Instant::now(); + let mut next_heartbeat = Instant::now() + WARMING_HEARTBEAT_INTERVAL; + let mut recent_warming_proofs = Vec::new(); + remember_warming_proof( + &mut recent_warming_proofs, + warming_presence_sha256.to_owned(), + ); + loop { + let last_error = match acquire_process_write_lease_guard( + &path, + &self.authority.expected, + &self.authority.expected_sha256, + &self.authority.activation, + &self.authority.activation_sha256, + &recent_warming_proofs, + ) { + 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_error + .as_deref() + .unwrap_or("the process-write-lease record remained absent"); + bail!( + "timed out waiting for Idunn process-write lease {}: {detail}", + path.display() + ); + } + let now = Instant::now(); + if now >= next_heartbeat { + let heartbeat = self + .publish("warming", "awaiting-process-write-lease") + .context("publishing Warming heartbeat while awaiting process lease")?; + remember_warming_proof(&mut recent_warming_proofs, heartbeat); + next_heartbeat = now + WARMING_HEARTBEAT_INTERVAL; + } + std::thread::sleep(Duration::from_millis(100)); + } + } + + pub(crate) fn process_write_lease_guard(&self) -> Result { + self.write_lease + .clone() + .context("Connector has not acquired its process-write lease") + } + + pub(crate) fn route_observation( + &mut self, + message_id: &str, + backend_ready: bool, + ) -> Result { + ensure!( + !message_id.is_empty(), + "route-observation message id is empty" + ); + let guard = self + .write_lease + .as_ref() + .context("Connector route observation lacks its process-write lease")?; + guard.require_current()?; + self.signed_route_observation(message_id, backend_ready) + } + + fn signed_route_observation( + &mut self, + message_id: &str, + backend_ready: bool, + ) -> Result { + let state = if backend_ready { "active" } else { "degraded" }; + let detail = format!("route-observation:{message_id}"); + let record = self.signed_record(state, &detail, unix_millis()?)?; + let payload = rmp_serde::to_vec(&record)?; + ensure!( + rmp_serde::from_slice::(&payload)? == record, + "route observation did not round-trip canonically" + ); + Ok(CultNetMessage::SnapshotResponseRaw { + message_id: message_id.into(), + documents: vec![presence_document( + &self.authority.expected.target, + &self.authority.expected.runtime_id, + &record, + payload, + None, + )?], + }) + } + + pub(crate) fn publish(&mut self, state: &str, detail: &str) -> Result { + if let Some(guard) = &self.write_lease { + guard.require_current()?; + } + 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) + } + + fn signed_record( + &mut self, + state: &str, + detail: &str, + 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: None, - release_witness_sha256: None, - source_commit: None, - deployment_id: None, + observed_at_unix_millis, signature_algorithm: "ed25519".into(), signature: Vec::new(), - private_state_exposed: false, + 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()?; - publish(self.endpoint, &self.runtime_id, &record) + Ok(record) } } -pub fn enroll_provider_health_identity(path: &Path) -> Result { - let binding = machine_binding()?; - enroll_provider_health_identity_with_binding(path, &binding) +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 enroll_provider_health_identity_with_binding(path: &Path, binding: &str) -> Result { - if path.exists() { - bail!("provider health identity already exists"); - } - let parent = path - .parent() - .ok_or_else(|| anyhow!("provider health identity path has no parent"))?; - std::fs::create_dir_all(parent)?; - std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700))?; - let mut seed = [0_u8; 32]; - OsRng.fill_bytes(&mut seed); - let key = SigningKey::from_bytes(&seed); - let public_key = key.verifying_key().to_bytes(); - let mut nonce = [0_u8; 32]; - OsRng.fill_bytes(&mut nonce); - let entry = ProviderHealthIdentity { - schema_version: IDENTITY_SCHEMA.into(), - identity_id: identity_id(&public_key), - public_key: public_key.to_vec(), - protected_private_seed: mask_seed(&seed, binding), - protector_kind: "linux_file_mode_machine_id_binding".into(), - protector_binding: binding.into(), - protector_version: "v1".into(), - assurance: "os_installation_file_bound_cloneable_baseline".into(), - created_at: chrono::Utc::now().to_rfc3339(), - enrollment_nonce: nonce.to_vec(), - }; - validate_identity(&entry)?; - let envelope = CultCacheEnvelope { - key: IDENTITY_KEY.into(), - r#type: IDENTITY_TYPE.into(), - payload: rmp_serde::to_vec(&entry)?, - stored_at: entry.created_at.clone(), - schema_id: Some(IDENTITY_SCHEMA.into()), - }; - let bytes = rmp_serde::to_vec(&vec![envelope])?; - let mut file = OpenOptions::new().create_new(true).write(true).open(path)?; - if let Err(error) = file.write_all(&bytes).and_then(|()| file.sync_all()) { - drop(file); - let _ = std::fs::remove_file(path); - return Err(error.into()); - } - std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?; - Ok(hex(&open_identity_with_binding(path, binding)? - .entry - .public_key)) +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())) + }) } -pub fn provider_health_public_key_hex(path: &Path) -> Result { - Ok(hex(&open_identity(path)?.entry.public_key)) -} - -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.identity_id != identity_id(&entry.public_key) - { - bail!("provider health identity is invalid"); - } - chrono::DateTime::parse_from_rfc3339(&entry.created_at)?; +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 + }) + .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" + ); + 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" + ); 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"); +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"); } - 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(), + ensure!( + !value.contains("://"), + "Expected Odin provider endpoint uses an unsupported transport" ); - seed.iter() - .zip(mask) - .map(|(left, right)| left ^ right) - .collect() + value + .parse() + .context("parsing bare Expected Odin RUDP endpoint") } -fn identity_id(public_key: &[u8]) -> String { - hex(&Sha256::digest([ID_DOMAIN, public_key].concat())) +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 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_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 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 publish(endpoint: SocketAddr, runtime_id: &str, record: &SignedDaemonHealth) -> Result<()> { - let payload = rmp_serde::to_vec(record)?; - if rmp_serde::from_slice::(&payload)? != *record { - bail!("signed health MessagePack did not round trip"); +fn load_process_write_lease( + path: &Path, + expected: &IdunnExpectedIncarnationRecord, + expected_sha256: &str, + activation: &IdunnRuntimeActivationRecord, + activation_sha256: &str, + warming_presence_sha256es: &[String], +) -> 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_sha256es, + ) + .map(Some) + }) +} + +fn acquire_process_write_lease_guard( + path: &Path, + expected: &IdunnExpectedIncarnationRecord, + expected_sha256: &str, + activation: &IdunnRuntimeActivationRecord, + activation_sha256: &str, + warming_presence_sha256es: &[String], +) -> 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_sha256es, + )?; + 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_sha256es: warming_presence_sha256es.to_vec(), + lease_sha256, + }), + })) +} + +fn exact_process_write_lease_sha256( + lease: &IdunnProcessWriteLeaseRecord, + expected: &IdunnExpectedIncarnationRecord, + expected_sha256: &str, + activation: &IdunnRuntimeActivationRecord, + activation_sha256: &str, + warming_presence_sha256es: &[String], +) -> 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 + && warming_presence_sha256es + .iter() + .any(|sha256| sha256 == &lease.warming_presence_sha256), + "process-write lease does not bind a provider-owned Warming proof for this incarnation" + ); + lease.canonical_sha256() +} + +fn remember_warming_proof(proofs: &mut Vec, sha256: String) { + proofs.push(sha256); + while proofs.len() > MAX_RECENT_WARMING_PROOFS { + proofs.remove(0); + } +} + +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(), - stored_at: chrono::DateTime::from_timestamp_millis( - record.observed_at_unix_millis.try_into()?, - ) - .context("health observation time is invalid")? - .to_rfc3339(), - payload_encoding: CultNetRawPayloadEncoding::Messagepack, + document: presence_document( + target, + runtime_id, + record, payload, - source_runtime_id: Some(runtime_id.into()), - source_agent_id: Some(record.signer_identity_id.clone()), - source_role: Some("daemon-health-publisher".into()), - tags: Some(vec![RUDP_PROTOCOL_ID.into()]), - }, + Some(vec![RUDP_PROTOCOL_ID.into()]), + )?, }; let socket = UdpSocket::bind(if endpoint.is_ipv4() { "0.0.0.0:0" @@ -366,103 +708,689 @@ 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(()) -} - -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 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 presence_document( + target: &str, + runtime_id: &str, + record: &GameCultRuntimePresenceHealthRecord, + payload: Vec, + tags: Option>, +) -> Result { + Ok(CultNetRawDocumentRecord { + 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("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("runtime-presence-health-publisher".into()), + tags, + }) +} + +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 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}; + + fn sha(digit: char) -> String { + format!("sha256-{}", digit.to_string().repeat(64)) + } #[test] - fn identity_round_trips_and_signed_health_is_private_free() { - let root = std::env::temp_dir().join(format!( - "codex-connector-health-{}-{}", - std::process::id(), - uuid::Uuid::new_v4() - )); - std::fs::create_dir_all(&root).unwrap(); - let path = root.join("identity.cc"); - let binding = "fixture-machine-binding"; - let public_key = enroll_provider_health_identity_with_binding(&path, binding).unwrap(); + fn systemd_signer_descriptor_contract_is_exact_and_process_bound() -> Result<()> { + let pid = 42; assert_eq!( - hex(&open_identity_with_binding(&path, binding) - .unwrap() - .entry - .public_key), - public_key + parse_runtime_signer_descriptor_contract( + "42", + "2", + &format!("{ACTIVATION_SIGNER_FD_NAME}:{PROVIDER_SIGNER_FD_NAME}"), + pid, + )?, + (3, 4) ); - let signer = open_identity_with_binding(&path, binding).unwrap(); - 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: None, - release_witness_sha256: None, - source_commit: None, - deployment_id: None, - signature_algorithm: "ed25519".into(), - signature: Vec::new(), - private_state_exposed: false, + 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 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-signer-fds-{}-{}", + std::process::id(), + unix_millis()? + )); + 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(); + + 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(), + ) + .env(SYSTEMD_LISTEN_FDS_ENVIRONMENT, "2") + .env( + SYSTEMD_LISTEN_FDNAMES_ENVIRONMENT, + format!("{ACTIVATION_SIGNER_FD_NAME}:{PROVIDER_SIGNER_FD_NAME}"), + ) + .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(()) + } + + 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 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, + }, + 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 unsigned = unsigned_record(&record).unwrap(); - record.signature = signer - .key - .sign(&signing_message(&unsigned)) - .to_bytes() - .to_vec(); - record.validate().unwrap(); - 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); - std::fs::remove_dir_all(root).unwrap(); + + let warming = publisher.signed_record("warming", "waiting", issued_at)?; + assert_eq!(warming.publisher_sequence, 1); + assert_eq!(warming.write_lease_sha256, None); + let heartbeat = publisher.signed_record("warming", "waiting", issued_at + 1)?; + assert_eq!(heartbeat.publisher_sequence, 2); + assert_ne!(heartbeat.canonical_sha256()?, warming.canonical_sha256()?); + 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)?; + let lease_sha256 = sha('8'); + 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_sha256es: vec![ + warming.canonical_sha256()?, + heartbeat.canonical_sha256()?, + ], + lease_sha256: lease_sha256.clone(), + }), + }); + let active = publisher.signed_record("active", "ready", issued_at)?; + assert_eq!(active.publisher_sequence, 3); + 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 route_response = publisher.signed_route_observation("challenge-23", true)?; + let CultNetMessage::SnapshotResponseRaw { + message_id, + documents, + } = route_response + else { + panic!("route observation was not a raw snapshot response"); + }; + assert_eq!(message_id, "challenge-23"); + let [document] = documents.as_slice() else { + panic!("route observation did not contain exactly one document"); + }; + assert_eq!(document.schema_id, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA); + assert_eq!(document.record_key, CONNECTOR_TARGET); + assert_eq!(document.tags, None); + let route_presence: GameCultRuntimePresenceHealthRecord = + rmp_serde::from_slice(&document.payload)?; + assert_eq!(route_presence.state, "active"); + assert_eq!(route_presence.detail, "route-observation:challenge-23"); + assert_eq!( + route_presence.write_lease_sha256.as_deref(), + Some(lease_sha256.as_str()) + ); + assert_eq!(route_presence.publisher_sequence, 4); + let claim = authenticate_runtime_presence_claim( + &document.payload, + &authority, + RuntimePresenceAuthenticationContext { + trusted_received_at_unix_millis: route_presence.observed_at_unix_millis, + maximum_age_millis: 1_000, + maximum_future_skew_millis: 1_000, + }, + )?; + assert_eq!(claim.record(), &route_presence); + + let degraded = publisher.signed_route_observation("challenge-24", false)?; + let CultNetMessage::SnapshotResponseRaw { documents, .. } = degraded else { + panic!("degraded route observation was not a raw snapshot response"); + }; + let degraded: GameCultRuntimePresenceHealthRecord = + rmp_serde::from_slice(&documents[0].payload)?; + assert_eq!(degraded.state, "degraded"); + assert_eq!(degraded.detail, "route-observation:challenge-24"); + + // A held historical guard is not enough: every route challenge must + // re-read the exact Idunn lease before signing its data-plane proof. + assert!( + publisher + .route_observation("challenge-stale", true) + .is_err() + ); + 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![0; 64], + }; + 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, + }; + let heartbeat_sha256 = sha('9'); + let provider_owned_warming = vec![warming_sha256.clone(), heartbeat_sha256.clone()]; + assert_eq!( + exact_process_write_lease_sha256( + &lease, + &expected, + &expected_sha256, + &activation, + &activation_sha256, + &provider_owned_warming, + )?, + lease.canonical_sha256()? + ); + lease.warming_presence_sha256 = heartbeat_sha256; + assert!( + exact_process_write_lease_sha256( + &lease, + &expected, + &expected_sha256, + &activation, + &activation_sha256, + &provider_owned_warming, + ) + .is_ok() + ); + lease.warming_presence_sha256 = sha('a'); + assert!( + exact_process_write_lease_sha256( + &lease, + &expected, + &expected_sha256, + &activation, + &activation_sha256, + &provider_owned_warming, + ) + .is_err() + ); + Ok(()) } } diff --git a/src/lib.rs b/src/lib.rs index 5cd7b68..d6e6b23 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -37,11 +37,6 @@ pub use daemon::load_daemon_config; pub use daemon::serve; #[cfg(feature = "daemon")] pub use daemon::write_daemon_config; -#[cfg(all(feature = "daemon", target_os = "linux"))] -pub use idunn_health::{ - CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT, ProviderHealthPublisher, - enroll_provider_health_identity, provider_health_public_key_hex, -}; #[cfg(feature = "daemon")] pub use provider_backend::CodexAppServerConfig; #[cfg(feature = "daemon")] @@ -727,8 +722,10 @@ impl CodexConnectorClient { max_frame_bytes: usize, response_timeout: Option, ) -> Result { - if !endpoint.ip().is_loopback() - || endpoint.port() == 0 + if endpoint.port() == 0 + || endpoint.ip().is_unspecified() + || endpoint.ip().is_multicast() + || endpoint.ip() == std::net::Ipv4Addr::BROADCAST || !(4096..=u32::MAX as usize).contains(&max_frame_bytes) || response_timeout.is_some_and(|timeout| timeout.is_zero()) { @@ -770,6 +767,45 @@ impl CodexConnectorClient { } } +#[cfg(test)] +mod client_endpoint_tests { + use super::*; + + #[test] + fn client_admits_explicit_unicast_and_refuses_non_destinations() { + for endpoint in ["127.0.0.1:4103", "172.19.0.1:4103", "[::1]:4103"] { + assert!( + CodexConnectorClient::new( + endpoint.parse().unwrap(), + "bounded-client-endpoint-key", + 64 * 1024, + Some(Duration::from_secs(1)), + ) + .is_ok(), + "{endpoint} should be an admissible explicit destination" + ); + } + for endpoint in [ + "0.0.0.0:4103", + "255.255.255.255:4103", + "224.0.0.1:4103", + "[::]:4103", + "[ff02::1]:4103", + "127.0.0.1:0", + ] { + assert!(matches!( + CodexConnectorClient::new( + endpoint.parse().unwrap(), + "bounded-client-endpoint-key", + 64 * 1024, + Some(Duration::from_secs(1)), + ), + Err(CodexConnectorClientError::InvalidConfig) + )); + } + } +} + #[derive(Debug, Error)] pub enum CodexConnectorClientError { #[error("invalid connector client configuration")] diff --git a/src/main.rs b/src/main.rs index 65ca37a..f3b54e8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,35 +1,13 @@ +use std::collections::HashSet; use std::ffi::OsString; 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_MODEL MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS\n codex-connector --enroll-provider-health-identity PATH.cc\n codex-connector --provider-health-public-key PATH.cc"; +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::>(); - #[cfg(target_os = "linux")] - if let [flag, path] = args.as_slice() { - if flag == "--enroll-provider-health-identity" { - match codex_connector::enroll_provider_health_identity(&PathBuf::from(path)) { - Ok(public_key) => println!("{public_key}"), - Err(error) => { - eprintln!("codex-connector health identity enrollment failed: {error}"); - std::process::exit(2); - } - } - return; - } - if flag == "--provider-health-public-key" { - match codex_connector::provider_health_public_key_hex(&PathBuf::from(path)) { - Ok(public_key) => println!("{public_key}"), - Err(error) => { - eprintln!("codex-connector health identity read failed: {error}"); - std::process::exit(2); - } - } - return; - } - } if args .first() .is_some_and(|value| value == "--initialize-single-caller-config") @@ -41,19 +19,68 @@ fn main() { } return; } - let config = match args.as_slice() { - [flag, path] if flag == "--config" => PathBuf::from(path), + if args + .first() + .is_some_and(|value| value == "--admit-caller-config") + { + if let Err(error) = admit_caller_config(&args[1..]) { + eprintln!("codex-connector caller admission failed: {error}"); + std::process::exit(2); + } + return; + } + 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); } } +fn admit_caller_config(args: &[OsString]) -> Result<(), String> { + let [ + source, + destination, + caller_runtime_id, + connection_key_file, + connection_key_epoch, + allowed_models, + max_concurrent_requests, + max_payload_bytes, + max_output_tokens, + ] = args + else { + return Err("expected exactly nine caller-admission values".to_string()); + }; + let mut config = codex_connector::load_daemon_config(&PathBuf::from(source)) + .map_err(|error| error.to_string())?; + let caller_runtime_id = os_text(caller_runtime_id, "caller runtime")?; + let caller = CodexCallerConfig { + caller_runtime_id, + connection_key_file: PathBuf::from(connection_key_file), + connection_key_epoch: parse_number(connection_key_epoch, "connection key epoch")?, + allowed_models: parse_allowed_models(allowed_models)?, + max_concurrent_requests: parse_number(max_concurrent_requests, "max concurrent requests")?, + max_payload_bytes: parse_number(max_payload_bytes, "max payload bytes")?, + max_output_tokens: parse_number(max_output_tokens, "max output tokens")?, + }; + config + .admit_caller(caller) + .map_err(|error| error.to_string())?; + codex_connector::write_daemon_config(&PathBuf::from(destination), &config) + .map_err(|error| error.to_string()) +} + fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> { let [ path, @@ -65,7 +92,7 @@ fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> { caller_runtime_id, connection_key_file, connection_key_epoch, - allowed_model, + allowed_models, max_concurrent_requests, max_payload_bytes, max_output_tokens, @@ -83,7 +110,7 @@ fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> { caller_runtime_id: os_text(caller_runtime_id, "caller runtime")?, connection_key_file: PathBuf::from(connection_key_file), connection_key_epoch: parse_number(connection_key_epoch, "connection key epoch")?, - allowed_models: vec![os_text(allowed_model, "allowed model")?], + allowed_models: parse_allowed_models(allowed_models)?, max_concurrent_requests: parse_number( max_concurrent_requests, "max concurrent requests", @@ -104,6 +131,19 @@ fn os_text(value: &OsString, field: &str) -> Result { .ok_or_else(|| format!("{field} must be non-empty UTF-8")) } +fn parse_allowed_models(value: &OsString) -> Result, String> { + let raw = os_text(value, "allowed models")?; + let mut seen = HashSet::new(); + let mut models = Vec::new(); + for model in raw.split(',') { + if model.is_empty() || model.trim() != model || !seen.insert(model) { + return Err("allowed models must be unique non-empty comma-separated IDs".to_string()); + } + models.push(model.to_string()); + } + Ok(models) +} + fn parse_number(value: &OsString, field: &str) -> Result where T: std::str::FromStr, @@ -113,6 +153,27 @@ where .map_err(|_| format!("{field} is not a valid integer")) } +#[cfg(test)] +mod tests { + use super::parse_allowed_models; + use std::ffi::OsString; + + #[test] + fn model_allowlist_parser_preserves_one_or_many_exact_models() { + assert_eq!( + parse_allowed_models(&OsString::from("gpt-5.6-luna")).unwrap(), + vec!["gpt-5.6-luna"] + ); + assert_eq!( + parse_allowed_models(&OsString::from("gpt-5.6-luna,gpt-5.6-terra")).unwrap(), + vec!["gpt-5.6-luna", "gpt-5.6-terra"] + ); + assert!(parse_allowed_models(&OsString::from("gpt-5.6-luna,")).is_err()); + assert!(parse_allowed_models(&OsString::from("gpt-5.6-luna, gpt-5.6-terra")).is_err()); + assert!(parse_allowed_models(&OsString::from("gpt-5.6-luna,gpt-5.6-luna")).is_err()); + } +} + fn parse_sha256(value: &OsString) -> Result<[u8; 32], String> { let value = os_text(value, "Codex SHA-256")?; if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) { diff --git a/src/provider_backend.rs b/src/provider_backend.rs index 91c4870..8873708 100644 --- a/src/provider_backend.rs +++ b/src/provider_backend.rs @@ -72,7 +72,11 @@ impl CodexProviderBackend { let max_result_bytes = config.max_result_bytes; let http: ureq::Agent = ureq::Agent::config_builder() .timeout_connect(Some(Duration::from_secs(30))) - .timeout_recv_response(Some(Duration::from_secs(30))) + // ureq carries the receive-response deadline into streamed body + // reads as a preceding phase. Keep both receive phases on the + // same provider window so a 30-second header bound cannot silently + // cap a valid long-running SSE response. + .timeout_recv_response(Some(Duration::from_secs(300))) .timeout_recv_body(Some(Duration::from_secs(300))) .max_redirects(0) .http_status_as_error(false) @@ -500,7 +504,7 @@ fn parse_responses_sse(reader: impl Read) -> Result Result &'static str { +fn provider_sse_read_error(error: &io::Error) -> &'static str { + if let Some(ureq_error) = error + .get_ref() + .and_then(|source| source.downcast_ref::()) + { + return match ureq_error { + ureq::Error::Timeout(_) => "provider SSE read timed out", + ureq::Error::Protocol(_) => "provider SSE protocol failed", + ureq::Error::Tls(_) | ureq::Error::Rustls(_) => "provider SSE secure transport failed", + ureq::Error::BodyStalled => "provider SSE body stalled", + ureq::Error::Io(inner) => provider_sse_io_error(inner.kind()), + _ => "provider SSE read failed", + }; + } + provider_sse_io_error(error.kind()) +} + +fn provider_sse_io_error(kind: io::ErrorKind) -> &'static str { match kind { io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "provider SSE read timed out", io::ErrorKind::ConnectionReset @@ -614,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, @@ -676,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 { @@ -1057,6 +1086,14 @@ mod tests { } } + struct WrappedFailingReader(Option); + + impl Read for WrappedFailingReader { + fn read(&mut self, _buf: &mut [u8]) -> io::Result { + Err(io::Error::other(self.0.take().unwrap())) + } + } + fn jwt(payload: Value) -> String { let payload = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload).unwrap()); format!("e30.{payload}.signature") @@ -1257,6 +1294,10 @@ mod tests { parse_responses_sse(FailingReader(io::ErrorKind::Other)).err(), Some("provider SSE read failed") ); + assert_eq!( + parse_responses_sse(WrappedFailingReader(Some(ureq::Error::BodyStalled))).err(), + Some("provider SSE body stalled") + ); } #[test]