GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bind operator service into signed runtime health

Browse source
This commit is contained in:
Meta Vens 2026-07-19 15:33:42 +01:00
commit 75b3813874
5 changed files with 445 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

98
epiphany-core/src/bin/epiphany-daemon-supervisor.rs
View file

@ -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<PathBuf>,
resident_self_store: Option<PathBuf>,
resident_provider_stale_seconds: u64,
operator_service_readiness_store: Option<PathBuf>,
operator_executor_trust_anchor: Option<PathBuf>,
operator_service_bind: Option<String>,
}
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,
})
}
}

48
epiphany-core/src/bin/epiphany-host-identity.rs
View file

@ -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 <enroll|status|enroll-trust-anchor|export-trust-anchor> [output]")
})?;
let values = std::env::args().skip(2).collect::<Vec<_>>();
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 <enroll|status|enroll-trust-anchor|export-trust-anchor> [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
}))?
);

21
epiphany-core/src/bin/epiphany-operator-command.rs
View file

@ -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<String>,
actor_capabilities: BTreeMap<String, Vec<OperatorCapability>>,
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,
})
}

9
epiphany-core/src/lib.rs
View file

@ -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,
};

286
epiphany-core/src/operator_command_service.rs
View file

@ -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<u8>,
}
#[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<usize>,
fixed_now: Option<&str>,
) -> Result<()> {
let mut iterations = 0usize;
let mut last_health = None::<Instant>;
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<String> {
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<Vec<u8>> {
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<EpiphanyOperatorCommandServiceHealth> {
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 == <EpiphanyOperatorCommandServiceHealth as DatabaseEntry>::TYPE
&& entry.key == SERVICE_HEALTH_KEY
});
let mut cache = CultCache::new();
cache.register_entry_type::<EpiphanyOperatorCommandServiceHealth>()?;
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<EpiphanyOperatorCommandServiceHealth> {
let mut matches = SingleFileMessagePackBackingStore::new(store)
.pull_all()?
.into_iter()
.filter(|entry| {
entry.r#type == <EpiphanyOperatorCommandServiceHealth as DatabaseEntry>::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"),
)