diff --git a/epiphany-core/src/bin/epiphany-daemon-supervisor.rs b/epiphany-core/src/bin/epiphany-daemon-supervisor.rs index 8ca5f6fa..38d1aeaf 100644 --- a/epiphany-core/src/bin/epiphany-daemon-supervisor.rs +++ b/epiphany-core/src/bin/epiphany-daemon-supervisor.rs @@ -19,6 +19,7 @@ use epiphany_core::EpiphanyProcessObservation as ProcessObservation; use epiphany_core::MemorySemanticProjectionInput; use epiphany_core::agent_memory_semantic_projection_input; use epiphany_core::authenticate_epiphany_cultmesh_semantic_projector_launch; +use epiphany_core::authenticate_operator_command_service_health; use epiphany_core::authenticate_resident_provider_pair; use epiphany_core::epiphany_cultmesh_daemon_poke_intent_from_status; use epiphany_core::epiphany_cultmesh_daemon_poke_receipt_for_intent; @@ -37,6 +38,7 @@ use epiphany_core::load_latest_epiphany_cultmesh_daemon_service_lifecycle_receip use epiphany_core::migrate_memory_semantic_projection_attempts_v0; use epiphany_core::observe_native_process as observe_process; use epiphany_core::query_epiphany_local_verse_context; +use epiphany_core::read_operator_command_trust_anchor; use epiphany_core::retire_epiphany_cultmesh_operator_status_documents; use epiphany_core::retire_memory_semantic_projection_claims_v0; use epiphany_core::retire_orphaned_memory_semantic_projection_attempts_v0; @@ -950,6 +952,61 @@ fn publish_managed_service_iteration_health( contradictions .push("resident health requires both heartbeat and Self provider stores".into()); } + match ( + args.operator_service_readiness_store.as_deref(), + args.operator_executor_trust_anchor.as_deref(), + args.operator_service_bind.as_deref(), + ) { + (Some(store), Some(anchor_path), Some(bind)) => { + expected += 1; + let result = read_operator_command_trust_anchor(anchor_path).and_then(|anchor| { + authenticate_operator_command_service_health( + store, + &args.runtime_id, + &release.release_id, + authenticated_release_witness_sha256, + &release.source_commit_sha, + bind, + &anchor, + Utc::now().timestamp_millis().max(0) as u64, + 15_000, + ) + }); + match result { + Ok(health) => { + let expected_executable = + epiphany_packaged_release_binary_path(release, "operator-command") + .and_then(|path| fs::canonicalize(path).map_err(Into::into)); + let observed = capture_process_instance(health.process_id); + match (expected_executable, observed) { + (Ok(expected_executable), Ok(process)) + if process.creation_token == health.process_creation_token + && process.executable_path == expected_executable + && process.executable_path.display().to_string() + == health.process_executable_path => + { + terminal_current += 1 + } + (Ok(_), Ok(_)) => contradictions.push( + "operator service process identity is not the signed packaged sibling" + .into(), + ), + (Err(error), _) | (_, Err(error)) => contradictions.push(format!( + "operator service process cannot be authenticated: {error:#}" + )), + } + } + Err(error) => contradictions.push(format!( + "operator service signed readiness is invalid: {error:#}" + )), + } + } + (None, None, None) => {} + _ => { + expected += 1; + contradictions.push("operator service health requires readiness store, executor trust anchor, and exact bind".into()); + } + } for service_id in required { let Some(policy) = policies .iter() @@ -5118,6 +5175,9 @@ struct Args { resident_heartbeat_store: Option, resident_self_store: Option, resident_provider_stale_seconds: u64, + operator_service_readiness_store: Option, + operator_executor_trust_anchor: Option, + operator_service_bind: Option, } impl Args { @@ -5175,6 +5235,9 @@ impl Args { let mut resident_heartbeat_store = None; let mut resident_self_store = None; let mut resident_provider_stale_seconds = 180_u64; + let mut operator_service_readiness_store = None; + let mut operator_executor_trust_anchor = None; + let mut operator_service_bind = None; while let Some(arg) = values.next() { match arg.as_str() { @@ -5401,6 +5464,26 @@ impl Args { anyhow::bail!("--resident-provider-stale-seconds must be positive"); } } + "--operator-service-readiness-store" => { + operator_service_readiness_store = + Some(PathBuf::from(values.next().context( + "missing --operator-service-readiness-store value", + )?)) + } + "--operator-executor-trust-anchor" => { + operator_executor_trust_anchor = Some(PathBuf::from( + values + .next() + .context("missing --operator-executor-trust-anchor value")?, + )) + } + "--operator-service-bind" => { + operator_service_bind = Some( + values + .next() + .context("missing --operator-service-bind value")?, + ) + } other => anyhow::bail!("unknown argument {other:?}"), } } @@ -5511,6 +5594,18 @@ impl Args { idunn_health_contract.as_deref(), idunn_deployment_request_id.as_deref(), )?; + let operator_health_fields = [ + operator_service_readiness_store.is_some(), + operator_executor_trust_anchor.is_some(), + operator_service_bind.is_some(), + ]; + if operator_health_fields.into_iter().any(|value| value) + && !operator_health_fields.into_iter().all(|value| value) + { + anyhow::bail!( + "operator service health store, executor trust anchor, and bind are all-or-none" + ); + } Ok(Self { command, @@ -5564,6 +5659,9 @@ impl Args { resident_heartbeat_store, resident_self_store, resident_provider_stale_seconds, + operator_service_readiness_store, + operator_executor_trust_anchor, + operator_service_bind, }) } } diff --git a/epiphany-core/src/bin/epiphany-host-identity.rs b/epiphany-core/src/bin/epiphany-host-identity.rs index af0ce42b..4e53f98d 100644 --- a/epiphany-core/src/bin/epiphany-host-identity.rs +++ b/epiphany-core/src/bin/epiphany-host-identity.rs @@ -1,7 +1,7 @@ use anyhow::{Result, anyhow}; use epiphany_core::{ - default_host_identity_store_path, enroll_default_host_identity, - export_host_identity_trust_anchor, open_default_host_identity, + default_host_identity_store_path, enroll_default_host_identity, enroll_host_identity_at, + export_host_identity_trust_anchor, open_default_host_identity, open_host_identity_at, }; use serde_json::json; @@ -9,19 +9,36 @@ fn main() -> Result<()> { let command = std::env::args().nth(1).ok_or_else(|| { anyhow!("usage: epiphany-host-identity [output]") })?; + let values = std::env::args().skip(2).collect::>(); + let store_index = values.iter().position(|value| value == "--store"); + let explicit_store = match store_index { + Some(index) if index + 2 == values.len() => { + Some(std::path::PathBuf::from(&values[index + 1])) + } + Some(_) => return Err(anyhow!("--store must be the final option with one path")), + None => None, + }; + let positional = &values[..store_index.unwrap_or(values.len())]; if matches!( command.as_str(), "enroll-trust-anchor" | "export-trust-anchor" ) { - let output = std::env::args() - .nth(2) - .ok_or_else(|| anyhow!("{command} requires an output path"))?; + let output = positional + .first() + .filter(|_| positional.len() == 1) + .ok_or_else(|| anyhow!("{command} requires one output path"))?; let signer = if command == "enroll-trust-anchor" { - enroll_default_host_identity()? + match explicit_store.as_deref() { + Some(store) => enroll_host_identity_at(store)?, + None => enroll_default_host_identity()?, + } } else { - open_default_host_identity()? + match explicit_store.as_deref() { + Some(store) => open_host_identity_at(store)?, + None => open_default_host_identity()?, + } }; - let anchor = export_host_identity_trust_anchor(&signer, std::path::Path::new(&output))?; + let anchor = export_host_identity_trust_anchor(&signer, std::path::Path::new(output))?; println!( "{}", serde_json::to_string_pretty(&json!({ @@ -33,9 +50,18 @@ fn main() -> Result<()> { ); return Ok(()); } + if !positional.is_empty() { + return Err(anyhow!("unexpected host identity positional argument")); + } let signer = match command.as_str() { - "enroll" => enroll_default_host_identity()?, - "status" => open_default_host_identity()?, + "enroll" => match explicit_store.as_deref() { + Some(store) => enroll_host_identity_at(store)?, + None => enroll_default_host_identity()?, + }, + "status" => match explicit_store.as_deref() { + Some(store) => open_host_identity_at(store)?, + None => open_default_host_identity()?, + }, _ => { return Err(anyhow!( "usage: epiphany-host-identity [output]" @@ -51,7 +77,7 @@ fn main() -> Result<()> { "assurance": entry.assurance, "protectorKind": entry.protector_kind, "createdAt": entry.created_at, - "store": default_host_identity_store_path()?, + "store": explicit_store.unwrap_or(default_host_identity_store_path()?), "privateStateExposed": false }))? ); diff --git a/epiphany-core/src/bin/epiphany-operator-command.rs b/epiphany-core/src/bin/epiphany-operator-command.rs index d47359c7..bc49f7d2 100644 --- a/epiphany-core/src/bin/epiphany-operator-command.rs +++ b/epiphany-core/src/bin/epiphany-operator-command.rs @@ -36,7 +36,14 @@ fn main() -> Result<()> { ) })?; print_readiness(&args, &signer, "serving")?; - serve_operator_command_rudp(socket, &config, &signer) + let health = epiphany_core::OperatorCommandServiceHealthConfig { + store: args.readiness_store.clone(), + bind: args.bind.to_string(), + release_id: args.release_id.clone(), + release_witness_sha256: args.release_witness_sha256.clone(), + source_commit: args.source_commit.clone(), + }; + serve_operator_command_rudp(socket, &config, &signer, &health) } fn print_readiness( @@ -71,6 +78,10 @@ struct Args { channel_ids: Vec, actor_capabilities: BTreeMap>, max_ttl_seconds: i64, + readiness_store: PathBuf, + release_id: String, + release_witness_sha256: String, + source_commit: String, status_only: bool, } @@ -111,6 +122,10 @@ impl Args { | "--resident-self-store" | "--runtime-store" | "--guild-id" + | "--readiness-store" + | "--release-id" + | "--release-witness-sha256" + | "--source-commit" | "--max-ttl-seconds" => { let value = values .next() @@ -152,6 +167,10 @@ impl Args { channel_ids: channels, actor_capabilities: actors, max_ttl_seconds, + readiness_store: required("--readiness-store")?.into(), + release_id: required("--release-id")?, + release_witness_sha256: required("--release-witness-sha256")?, + source_commit: required("--source-commit")?, status_only, }) } diff --git a/epiphany-core/src/lib.rs b/epiphany-core/src/lib.rs index 85133302..8b470833 100644 --- a/epiphany-core/src/lib.rs +++ b/epiphany-core/src/lib.rs @@ -832,10 +832,11 @@ pub use operator_command::{ pub use operator_command_service::{ EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_SCHEMA_VERSION, EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_TYPE, EPIPHANY_OPERATOR_COMMAND_RUDP_CONNECTION_ID, - EpiphanyOperatorCommandResultReceipt, EpiphanyOperatorCommandWireResult, - OperatorCommandInteropFixtureManifest, OperatorCommandServiceConfig, - OperatorCommandServiceReadiness, execute_operator_command_admission, - operator_command_result_receipt_signing_payload, + EpiphanyOperatorCommandResultReceipt, EpiphanyOperatorCommandServiceHealth, + EpiphanyOperatorCommandWireResult, OperatorCommandInteropFixtureManifest, + OperatorCommandServiceConfig, OperatorCommandServiceHealthConfig, + OperatorCommandServiceReadiness, authenticate_operator_command_service_health, + execute_operator_command_admission, operator_command_result_receipt_signing_payload, operator_command_result_receipt_signing_purpose, read_operator_command_trust_anchor, serve_operator_command_rudp, write_operator_command_interop_fixture, }; diff --git a/epiphany-core/src/operator_command_service.rs b/epiphany-core/src/operator_command_service.rs index 16fde6e1..9c15d4a8 100644 --- a/epiphany-core/src/operator_command_service.rs +++ b/epiphany-core/src/operator_command_service.rs @@ -1,4 +1,8 @@ use anyhow::{Context, Result, anyhow, bail}; +use chrono::Utc; +use cultcache_rs::{ + CacheBackingStore, CultCache, DatabaseEntry, SingleFileMessagePackBackingStore, +}; use cultnet_rs::{ CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract, @@ -9,6 +13,7 @@ use sha2::{Digest, Sha256}; use std::net::UdpSocket; use std::path::{Path, PathBuf}; use std::time::Duration; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; use crate::{ BIFROST_OPERATOR_COMMAND_DELIVERY_TYPE, BifrostOperatorCommandAdmission, HostIdentitySigner, @@ -22,6 +27,10 @@ pub const EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_SCHEMA_VERSION: &str = "epiphany.operator_command.sealed_result.v1"; const RESULT_SIGNING_PURPOSE: &str = "epiphany.operator-command.sealed-result.v1"; pub const EPIPHANY_OPERATOR_COMMAND_RUDP_CONNECTION_ID: u32 = 0xe91f_0001; +pub const EPIPHANY_OPERATOR_COMMAND_SERVICE_HEALTH_SCHEMA_VERSION: &str = + "epiphany.operator_command.service_health.v0"; +const SERVICE_HEALTH_KEY: &str = "operator-command-service"; +const SERVICE_HEALTH_SIGNING_PURPOSE: &str = "epiphany.operator-command.service-health.v0"; #[derive(Clone, Debug)] pub struct OperatorCommandServiceConfig { @@ -33,6 +42,55 @@ pub struct OperatorCommandServiceConfig { pub trusted_bifrost_identity: HostIdentityTrustAnchorEntry, } +#[derive(Clone, Debug)] +pub struct OperatorCommandServiceHealthConfig { + pub store: PathBuf, + pub bind: String, + pub release_id: String, + pub release_witness_sha256: String, + pub source_commit: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache( + type = "epiphany.operator_command.service_health", + schema = "EpiphanyOperatorCommandServiceHealth" +)] +pub struct EpiphanyOperatorCommandServiceHealth { + #[cultcache(key = 0)] + pub schema_version: String, + #[cultcache(key = 1)] + pub runtime_id: String, + #[cultcache(key = 2)] + pub release_id: String, + #[cultcache(key = 3)] + pub release_witness_sha256: String, + #[cultcache(key = 4)] + pub source_commit: String, + #[cultcache(key = 5)] + pub bind: String, + #[cultcache(key = 6)] + pub config_sha256: String, + #[cultcache(key = 7)] + pub bifrost_identity_id: String, + #[cultcache(key = 8)] + pub executor_identity_id: String, + #[cultcache(key = 9)] + pub observed_at: String, + #[cultcache(key = 10)] + pub observed_at_millis: u64, + #[cultcache(key = 11)] + pub process_id: u32, + #[cultcache(key = 12)] + pub process_creation_token: u64, + #[cultcache(key = 13)] + pub process_executable_path: String, + #[cultcache(key = 14)] + pub private_state_exposed: bool, + #[cultcache(key = 15)] + pub executor_signature: Vec, +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct EpiphanyOperatorCommandResultReceipt { @@ -192,6 +250,7 @@ pub fn serve_operator_command_rudp( socket: UdpSocket, config: &OperatorCommandServiceConfig, signer: &HostIdentitySigner, + health: &OperatorCommandServiceHealthConfig, ) -> Result<()> { socket.set_read_timeout(Some(Duration::from_millis(100)))?; let mut transport = @@ -200,22 +259,30 @@ pub fn serve_operator_command_rudp( socket, EPIPHANY_OPERATOR_COMMAND_RUDP_CONNECTION_ID, ))?; - serve_operator_command_rudp_loop(&mut transport, config, signer, None, None) + serve_operator_command_rudp_loop(&mut transport, config, signer, Some(health), None, None) } fn serve_operator_command_rudp_loop( transport: &mut CultNetRudpSocketTransportConnection, config: &OperatorCommandServiceConfig, signer: &HostIdentitySigner, + health: Option<&OperatorCommandServiceHealthConfig>, iteration_limit: Option, fixed_now: Option<&str>, ) -> Result<()> { let mut iterations = 0usize; + let mut last_health = None::; loop { if iteration_limit.is_some_and(|limit| iterations >= limit) { return Ok(()); } iterations += 1; + if let Some(health) = health { + if last_health.is_none_or(|at| at.elapsed() >= Duration::from_secs(5)) { + publish_operator_command_service_health(&health.store, config, signer, health)?; + last_health = Some(Instant::now()); + } + } match transport.receive_once() { Ok(Some(frame)) => { let response = match fixed_now { @@ -251,6 +318,156 @@ fn serve_operator_command_rudp_loop( } } +fn service_config_sha256( + config: &OperatorCommandServiceConfig, + health: &OperatorCommandServiceHealthConfig, +) -> Result { + let bytes = rmp_serde::to_vec(&( + config.policy.runtime_id.as_str(), + config.policy.discord_guild_id.as_str(), + &config.policy.allowed_channel_ids, + &config.policy.actor_capabilities, + config.policy.max_ttl_seconds, + config.trusted_bifrost_identity.identity_id.as_str(), + health.bind.as_str(), + health.release_id.as_str(), + health.release_witness_sha256.as_str(), + health.source_commit.as_str(), + ))?; + Ok(format!("sha256-{:x}", Sha256::digest(bytes))) +} + +fn service_health_signing_payload(value: &EpiphanyOperatorCommandServiceHealth) -> Result> { + rmp_serde::to_vec(&( + value.schema_version.as_str(), + value.runtime_id.as_str(), + value.release_id.as_str(), + value.release_witness_sha256.as_str(), + value.source_commit.as_str(), + value.bind.as_str(), + value.config_sha256.as_str(), + value.bifrost_identity_id.as_str(), + value.executor_identity_id.as_str(), + value.observed_at.as_str(), + value.observed_at_millis, + value.process_id, + value.process_creation_token, + value.process_executable_path.as_str(), + value.private_state_exposed, + )) + .map_err(Into::into) +} + +pub fn publish_operator_command_service_health( + store: &Path, + config: &OperatorCommandServiceConfig, + signer: &HostIdentitySigner, + health: &OperatorCommandServiceHealthConfig, +) -> Result { + let observed_at_millis = SystemTime::now() + .duration_since(UNIX_EPOCH)? + .as_millis() + .try_into() + .context("operator health time exceeds u64")?; + let process = crate::capture_process_instance(std::process::id())?; + let mut value = EpiphanyOperatorCommandServiceHealth { + schema_version: EPIPHANY_OPERATOR_COMMAND_SERVICE_HEALTH_SCHEMA_VERSION.into(), + runtime_id: config.policy.runtime_id.clone(), + release_id: health.release_id.clone(), + release_witness_sha256: health.release_witness_sha256.clone(), + source_commit: health.source_commit.clone(), + bind: health.bind.clone(), + config_sha256: service_config_sha256(config, health)?, + bifrost_identity_id: config.trusted_bifrost_identity.identity_id.clone(), + executor_identity_id: signer.entry().identity_id.clone(), + observed_at: Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true), + observed_at_millis, + process_id: process.process_id, + process_creation_token: process.creation_token, + process_executable_path: process.executable_path.display().to_string(), + private_state_exposed: false, + executor_signature: Vec::new(), + }; + value.executor_signature = signer + .sign( + SERVICE_HEALTH_SIGNING_PURPOSE, + &service_health_signing_payload(&value)?, + )? + .signature; + let backing = SingleFileMessagePackBackingStore::new(store); + let existing = backing.pull_all()?.into_iter().find(|entry| { + entry.r#type == ::TYPE + && entry.key == SERVICE_HEALTH_KEY + }); + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + let (replacement, _) = cache.prepare_entry(SERVICE_HEALTH_KEY, &value)?; + let committed = match existing { + Some(expected) => backing.compare_and_swap_entry(&expected, replacement)?, + None => backing.insert_entry_if_absent(replacement)?, + }; + if !committed { + bail!("operator service health lost exact CAS"); + } + Ok(value) +} + +pub fn authenticate_operator_command_service_health( + store: &Path, + runtime_id: &str, + release_id: &str, + release_witness_sha256: &str, + source_commit: &str, + bind: &str, + executor: &HostIdentityTrustAnchorEntry, + now_millis: u64, + freshness_millis: u64, +) -> Result { + let mut matches = SingleFileMessagePackBackingStore::new(store) + .pull_all()? + .into_iter() + .filter(|entry| { + entry.r#type == ::TYPE + && entry.key == SERVICE_HEALTH_KEY + }); + let envelope = matches + .next() + .context("operator service health is absent")?; + if matches.next().is_some() { + bail!("operator service health has duplicate owner state"); + } + let value: EpiphanyOperatorCommandServiceHealth = rmp_serde::from_slice(&envelope.payload)?; + if value.schema_version != EPIPHANY_OPERATOR_COMMAND_SERVICE_HEALTH_SCHEMA_VERSION + || value.private_state_exposed + || value.runtime_id != runtime_id + || value.release_id != release_id + || value.release_witness_sha256 != release_witness_sha256 + || value.source_commit != source_commit + || value.bind != bind + || !value.config_sha256.starts_with("sha256-") + || value.config_sha256.len() != 71 + || value.bifrost_identity_id.trim().is_empty() + || value.executor_identity_id != executor.identity_id + || value.process_id == 0 + || value.process_creation_token == 0 + || value.process_executable_path.trim().is_empty() + || value.observed_at_millis > now_millis + || now_millis.saturating_sub(value.observed_at_millis) > freshness_millis + { + bail!("operator service health is stale or not bound to the exact release/config/identity"); + } + crate::verify_host_identity_trust_anchor_signature( + executor, + SERVICE_HEALTH_SIGNING_PURPOSE, + &service_health_signing_payload(&value)?, + &crate::HostIdentitySignature { + identity_id: value.executor_identity_id.clone(), + signature: value.executor_signature.clone(), + }, + )?; + Ok(value) +} + fn process_wire_frame( config: &OperatorCommandServiceConfig, signer: &HostIdentitySigner, @@ -565,6 +782,72 @@ mod tests { Ok(value) } + #[test] + fn signed_service_health_binds_release_config_identity_and_freshness() -> Result<()> { + let temp = tempfile::tempdir()?; + let (config, executor, _) = fixture(temp.path())?; + let executor_anchor = crate::export_host_identity_trust_anchor( + &executor, + &temp.path().join("executor-anchor.cc"), + )?; + let health_config = OperatorCommandServiceHealthConfig { + store: temp.path().join("operator-health.cc"), + bind: "127.0.0.1:17874".into(), + release_id: "release-1".into(), + release_witness_sha256: format!("sha256-{}", "a".repeat(64)), + source_commit: "b".repeat(40), + }; + let published = publish_operator_command_service_health( + &health_config.store, + &config, + &executor, + &health_config, + )?; + assert_eq!( + authenticate_operator_command_service_health( + &health_config.store, + &config.policy.runtime_id, + &health_config.release_id, + &health_config.release_witness_sha256, + &health_config.source_commit, + &health_config.bind, + &executor_anchor, + published.observed_at_millis, + 15_000, + )?, + published + ); + assert!( + authenticate_operator_command_service_health( + &health_config.store, + &config.policy.runtime_id, + "alien-release", + &health_config.release_witness_sha256, + &health_config.source_commit, + &health_config.bind, + &executor_anchor, + published.observed_at_millis, + 15_000, + ) + .is_err() + ); + assert!( + authenticate_operator_command_service_health( + &health_config.store, + &config.policy.runtime_id, + &health_config.release_id, + &health_config.release_witness_sha256, + &health_config.source_commit, + &health_config.bind, + &executor_anchor, + published.observed_at_millis + 15_001, + 15_000, + ) + .is_err() + ); + Ok(()) + } + #[test] fn seals_exact_result_and_replays_same_receipt() -> Result<()> { let temp = tempfile::tempdir()?; @@ -697,6 +980,7 @@ mod tests { &mut transport, &config, &executor, + None, Some(100), Some("2026-07-19T12:00:01Z"), )