From 78b4331ee58cd418c2fa1b872bc716a76dd0150b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 3 Sep 2026 09:27:00 +0200 Subject: [PATCH] Expose managed Connector route presence --- src/daemon.rs | 217 ++++++++++++++++++++++++++++++++++++++++++-- src/idunn_health.rs | 143 +++++++++++++++++++++++++---- 2 files changed, 336 insertions(+), 24 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index 461c00f..c7287bc 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -15,6 +15,12 @@ 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, @@ -321,8 +327,11 @@ pub fn serve( config.max_expiry_skew_ms, )?)); #[cfg(target_os = "linux")] - if let Some(publisher) = prepared_presence_publisher { - start_periodic_runtime_presence_publisher(publisher, backend.clone())?; + 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!( @@ -349,13 +358,23 @@ pub fn serve( }; 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)?; } @@ -379,11 +398,13 @@ fn prepare_runtime_presence_publisher( #[cfg(target_os = "linux")] fn start_periodic_runtime_presence_publisher( - mut publisher: crate::idunn_health::RuntimePresencePublisher, + publisher: Arc>, backend: Arc, ) -> Result<(), CodexDaemonError> { let (state, detail) = provider_health_observation(&backend); publisher + .lock() + .map_err(|_| CodexDaemonError::RuntimePresencePoisoned)? .publish(state, detail) .map_err(CodexDaemonError::Health)?; thread::Builder::new() @@ -392,7 +413,11 @@ fn start_periodic_runtime_presence_publisher( loop { thread::sleep(Duration::from_secs(10)); let (state, detail) = provider_health_observation(&backend); - if let Err(error) = publisher.publish(state, detail) { + 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}"); } } @@ -568,6 +593,9 @@ 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> { @@ -576,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)? @@ -592,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 { @@ -669,6 +763,12 @@ pub enum CodexDaemonError { #[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")] @@ -682,6 +782,8 @@ pub enum CodexDaemonError { #[cfg(test)] mod tests { use super::*; + #[cfg(target_os = "linux")] + use std::cell::Cell; use std::io::Cursor; fn config(root: &Path) -> CodexDaemonConfig { @@ -721,6 +823,105 @@ mod tests { ); } + #[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!( diff --git a/src/idunn_health.rs b/src/idunn_health.rs index f871b68..572a51f 100644 --- a/src/idunn_health.rs +++ b/src/idunn_health.rs @@ -32,7 +32,7 @@ const SYSTEMD_LISTEN_FDS_START: RawFd = 3; const ACTIVATION_SIGNER_FD_NAME: &str = IDUNN_RUNTIME_ACTIVATION_CREDENTIAL_NAME; const PROVIDER_SIGNER_FD_NAME: &str = "gamecult-runtime-presence-identity"; -const CONNECTOR_TARGET: &str = "codex-connector"; +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"; @@ -219,6 +219,48 @@ impl RuntimePresencePublisher { .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()?; @@ -627,21 +669,13 @@ fn publish_presence( "codex-connector-presence:{}:{}:{}", target, record.runtime_instance_id, record.publisher_sequence ), - document: 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, + 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("runtime-presence-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" @@ -688,6 +722,30 @@ fn publish_presence( Ok(()) } +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(), @@ -1104,6 +1162,7 @@ mod tests { 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, @@ -1113,7 +1172,7 @@ mod tests { activation: activation.clone(), activation_sha256: activation.canonical_sha256()?, warming_presence_sha256: warming.canonical_sha256()?, - lease_sha256: sha('8'), + lease_sha256: lease_sha256.clone(), }), }); let active = publisher.signed_record("active", "ready", issued_at)?; @@ -1138,6 +1197,58 @@ mod tests { }, )?; 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, 3); + 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)