diff --git a/AGENTS.md b/AGENTS.md index 00424b9..2b6608b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,12 +16,6 @@ repair the store. Consumers own native requests, provider lowering, prompts, schemas, tool execution, retry between passes, interpretation, decisions, and state admission. -Keep the feature boundary exact. `--no-default-features` is the contract-only -surface used where a Mind seals provider request bytes. The default `client` -surface adds only authenticated framing and socket transport. `daemon` adds the -service authorities. Contract-only consumers must not compile client crypto or -socket physiology by accident. - Do not link upstream Codex crates into this package. Idunn freezes the official Codex binary as an exact package input. The child receives auth RPC only, never consumer identity, prompts, tools, provider requests, or model output. 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 dea5c3f..e4087ba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,39 +17,31 @@ path = "src/main.rs" required-features = ["daemon"] [features] -default = ["client"] -client = [ - "dep:aes-gcm", - "dep:rand_core", - "dep:zeroize", -] +default = [] daemon = [ - "client", "dep:anyhow", "dep:chrono", "dep:cultcache-rs", "dep:cultnet-rs", - "dep:ed25519-dalek", - "dep:serde_bytes", + "dep:fs2", + "dep:libc", "dep:ureq", - "dep:uuid", ] [dependencies] -aes-gcm = { version = "0.10", optional = true } +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 } -rand_core = { version = "0.6", features = ["getrandom"], 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 = { version = "1", optional = true } +zeroize = "1" diff --git a/README.md b/README.md index f1738e5..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 @@ -12,13 +20,11 @@ one credential store. The public connector reads the credential only after the child's private auth RPC completes, then performs the exact Responses request. It owns no prompt policy or decision authority. -The library has three explicit compile surfaces. With no features it exposes -only the v2 provider/result contracts and their deterministic validation. The -default `client` feature adds encryption, bounded direct-pipe framing, and the -one-request socket client. It still compiles no CultCache, HTTP, TLS, provider -auth, replay, or daemon code. The `daemon` feature includes the client surface -and the service. Epiphany and Ghostlight can therefore type and seal their exact -provider requests without importing transport physiology into their Minds. +The default library is the shared consumer ABI: v2 typed contracts, exact +provider lowering, encryption, bounded direct-pipe framing, one-request socket +transport, and exact result validation. It does not compile CultCache, HTTP, +TLS, provider auth, replay, or daemon code. Epiphany and Ghostlight consume this +surface instead of copying wire law. Connection establishment is bounded independently. Response read/write timeout is optional so a consumer's typed outer pass may remain the sole deadline owner. @@ -47,21 +53,62 @@ No upstream Codex crate is linked into the package. The official binary is a pinned deployment input, keeping Codex's application build graph outside both consumers and this small daemon. -Focused verification proves each real build surface independently: +Focused verification uses the two real build surfaces: ```text -cargo test --lib --no-default-features cargo test --lib cargo test --lib --features daemon 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 3769b5b..d6e6b23 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,9 +1,6 @@ use std::collections::HashSet; -#[cfg(feature = "client")] use std::io::{Read, Write}; -#[cfg(feature = "client")] use std::net::{SocketAddr, TcpStream}; -#[cfg(feature = "client")] use std::time::Duration; #[cfg(feature = "daemon")] @@ -11,18 +8,14 @@ use std::collections::HashMap; #[cfg(feature = "daemon")] use std::path::Path; -#[cfg(feature = "client")] use aes_gcm::aead::{Aead, Payload}; -#[cfg(feature = "client")] use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; #[cfg(feature = "daemon")] use cultcache_rs::{CultCache, DatabaseEntry, OwnedRedbMessagePackBackingStore}; -#[cfg(feature = "client")] use rand_core::{OsRng, RngCore}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use thiserror::Error; -#[cfg(feature = "client")] use zeroize::{Zeroize, Zeroizing}; #[cfg(feature = "daemon")] @@ -44,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")] @@ -66,11 +54,9 @@ pub const RESULT_SCHEMA_ID: &str = "gamecult.codex.transport_result.v2"; pub const RECEIPT_SCHEMA_ID: &str = "gamecult.codex.transport_receipt.v2"; pub const ENVELOPE_SCHEMA_ID: &str = "gamecult.codex.transport_envelope.v2"; -#[cfg(feature = "client")] #[derive(Clone, PartialEq, Eq)] pub struct CodexTransportKey([u8; 32]); -#[cfg(feature = "client")] impl CodexTransportKey { pub fn from_connection_secret(secret: &str) -> Result { if secret.trim().is_empty() || secret.trim() != secret { @@ -80,7 +66,6 @@ impl CodexTransportKey { } } -#[cfg(feature = "client")] impl Drop for CodexTransportKey { fn drop(&mut self) { self.0.zeroize(); @@ -644,14 +629,12 @@ pub fn canonical_invocation_bytes( rmp_serde::to_vec(invocation).map_err(|_| ContractError::Encoding) } -#[cfg(feature = "client")] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum CodexEnvelopeKind { Invocation, Result, } -#[cfg(feature = "client")] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct CodexTransportEnvelope { pub schema_id: String, @@ -662,7 +645,6 @@ pub struct CodexTransportEnvelope { pub ciphertext: Vec, } -#[cfg(feature = "client")] pub fn encrypt_invocation( invocation: &CodexTransportInvocation, security: &CodexTransportKey, @@ -677,7 +659,6 @@ pub fn encrypt_invocation( ) } -#[cfg(feature = "client")] pub fn decrypt_invocation( envelope: &CodexTransportEnvelope, security: &CodexTransportKey, @@ -694,7 +675,6 @@ pub fn decrypt_invocation( Ok(invocation) } -#[cfg(feature = "client")] pub fn encrypt_result( result: &CodexTransportResult, security: &CodexTransportKey, @@ -709,7 +689,6 @@ pub fn encrypt_result( ) } -#[cfg(feature = "client")] pub fn decrypt_result( envelope: &CodexTransportEnvelope, security: &CodexTransportKey, @@ -728,7 +707,6 @@ pub fn decrypt_result( Ok(result) } -#[cfg(feature = "client")] #[derive(Clone)] pub struct CodexConnectorClient { endpoint: SocketAddr, @@ -737,7 +715,6 @@ pub struct CodexConnectorClient { response_timeout: Option, } -#[cfg(feature = "client")] impl CodexConnectorClient { pub fn new( endpoint: SocketAddr, @@ -745,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()) { @@ -788,7 +767,45 @@ impl CodexConnectorClient { } } -#[cfg(feature = "client")] +#[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")] @@ -803,14 +820,12 @@ pub enum CodexConnectorClientError { Transport(#[from] ServiceError), } -#[cfg(feature = "client")] #[derive(Debug)] enum TransportFrameError { Connection(std::io::Error), Size, } -#[cfg(feature = "client")] fn client_frame_error(error: TransportFrameError) -> CodexConnectorClientError { match error { TransportFrameError::Connection(error) => CodexConnectorClientError::Connection(error), @@ -818,7 +833,6 @@ fn client_frame_error(error: TransportFrameError) -> CodexConnectorClientError { } } -#[cfg(feature = "client")] fn read_transport_frame( reader: &mut impl Read, max_frame_bytes: usize, @@ -838,7 +852,6 @@ fn read_transport_frame( Ok(payload) } -#[cfg(feature = "client")] fn write_transport_frame( writer: &mut impl Write, payload: &[u8], @@ -854,7 +867,6 @@ fn write_transport_frame( .map_err(TransportFrameError::Connection) } -#[cfg(feature = "client")] fn encrypt_message( caller_runtime_id: &str, request_id: &str, @@ -892,7 +904,6 @@ fn encrypt_message( }) } -#[cfg(feature = "client")] fn decrypt_message( envelope: &CodexTransportEnvelope, security: &CodexTransportKey, @@ -915,7 +926,6 @@ fn decrypt_message( .map_err(|_| ServiceError::Encryption) } -#[cfg(feature = "client")] fn envelope_aad( schema_id: &str, caller_runtime_id: &str, @@ -926,7 +936,6 @@ fn envelope_aad( .map_err(|_| ServiceError::Encoding) } -#[cfg(feature = "client")] fn validate_envelope( envelope: &CodexTransportEnvelope, expected_kind: CodexEnvelopeKind, @@ -1670,7 +1679,6 @@ mod tests { .unwrap() } - #[cfg(feature = "client")] fn security(key: &str) -> CodexTransportKey { CodexTransportKey::from_connection_secret(key).unwrap() } @@ -1682,7 +1690,6 @@ mod tests { CodexTransportInvocation::new(caller_runtime_id, 2_000, [7; 32], request).unwrap() } - #[cfg(feature = "client")] #[test] fn encrypted_invocation_hides_content_and_binds_outer_identity() { let invocation = invocation(); @@ -1711,7 +1718,6 @@ mod tests { ); } - #[cfg(feature = "client")] #[test] fn shared_client_transports_one_exact_encrypted_request_and_result() { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); 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 6776e86..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) @@ -631,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, @@ -693,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 {