From 07de82597da0e21dfac915ebce0ff217921ab461 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 24 Aug 2026 16:23:23 +0200 Subject: [PATCH] Add bounded standalone connector daemon --- Cargo.lock | 360 +++++++++++++++++++++++++++++++++- Cargo.toml | 5 + README.md | 13 +- src/daemon.rs | 418 ++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 12 +- src/main.rs | 16 ++ src/provider_backend.rs | 44 ++++- 7 files changed, 853 insertions(+), 15 deletions(-) create mode 100644 src/daemon.rs create mode 100644 src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 90f8929..ba66579 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -37,6 +37,21 @@ dependencies = [ "subtle", ] +[[package]] +name = "android_system_properties" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" +dependencies = [ + "libc", +] + +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + [[package]] name = "autocfg" version = "1.5.1" @@ -58,6 +73,12 @@ dependencies = [ "generic-array", ] +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + [[package]] name = "bytes" version = "1.12.1" @@ -80,6 +101,20 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "chrono" +version = "0.4.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" +dependencies = [ + "iana-time-zone", + "js-sys", + "num-traits", + "serde", + "wasm-bindgen", + "windows-link", +] + [[package]] name = "cipher" version = "0.4.4" @@ -96,6 +131,7 @@ version = "0.1.0" dependencies = [ "aes-gcm", "base64", + "cultcache-rs", "rand_core", "rmp-serde", "serde", @@ -106,6 +142,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -135,6 +177,34 @@ dependencies = [ "cipher", ] +[[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", + "fs2", + "redb", + "rmp-serde", + "serde", + "serde_bytes", + "serde_json", + "uuid", + "windows-sys 0.59.0", +] + +[[package]] +name = "cultcache-rs-derive" +version = "0.1.0" +source = "git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c#fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "digest" version = "0.10.7" @@ -151,6 +221,40 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -172,6 +276,17 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + [[package]] name = "ghash" version = "0.5.1" @@ -198,6 +313,30 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + [[package]] name = "inout" version = "0.1.4" @@ -213,6 +352,17 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "js-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + [[package]] name = "libc" version = "0.2.189" @@ -258,6 +408,12 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "polyval" version = "0.6.2" @@ -288,13 +444,28 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + [[package]] name = "rand_core" version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ - "getrandom", + "getrandom 0.2.17", +] + +[[package]] +name = "redb" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de6c3b63e007e90ce536ec2ae4690826136a20ec8dbbbb400daef1bb999d2e36" +dependencies = [ + "libc", ] [[package]] @@ -305,10 +476,10 @@ checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" dependencies = [ "cc", "cfg-if", - "getrandom", + "getrandom 0.2.17", "libc", "untrusted", - "windows-sys", + "windows-sys 0.52.0", ] [[package]] @@ -365,6 +536,12 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + [[package]] name = "serde" version = "1.0.229" @@ -375,6 +552,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde_bytes" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8" +dependencies = [ + "serde", + "serde_core", +] + [[package]] name = "serde_core" version = "1.0.229" @@ -392,7 +579,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.4", ] [[package]] @@ -425,12 +612,29 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "subtle" version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "3.0.4" @@ -459,7 +663,7 @@ checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.4", ] [[package]] @@ -524,6 +728,17 @@ version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e" +[[package]] +name = "uuid" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f053576934f05a761a402421fbbe3d425d9366f75f978806a037b3ca481abecc" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "version_check" version = "0.9.5" @@ -536,6 +751,51 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasm-bindgen" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 2.0.119", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" +dependencies = [ + "unicode-ident", +] + [[package]] name = "webpki-roots" version = "1.0.9" @@ -545,6 +805,87 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.52.0" @@ -554,6 +895,15 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-targets" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index f681ec3..9d98e49 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,9 +11,14 @@ repository = "https://github.com/GameCult/CodexConnector" [lib] path = "src/lib.rs" +[[bin]] +name = "codex-connector" +path = "src/main.rs" + [dependencies] aes-gcm = "0.10" base64 = "0.23" +cultcache-rs = { git = "https://github.com/GameCult/cultcache-rs.git", rev = "fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c" } rand_core = { version = "0.6", features = ["getrandom"] } rmp-serde = "1" serde = { version = "1", features = ["derive"] } diff --git a/README.md b/README.md index cb384a5..963ff80 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,12 @@ child's private auth RPC completes, then performs the exact Responses request. It owns no prompt policy or decision authority. The current source establishes the v2 typed contract, its transport-neutral -service core, and the private Codex-auth/provider backend before adding the -public listener shell. The core admits distinct caller keys and policies, +service core, the private Codex-auth/provider backend, and one public daemon +entrypoint. The daemon reads one typed CultCache configuration document, binds +only loopback, uses CultNet's four-byte big-endian direct-pipe framing around +the encrypted MessagePack envelope, bounds connection threads and frame sizes, +and keeps provider execution outside the service-state lock. The core admits +distinct caller keys and policies, returns an exact cached response for byte-identical retries, refuses conflicting or concurrent reuse of one caller/request identity, and leaves provider execution outside its lock. The backend verifies the exact child binary and @@ -27,6 +31,11 @@ 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. +This is not yet a deployable replacement for the live connector. Persistent +replay recovery, redacted CultMesh/Odin publication, shared consumer client +cuts, and the independent Idunn target remain open. The old daemon remains the +sole live credential writer until those proofs land. + ## Contract invariant Each consumer internally derives an exact typed Codex provider request. The diff --git a/src/daemon.rs b/src/daemon.rs new file mode 100644 index 0000000..acca6c6 --- /dev/null +++ b/src/daemon.rs @@ -0,0 +1,418 @@ +use std::collections::HashSet; +use std::fs; +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use zeroize::Zeroizing; + +use crate::{ + CodexAppServerConfig, CodexCallerAdmission, CodexProviderBackend, CodexProviderBackendError, + CodexTransportAdmission, CodexTransportEnvelope, CodexTransportService, ServiceError, +}; + +const CONFIG_EPOCH: u32 = 1; +const CONFIG_KEY: &str = "runtime"; +const FRAME_OVERHEAD_BUDGET: usize = 4096; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct CodexCallerConfig { + pub caller_runtime_id: String, + pub connection_key_file: PathBuf, + pub allowed_models: Vec, + pub max_concurrent_requests: usize, + pub max_payload_bytes: usize, + pub max_output_tokens: u32, +} + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "gamecult.codex.connector_config.v1")] +pub struct CodexDaemonConfig { + #[cultcache(key = 0)] + pub epoch: u32, + #[cultcache(key = 1)] + pub bind: String, + #[cultcache(key = 2)] + pub codex_executable: PathBuf, + #[cultcache(key = 3)] + pub codex_executable_sha256: [u8; 32], + #[cultcache(key = 4)] + pub codex_home: PathBuf, + #[cultcache(key = 5)] + pub max_frame_bytes: usize, + #[cultcache(key = 6)] + pub max_connections: usize, + #[cultcache(key = 7)] + pub socket_timeout_ms: u64, + #[cultcache(key = 8)] + pub max_expiry_skew_ms: u64, + #[cultcache(key = 9)] + pub callers: Vec, +} + +impl CodexDaemonConfig { + pub fn validate(&self) -> Result { + if self.epoch != CONFIG_EPOCH { + return Err(CodexDaemonError::InvalidConfig("epoch")); + } + let bind = self + .bind + .parse::() + .map_err(|_| CodexDaemonError::InvalidConfig("bind"))?; + if !bind.ip().is_loopback() { + return Err(CodexDaemonError::InvalidConfig("bind")); + } + if self.codex_executable.as_os_str().is_empty() + || self.codex_executable_sha256 == [0; 32] + || self.codex_home.as_os_str().is_empty() + || self.max_frame_bytes < FRAME_OVERHEAD_BUDGET * 2 + || self.max_frame_bytes > u32::MAX as usize + || self.max_connections == 0 + || self.socket_timeout_ms == 0 + || self.max_expiry_skew_ms == 0 + || self.callers.is_empty() + { + return Err(CodexDaemonError::InvalidConfig("runtime bounds")); + } + let mut caller_ids = HashSet::new(); + for caller in &self.callers { + 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.allowed_models.is_empty() + || caller + .allowed_models + .iter() + .any(|model| model.trim().is_empty() || model.trim() != model) + || caller.max_concurrent_requests == 0 + || caller.max_payload_bytes < 4096 + || caller.max_payload_bytes > self.max_frame_bytes - FRAME_OVERHEAD_BUDGET + || caller.max_output_tokens == 0 + { + return Err(CodexDaemonError::InvalidConfig("caller admission")); + } + } + Ok(bind) + } +} + +pub fn load_daemon_config(path: &Path) -> Result { + let mut cache = CultCache::new(); + cache + .register_entry_type::() + .map_err(config_store_error)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path)); + cache + .pull_all_backing_stores() + .map_err(config_store_error)?; + let config = cache + .get_required::(CONFIG_KEY) + .map_err(config_store_error)?; + config.validate()?; + Ok(config) +} + +pub fn write_daemon_config( + path: &Path, + config: &CodexDaemonConfig, +) -> Result<(), CodexDaemonError> { + config.validate()?; + let mut cache = CultCache::new(); + cache + .register_entry_type::() + .map_err(config_store_error)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path)); + cache + .pull_all_backing_stores() + .map_err(config_store_error)?; + cache.put(CONFIG_KEY, config).map_err(config_store_error)?; + 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::new( + load_admissions(&config)?, + config.max_expiry_skew_ms, + )?)); + let backend = Arc::new(CodexProviderBackend::start(CodexAppServerConfig { + executable: config.codex_executable.clone(), + executable_sha256: config.codex_executable_sha256, + codex_home: config.codex_home.clone(), + max_result_bytes: config.max_frame_bytes - FRAME_OVERHEAD_BUDGET, + })?); + let readiness = backend.readiness()?; + let listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?; + let active_connections = Arc::new(AtomicUsize::new(0)); + eprintln!( + "codex-connector ready at {} for {} callers using {:?}", + listener.local_addr().map_err(CodexDaemonError::Listen)?, + config.callers.len(), + readiness.auth_mode + ); + + for accepted in listener.incoming() { + let stream = match accepted { + Ok(stream) => stream, + Err(error) => return Err(CodexDaemonError::Listen(error)), + }; + if !stream + .peer_addr() + .map_err(CodexDaemonError::Connection)? + .ip() + .is_loopback() + { + continue; + } + let Some(permit) = + ConnectionPermit::acquire(active_connections.clone(), config.max_connections) + else { + continue; + }; + let service = service.clone(); + let backend = backend.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); + }) + .map_err(CodexDaemonError::Thread)?; + } + Err(CodexDaemonError::ListenerClosed) +} + +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']); + CodexCallerAdmission::new( + caller.caller_runtime_id.clone(), + secret.to_string(), + caller.allowed_models.clone(), + caller.max_concurrent_requests, + caller.max_payload_bytes, + caller.max_output_tokens, + ) + .map_err(CodexDaemonError::Service) + }) + .collect() +} + +fn serve_connection( + mut stream: TcpStream, + service: &Mutex, + backend: &CodexProviderBackend, + max_frame_bytes: usize, + timeout: Duration, +) -> Result<(), CodexDaemonError> { + stream + .set_read_timeout(Some(timeout)) + .and_then(|()| stream.set_write_timeout(Some(timeout))) + .map_err(CodexDaemonError::Connection)?; + let request = read_frame(&mut stream, max_frame_bytes)?; + let envelope: CodexTransportEnvelope = + rmp_serde::from_slice(&request).map_err(|_| CodexDaemonError::FrameEncoding)?; + let admission = service + .lock() + .map_err(|_| CodexDaemonError::ServicePoisoned)? + .begin(&envelope, unix_ms()?)?; + let response = match admission { + CodexTransportAdmission::Reply(response) => response, + CodexTransportAdmission::Execute(claim) => { + let result = backend.execute(claim.invocation()); + service + .lock() + .map_err(|_| CodexDaemonError::ServicePoisoned)? + .complete(*claim, result)? + } + }; + let response = rmp_serde::to_vec(&response).map_err(|_| CodexDaemonError::FrameEncoding)?; + write_frame(&mut stream, &response, max_frame_bytes) +} + +fn read_frame(reader: &mut impl Read, max_frame_bytes: usize) -> Result, CodexDaemonError> { + let mut length = [0_u8; 4]; + reader + .read_exact(&mut length) + .map_err(CodexDaemonError::Connection)?; + let length = u32::from_be_bytes(length) as usize; + if length == 0 || length > max_frame_bytes { + return Err(CodexDaemonError::FrameSize); + } + let mut payload = vec![0_u8; length]; + reader + .read_exact(&mut payload) + .map_err(CodexDaemonError::Connection)?; + Ok(payload) +} + +fn write_frame( + writer: &mut impl Write, + payload: &[u8], + max_frame_bytes: usize, +) -> Result<(), CodexDaemonError> { + if payload.is_empty() || payload.len() > max_frame_bytes || payload.len() > u32::MAX as usize { + return Err(CodexDaemonError::FrameSize); + } + writer + .write_all(&(payload.len() as u32).to_be_bytes()) + .and_then(|()| writer.write_all(payload)) + .and_then(|()| writer.flush()) + .map_err(CodexDaemonError::Connection) +} + +fn unix_ms() -> Result { + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| CodexDaemonError::Clock)? + .as_millis(); + u64::try_from(millis).map_err(|_| CodexDaemonError::Clock) +} + +fn config_store_error(error: impl std::fmt::Display) -> CodexDaemonError { + CodexDaemonError::ConfigStore(error.to_string()) +} + +struct ConnectionPermit { + active: Arc, +} + +impl ConnectionPermit { + fn acquire(active: Arc, limit: usize) -> Option { + let mut observed = active.load(Ordering::Acquire); + loop { + if observed >= limit { + return None; + } + match active.compare_exchange_weak( + observed, + observed + 1, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => return Some(Self { active }), + Err(current) => observed = current, + } + } + } +} + +impl Drop for ConnectionPermit { + fn drop(&mut self) { + self.active.fetch_sub(1, Ordering::AcqRel); + } +} + +#[derive(Debug, Error)] +pub enum CodexDaemonError { + #[error("connector config store failed: {0}")] + ConfigStore(String), + #[error("invalid connector config field {0}")] + InvalidConfig(&'static str), + #[error("connector connection key is unavailable")] + ConnectionKey(#[source] std::io::Error), + #[error(transparent)] + Service(#[from] ServiceError), + #[error(transparent)] + Provider(#[from] CodexProviderBackendError), + #[error("connector service lock was poisoned")] + ServicePoisoned, + #[error("connector listener failed")] + Listen(#[source] std::io::Error), + #[error("connector request thread failed")] + Thread(#[source] std::io::Error), + #[error("connector connection failed")] + Connection(#[source] std::io::Error), + #[error("connector frame exceeded its bound")] + FrameSize, + #[error("connector frame was not valid MessagePack")] + FrameEncoding, + #[error("system clock is before the Unix epoch")] + Clock, + #[error("connector listener closed")] + ListenerClosed, +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + + fn config(root: &Path) -> CodexDaemonConfig { + CodexDaemonConfig { + epoch: CONFIG_EPOCH, + bind: "127.0.0.1:4103".to_string(), + codex_executable: root.join("codex"), + codex_executable_sha256: [3; 32], + codex_home: root.join("codex-home"), + max_frame_bytes: 1024 * 1024, + max_connections: 16, + socket_timeout_ms: 30_000, + max_expiry_skew_ms: 300_000, + callers: vec![CodexCallerConfig { + caller_runtime_id: "epiphany-yggdrasil".to_string(), + connection_key_file: root.join("epiphany.key"), + allowed_models: vec!["gpt-5.4".to_string()], + max_concurrent_requests: 4, + max_payload_bytes: 512 * 1024, + max_output_tokens: 32_768, + }], + } + } + + #[test] + fn typed_cultcache_config_round_trips_and_refuses_epoch_substitution() { + let root = std::env::temp_dir().join(format!( + "codex-connector-config-{}-{}", + std::process::id(), + unix_ms().unwrap() + )); + fs::create_dir_all(&root).unwrap(); + let path = root.join("connector.cc"); + let expected = config(&root); + write_daemon_config(&path, &expected).unwrap(); + assert_eq!(load_daemon_config(&path).unwrap(), expected); + + let mut stale = expected; + stale.epoch = 0; + assert!(matches!( + write_daemon_config(&path, &stale), + Err(CodexDaemonError::InvalidConfig("epoch")) + )); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn direct_pipe_frame_is_big_endian_and_refuses_oversize_before_allocation() { + let mut encoded = Vec::new(); + write_frame(&mut encoded, b"typed", 16).unwrap(); + assert_eq!(&encoded[..4], &5_u32.to_be_bytes()); + assert_eq!(read_frame(&mut Cursor::new(encoded), 16).unwrap(), b"typed"); + + let oversized = 17_u32.to_be_bytes().to_vec(); + assert!(matches!( + read_frame(&mut Cursor::new(oversized), 16), + Err(CodexDaemonError::FrameSize) + )); + } +} diff --git a/src/lib.rs b/src/lib.rs index 478c719..99c24c2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,9 +6,17 @@ use rand_core::{OsRng, RngCore}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use thiserror::Error; +use zeroize::Zeroizing; +mod daemon; mod provider_backend; +pub use daemon::CodexCallerConfig; +pub use daemon::CodexDaemonConfig; +pub use daemon::CodexDaemonError; +pub use daemon::load_daemon_config; +pub use daemon::serve; +pub use daemon::write_daemon_config; pub use provider_backend::CodexAppServerConfig; pub use provider_backend::CodexAuthMode; pub use provider_backend::CodexAuthReadiness; @@ -782,8 +790,8 @@ impl CodexCallerAdmission { { return Err(ServiceError::InvalidAdmission); } - let connection_key = connection_key.into(); - let security = CodexTransportKey::from_connection_secret(&connection_key)?; + let connection_key = Zeroizing::new(connection_key.into()); + let security = CodexTransportKey::from_connection_secret(connection_key.as_str())?; Ok(Self { caller_runtime_id, security, diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..179b922 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,16 @@ +use std::path::PathBuf; + +fn main() { + let mut args = std::env::args_os().skip(1); + let config = match (args.next(), args.next(), args.next()) { + (Some(flag), Some(path), None) if flag == "--config" => PathBuf::from(path), + _ => { + eprintln!("usage: codex-connector --config PATH.cc"); + std::process::exit(2); + } + }; + if let Err(error) = codex_connector::serve(&config) { + eprintln!("codex-connector stopped: {error}"); + std::process::exit(1); + } +} diff --git a/src/provider_backend.rs b/src/provider_backend.rs index 83e1ae0..4f922e3 100644 --- a/src/provider_backend.rs +++ b/src/provider_backend.rs @@ -37,6 +37,7 @@ pub struct CodexAppServerConfig { pub executable: PathBuf, pub executable_sha256: [u8; 32], pub codex_home: PathBuf, + pub max_result_bytes: usize, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -56,10 +57,17 @@ pub struct CodexAuthReadiness { pub struct CodexProviderBackend { authority: Mutex, http: ureq::Agent, + max_result_bytes: usize, } impl CodexProviderBackend { pub fn start(config: CodexAppServerConfig) -> Result { + if config.max_result_bytes < 4096 { + return Err(CodexProviderBackendError::InvalidConfiguration( + "max_result_bytes", + )); + } + 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))) @@ -70,6 +78,7 @@ impl CodexProviderBackend { Ok(Self { authority: Mutex::new(CodexAppServerAuthority::spawn(config)?), http, + max_result_bytes, }) } @@ -82,7 +91,23 @@ impl CodexProviderBackend { Ok(credential.redacted(&authority.app_server_user_agent)) } - pub fn execute( + pub fn execute(&self, invocation: &CodexTransportInvocation) -> CodexTransportResult { + match self.try_execute(invocation) { + Ok(result) => result, + Err(CodexProviderBackendError::ProviderRequest) => failed_result( + invocation, + "provider_request", + "provider request could not be rendered", + ), + Err(_) => failed_result( + invocation, + "credential_authority", + "Codex credential authority failed", + ), + } + } + + fn try_execute( &self, invocation: &CodexTransportInvocation, ) -> Result { @@ -168,11 +193,18 @@ impl CodexProviderBackend { transport: TRANSPORT_ID.to_string(), outcome: parsed.outcome, }; - Ok(CodexTransportResult::transported( - invocation, - parsed.events, - receipt, - )) + let result = CodexTransportResult::transported(invocation, parsed.events, receipt); + if rmp_serde::to_vec(&result) + .map(|bytes| bytes.len() > self.max_result_bytes) + .unwrap_or(true) + { + return Ok(failed_result( + invocation, + "result_size", + "provider result exceeded its byte bound", + )); + } + Ok(result) } }