bingshuo/hololake-system-architecture
Watch
1
0
Fork
You've already forked hololake-system-architecture
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

feat(hololake): add signed dynamic capability routing

Browse source
This commit is contained in:
冰朔 2026-08-13 12:45:35 +08:00
commit ec8bbfd604
12 changed files with 1085 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

1
product-source/hololake-native-desktop/src-tauri/Cargo.lock generated
View file

@ -1345,6 +1345,7 @@ checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
name = "hololake-native-desktop"
version = "0.1.0"
dependencies = [
"base64 0.22.1",
"dirs",
"fs2",
"ring",

1
product-source/hololake-native-desktop/src-tauri/Cargo.toml
View file

@ -18,6 +18,7 @@ tauri-build = { version = "2.5.4", features = [] }
dirs = "6"
fs2 = "0.4"
ring = "0.17"
base64 = "0.22"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tauri = { version = "=2.10.2", features = [] }

5
product-source/hololake-native-desktop/src-tauri/dynamic-routing-trust.json Normal file
View file

@ -0,0 +1,5 @@
{
"schema": "hololake.dynamic-routing-trust/v1",
"state": "UNPROVISIONED_FAIL_CLOSED",
"signers": []
}

244
product-source/hololake-native-desktop/src-tauri/src/direct_local_broker.rs
View file

@ -1,6 +1,11 @@
use crate::direct_local_session::{
append_event_at, direct_session_root, open_at, resume_at, AppendSessionEventInput,
OpenSessionInput, ResumeSessionInput,
append_event_at, authenticate_at, direct_session_root, open_at, resume_at,
AppendSessionEventInput, AuthenticateSessionInput, OpenSessionInput, ResumeSessionInput,
};
use crate::dynamic_capability_routing::{
install_trusted_registry_at, record_health_at, resolve_at as resolve_capability_route_at,
routing_root as dynamic_routing_root, DynamicNodeRegistry, ResolveCapabilityRouteInput,
SignedNodeHealth,
};
use ring::digest::{digest, SHA256};
use serde::{Deserialize, Serialize};
@ -12,7 +17,7 @@ use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
mpsc, Arc,
};
use std::thread;
use std::time::Duration;
@ -25,12 +30,16 @@ const MAX_REQUEST_BYTES: u64 = 1024 * 1024;
pub struct DirectLocalBrokerHandle {
shutdown: Arc<AtomicBool>,
socket_path: PathBuf,
worker: Option<thread::JoinHandle<()>>,
}
impl Drop for DirectLocalBrokerHandle {
fn drop(&mut self) {
self.shutdown.store(true, Ordering::Release);
let _ = UnixStream::connect(&self.socket_path);
if let Some(worker) = self.worker.take() {
let _ = worker.join();
}
}
}
@ -45,6 +54,30 @@ enum BrokerRequest {
OpenSession(OpenSessionInput),
ResumeSession(ResumeSessionInput),
AppendEvent(AppendSessionEventInput),
ResolveCapabilityRoute(AuthenticatedRouteInput),
InstallDynamicNodeRegistry(AuthenticatedRegistryInput),
RecordSignedNodeHealth(AuthenticatedHealthInput),
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct AuthenticatedRouteInput {
session: AuthenticateSessionInput,
route: ResolveCapabilityRouteInput,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct AuthenticatedRegistryInput {
session: AuthenticateSessionInput,
registry: DynamicNodeRegistry,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct AuthenticatedHealthInput {
session: AuthenticateSessionInput,
health: SignedNodeHealth,
}
#[derive(Debug, Serialize)]
@ -75,14 +108,16 @@ struct ConnectorDescriptor {
state: String,
transport: String,
socket_path: String,
process_id: u32,
}
pub fn start(app: &AppHandle) -> Result<DirectLocalBrokerHandle, Box<dyn std::error::Error>> {
let session_root = direct_session_root(app).map_err(std::io::Error::other)?;
let routing_root = dynamic_routing_root(app).map_err(std::io::Error::other)?;
let app_data = app.path().app_data_dir()?;
let descriptor_path = app_data.join("direct-local-broker-v1.json");
let socket_path = short_socket_path(&app_data);
start_at(session_root, descriptor_path, socket_path).map_err(|error| error.into())
start_at(session_root, routing_root, descriptor_path, socket_path).map_err(|error| error.into())
}
pub fn run_connector() -> Result<(), String> {
@ -98,8 +133,7 @@ pub fn run_connector() -> Result<(), String> {
{
return Err("HOLOLAKE_BROKER_DESCRIPTOR_UNSUPPORTED".into());
}
let mut stream = UnixStream::connect(&descriptor.socket_path)
.map_err(|error| format!("HOLOLAKE_DIRECT_LOCAL_BROKER_NOT_RUNNING: {error}"))?;
let mut stream = connect_with_startup_grace(&descriptor)?;
let read_stream = stream
.try_clone()
.map_err(|error| format!("HOLOLAKE_CONNECTOR_CLONE_FAILED: {error}"))?;
@ -125,6 +159,34 @@ pub fn run_connector() -> Result<(), String> {
Ok(())
}
fn connect_with_startup_grace(descriptor: &ConnectorDescriptor) -> Result<UnixStream, String> {
if descriptor.process_id == 0 {
return Err("HOLOLAKE_BROKER_DESCRIPTOR_UNSUPPORTED".into());
}
let mut last_error = None;
for _ in 0..50 {
match UnixStream::connect(&descriptor.socket_path) {
Ok(stream) => return Ok(stream),
Err(error)
if matches!(
error.kind(),
std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
) =>
{
last_error = Some(error);
thread::sleep(Duration::from_millis(100));
}
Err(error) => return Err(format!("HOLOLAKE_DIRECT_LOCAL_BROKER_NOT_RUNNING: {error}")),
}
}
Err(format!(
"HOLOLAKE_DIRECT_LOCAL_BROKER_NOT_RUNNING: {}",
last_error
.map(|error| error.to_string())
.unwrap_or_else(|| "startup grace exhausted".into())
))
}
fn connector_descriptor_path() -> Result<PathBuf, String> {
if let Some(path) = std::env::var_os("HOLOLAKE_BROKER_DESCRIPTOR") {
return Ok(PathBuf::from(path));
@ -139,6 +201,7 @@ fn connector_descriptor_path() -> Result<PathBuf, String> {
fn start_at(
session_root: PathBuf,
routing_root: PathBuf,
descriptor_path: PathBuf,
socket_path: PathBuf,
) -> Result<DirectLocalBrokerHandle, String> {
@ -179,28 +242,43 @@ fn start_at(
let shutdown = Arc::new(AtomicBool::new(false));
let worker_shutdown = Arc::clone(&shutdown);
let worker_socket = socket_path.clone();
thread::Builder::new()
let (ready_sender, ready_receiver) = mpsc::sync_channel(1);
let worker = thread::Builder::new()
.name("hololake-direct-local-broker".into())
.spawn(move || {
serve(listener, session_root, &worker_shutdown);
let _ = ready_sender.send(());
serve(listener, session_root, routing_root, &worker_shutdown);
let _ = fs::remove_file(worker_socket);
})
.map_err(|error| format!("HOLOLAKE_BROKER_THREAD_FAILED: {error}"))?;
ready_receiver
.recv_timeout(Duration::from_secs(2))
.map_err(|error| format!("HOLOLAKE_BROKER_STARTUP_FAILED: {error}"))?;
Ok(DirectLocalBrokerHandle {
shutdown,
socket_path,
worker: Some(worker),
})
}
fn serve(listener: UnixListener, session_root: PathBuf, shutdown: &AtomicBool) {
fn serve(
listener: UnixListener,
session_root: PathBuf,
routing_root: PathBuf,
shutdown: &AtomicBool,
) {
while !shutdown.load(Ordering::Acquire) {
match listener.accept() {
Ok((stream, _)) => {
if shutdown.load(Ordering::Acquire) {
break;
}
let root = session_root.clone();
let routes = routing_root.clone();
let _ = thread::Builder::new()
.name("hololake-direct-local-client".into())
.spawn(move || serve_connection(stream, &root));
.spawn(move || serve_connection(stream, &root, &routes));
}
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
thread::sleep(Duration::from_millis(25));
@ -210,7 +288,7 @@ fn serve(listener: UnixListener, session_root: PathBuf, shutdown: &AtomicBool) {
}
}
fn serve_connection(mut stream: UnixStream, session_root: &Path) {
fn serve_connection(mut stream: UnixStream, session_root: &Path, routing_root: &Path) {
let read_stream = match stream.try_clone() {
Ok(stream) => stream,
Err(_) => return,
@ -232,7 +310,7 @@ fn serve_connection(mut stream: UnixStream, session_root: &Path) {
let response = if read as u64 > MAX_REQUEST_BYTES || !bytes.ends_with(b"\n") {
BrokerResponse::error("HOLOLAKE_BROKER_REQUEST_TOO_LARGE")
} else {
dispatch(session_root, &bytes[..bytes.len() - 1])
dispatch(session_root, routing_root, &bytes[..bytes.len() - 1])
};
if serde_json::to_writer(&mut stream, &response).is_err()
|| stream.write_all(b"\n").is_err()
@ -246,7 +324,7 @@ fn serve_connection(mut stream: UnixStream, session_root: &Path) {
}
}
fn dispatch(session_root: &Path, bytes: &[u8]) -> BrokerResponse {
fn dispatch(session_root: &Path, routing_root: &Path, bytes: &[u8]) -> BrokerResponse {
let request: BrokerRequest = match serde_json::from_slice(bytes) {
Ok(request) => request,
Err(error) => {
@ -266,6 +344,35 @@ fn dispatch(session_root: &Path, bytes: &[u8]) -> BrokerResponse {
.and_then(|receipt| serde_json::to_value(receipt).map_err(|error| error.to_string())),
BrokerRequest::AppendEvent(input) => append_event_at(session_root, input)
.and_then(|receipt| serde_json::to_value(receipt).map_err(|error| error.to_string())),
BrokerRequest::ResolveCapabilityRoute(input) => {
if let Err(error) = authenticate_at(session_root, &input.session) {
return BrokerResponse::error(&error);
}
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_millis() as u64)
.map_err(|error| format!("HOLOLAKE_SYSTEM_CLOCK_INVALID: {error}"));
now.and_then(|now| resolve_capability_route_at(routing_root, input.route, now))
.and_then(|receipt| {
serde_json::to_value(receipt).map_err(|error| error.to_string())
})
}
BrokerRequest::InstallDynamicNodeRegistry(input) => {
if let Err(error) = authenticate_at(session_root, &input.session) {
return BrokerResponse::error(&error);
}
install_trusted_registry_at(routing_root, input.registry).and_then(|receipt| {
serde_json::to_value(receipt).map_err(|error| error.to_string())
})
}
BrokerRequest::RecordSignedNodeHealth(input) => {
if let Err(error) = authenticate_at(session_root, &input.session) {
return BrokerResponse::error(&error);
}
record_health_at(routing_root, input.health).and_then(|receipt| {
serde_json::to_value(receipt).map_err(|error| error.to_string())
})
}
};
match result {
Ok(value) => BrokerResponse::success(value),
@ -351,7 +458,9 @@ mod tests {
let descriptor = temp.path().join("broker.json");
let sessions = temp.path().join("sessions");
fs::create_dir_all(&sessions).unwrap();
let _broker = start_at(sessions.clone(), descriptor, socket.clone()).unwrap();
let routes = temp.path().join("routes");
fs::create_dir_all(&routes).unwrap();
let _broker = start_at(sessions.clone(), routes, descriptor, socket.clone()).unwrap();
let ticket = issue_ticket_at(
&sessions,
IssueDiscoveryTicketInput {
@ -407,7 +516,15 @@ mod tests {
},
)
.unwrap();
let first = start_at(sessions.clone(), descriptor.clone(), socket.clone()).unwrap();
let routes = temp.path().join("routes");
fs::create_dir_all(&routes).unwrap();
let first = start_at(
sessions.clone(),
routes.clone(),
descriptor.clone(),
socket.clone(),
)
.unwrap();
let opened = request(
&socket,
serde_json::json!({
@ -427,7 +544,7 @@ mod tests {
}
thread::sleep(Duration::from_millis(10));
}
let _second = start_at(sessions, descriptor, socket.clone()).unwrap();
let _second = start_at(sessions, routes, descriptor, socket.clone()).unwrap();
let resumed = request(
&socket,
serde_json::json!({
@ -442,4 +559,99 @@ mod tests {
);
assert_eq!(resumed["result"]["state"], "RESUMED");
}
#[test]
fn broker_route_resolution_requires_the_active_session_secret() {
let temp = TempDir::new().unwrap();
let socket = temp.path().join("runtime/broker.sock");
let descriptor = temp.path().join("broker.json");
let sessions = temp.path().join("sessions");
let routes = temp.path().join("routes");
fs::create_dir_all(&sessions).unwrap();
fs::create_dir_all(&routes).unwrap();
let _broker = start_at(sessions.clone(), routes, descriptor, socket.clone()).unwrap();
let response = request(
&socket,
serde_json::json!({
"operation": "RESOLVE_CAPABILITY_ROUTE",
"input": {
"session": {
"accountId": "human-1",
"sessionId": "invented-session",
"resumeSecret": "invented-secret-long-enough"
},
"route": {
"humanId": "human-1",
"personaId": "ICE-P-ZY001",
"domainId": "DOM-FIFTH-0001",
"capabilityId": "GH-PNCC-READ"
}
}
}),
);
assert_eq!(response["ok"], false);
assert_eq!(
response["error"],
"HOLOLAKE_DIRECT_ACTIVE_SESSION_NOT_FOUND"
);
}
#[test]
fn authenticated_ai_still_cannot_install_registry_before_trust_is_provisioned() {
let temp = TempDir::new().unwrap();
let socket = temp.path().join("runtime/broker.sock");
let descriptor = temp.path().join("broker.json");
let sessions = temp.path().join("sessions");
let routes = temp.path().join("routes");
fs::create_dir_all(&sessions).unwrap();
fs::create_dir_all(&routes).unwrap();
let _broker = start_at(sessions.clone(), routes, descriptor, socket.clone()).unwrap();
let ticket = issue_ticket_at(
&sessions,
IssueDiscoveryTicketInput {
account_id: "human-1".into(),
lane_id: "DEV-1".into(),
client_instance_id: "codex-1".into(),
},
)
.unwrap();
let opened = request(
&socket,
serde_json::json!({
"operation": "OPEN_SESSION",
"input": {
"accountId": "human-1",
"laneId": "DEV-1",
"clientInstanceId": "codex-1",
"discoveryTicket": ticket.discovery_ticket
}
}),
);
let response = request(
&socket,
serde_json::json!({
"operation": "INSTALL_DYNAMIC_NODE_REGISTRY",
"input": {
"session": {
"accountId": "human-1",
"sessionId": opened["result"]["sessionId"],
"resumeSecret": opened["result"]["resumeSecret"]
},
"registry": {
"schema": "hololake.dynamic-node-registry/v1",
"registryId": "candidate-registry",
"version": 1,
"nodes": [],
"signerId": "invented-signer",
"signature": "invented-signature"
}
}
}),
);
assert_eq!(response["ok"], false);
assert_eq!(
response["error"],
"HOLOLAKE_DYNAMIC_ROUTING_TRUST_UNPROVISIONED"
);
}
}

25
product-source/hololake-native-desktop/src-tauri/src/direct_local_session.rs
View file

@ -62,6 +62,14 @@ pub struct AppendSessionEventInput {
pub expected_previous_sequence: u64,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AuthenticateSessionInput {
pub account_id: String,
pub session_id: String,
pub resume_secret: String,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DirectSessionReceipt {
@ -397,6 +405,23 @@ pub(crate) fn append_event_at(
Ok(event_receipt("APPENDED", &event))
}
pub(crate) fn authenticate_at(root: &Path, input: &AuthenticateSessionInput) -> Result<(), String> {
validate_identifier(&input.account_id, "ACCOUNT")?;
validate_identifier(&input.session_id, "SESSION")?;
validate_secret(&input.resume_secret, "RESUME_SECRET")?;
let account_key = sha256_hex(input.account_id.as_bytes());
require_active_session(root, &account_key, &input.session_id)?;
let path = session_path(root, &account_key, &input.session_id);
let _lock = lock_session(&path)?;
let record = read_session(&path)?;
authorize(
&record,
&account_key,
&input.session_id,
&input.resume_secret,
)
}
fn session_path(root: &Path, account_key: &str, session_id: &str) -> PathBuf {
root.join("accounts")
.join(account_key)

690
product-source/hololake-native-desktop/src-tauri/src/dynamic_capability_routing.rs Normal file
View file

@ -0,0 +1,690 @@
use base64::{engine::general_purpose::STANDARD as BASE64, Engine as _};
use fs2::FileExt;
use ring::signature::{UnparsedPublicKey, ED25519};
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::collections::BTreeSet;
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use tauri::{AppHandle, Manager};
use uuid::Uuid;
const REGISTRY_SCHEMA: &str = "hololake.dynamic-node-registry/v1";
const HEALTH_SCHEMA: &str = "hololake.signed-node-health/v1";
const ROUTE_SCHEMA: &str = "hololake.capability-route-receipt/v1";
const MAX_IDENTIFIER_BYTES: usize = 128;
const MAX_CLOCK_SKEW_MS: u64 = 300_000;
const MAX_NODES: usize = 4096;
const MAX_BINDINGS_PER_NODE: usize = 256;
const EMBEDDED_ROUTING_TRUST: &str = include_str!("../dynamic-routing-trust.json");
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct DynamicNodeRegistry {
schema: String,
registry_id: String,
version: u64,
nodes: Vec<DynamicNodeRecord>,
signer_id: String,
signature: String,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct DynamicRoutingTrust {
schema: String,
state: RoutingTrustState,
signers: Vec<RoutingTrustSigner>,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
enum RoutingTrustState {
UnprovisionedFailClosed,
Provisioned,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct RoutingTrustSigner {
signer_id: String,
public_key: String,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct DynamicNodeRecord {
node_id: String,
domain_id: String,
endpoint: String,
protocol_public_key: String,
human_ids: Vec<String>,
persona_ids: Vec<String>,
capabilities: Vec<String>,
priority: u32,
enabled: bool,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct SignedNodeHealth {
schema: String,
registry_version: u64,
node_id: String,
observed_at_unix_ms: u64,
expires_at_unix_ms: u64,
sequence: u64,
status: HealthStatus,
capabilities: Vec<String>,
signature: String,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
enum HealthStatus {
Ready,
Degraded,
Offline,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ResolveCapabilityRouteInput {
human_id: String,
persona_id: String,
domain_id: String,
capability_id: String,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CapabilityRouteReceipt {
schema: &'static str,
state: &'static str,
registry_id: String,
registry_version: u64,
node_id: String,
domain_id: String,
endpoint: String,
capability_id: String,
human_id: String,
persona_id: String,
health_sequence: u64,
health_expires_at_unix_ms: u64,
write_authority_granted: bool,
receipt_id: String,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct RegistryInstallReceipt {
schema: &'static str,
state: &'static str,
registry_id: String,
version: u64,
node_count: usize,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct HealthRecordReceipt {
schema: &'static str,
state: &'static str,
node_id: String,
sequence: u64,
expires_at_unix_ms: u64,
}
pub(crate) fn install_trusted_registry_at(
root: &Path,
registry: DynamicNodeRegistry,
) -> Result<RegistryInstallReceipt, String> {
let public_key = trusted_registry_key(EMBEDDED_ROUTING_TRUST, &registry.signer_id)?;
install_registry_at(root, registry, &public_key)
}
pub(crate) fn routing_root(app: &AppHandle) -> Result<PathBuf, String> {
let root = app
.path()
.app_data_dir()
.map_err(|error| format!("HOLOLAKE_APP_DATA_UNAVAILABLE: {error}"))?
.join("dynamic-capability-routing-v1");
fs::create_dir_all(&root)
.map_err(|error| format!("HOLOLAKE_DYNAMIC_ROUTING_STORAGE_UNAVAILABLE: {error}"))?;
Ok(root)
}
fn install_registry_at(
root: &Path,
registry: DynamicNodeRegistry,
trusted_public_key: &str,
) -> Result<RegistryInstallReceipt, String> {
let _lock = lock_file(&root.join("registry.lock"), "REGISTRY")?;
validate_registry(&registry)?;
verify_signature(
trusted_public_key,
&registry.signature,
canonical_registry_bytes(&registry).as_bytes(),
"REGISTRY",
)?;
let path = root.join("registry.json");
if path.exists() {
let current: DynamicNodeRegistry = read_json(&path, "REGISTRY")?;
if registry.registry_id != current.registry_id || registry.version <= current.version {
return Err("HOLOLAKE_DYNAMIC_REGISTRY_VERSION_NOT_INCREASING".into());
}
}
write_json_atomic(&path, &registry, "REGISTRY")?;
Ok(RegistryInstallReceipt {
schema: REGISTRY_SCHEMA,
state: "INSTALLED",
registry_id: registry.registry_id,
version: registry.version,
node_count: registry.nodes.len(),
})
}
pub(crate) fn record_health_at(
root: &Path,
health: SignedNodeHealth,
) -> Result<HealthRecordReceipt, String> {
validate_identifier(&health.node_id, "NODE")?;
let _lock = lock_file(
&root.join("health").join(format!("{}.lock", health.node_id)),
"NODE_HEALTH",
)?;
validate_capabilities(&health.capabilities)?;
if health.schema != HEALTH_SCHEMA || health.expires_at_unix_ms <= health.observed_at_unix_ms {
return Err("HOLOLAKE_NODE_HEALTH_INVALID".into());
}
let registry: DynamicNodeRegistry = read_json(&root.join("registry.json"), "REGISTRY")?;
if health.registry_version != registry.version {
return Err("HOLOLAKE_NODE_HEALTH_REGISTRY_VERSION_MISMATCH".into());
}
let node = registry
.nodes
.iter()
.find(|node| node.node_id == health.node_id)
.ok_or("HOLOLAKE_NODE_NOT_REGISTERED")?;
verify_signature(
&node.protocol_public_key,
&health.signature,
canonical_health_bytes(&health).as_bytes(),
"NODE_HEALTH",
)?;
let path = root.join("health").join(format!("{}.json", health.node_id));
if path.exists() {
let current: SignedNodeHealth = read_json(&path, "NODE_HEALTH")?;
if health.sequence <= current.sequence {
return Err("HOLOLAKE_NODE_HEALTH_REPLAYED".into());
}
}
write_json_atomic(&path, &health, "NODE_HEALTH")?;
Ok(HealthRecordReceipt {
schema: HEALTH_SCHEMA,
state: "RECORDED",
node_id: health.node_id,
sequence: health.sequence,
expires_at_unix_ms: health.expires_at_unix_ms,
})
}
pub(crate) fn resolve_at(
root: &Path,
input: ResolveCapabilityRouteInput,
now: u64,
) -> Result<CapabilityRouteReceipt, String> {
validate_identifier(&input.human_id, "HUMAN")?;
validate_identifier(&input.persona_id, "PERSONA")?;
validate_identifier(&input.domain_id, "DOMAIN")?;
validate_identifier(&input.capability_id, "CAPABILITY")?;
let registry: DynamicNodeRegistry = read_json(&root.join("registry.json"), "REGISTRY")?;
let mut eligible = Vec::new();
for node in registry.nodes.iter().filter(|node| {
node.enabled
&& node.domain_id == input.domain_id
&& node.human_ids.contains(&input.human_id)
&& node.persona_ids.contains(&input.persona_id)
&& node.capabilities.contains(&input.capability_id)
}) {
let path = root.join("health").join(format!("{}.json", node.node_id));
let Ok(health) = read_json::<SignedNodeHealth>(&path, "NODE_HEALTH") else {
continue;
};
if health.registry_version != registry.version
|| health.status != HealthStatus::Ready
|| health.observed_at_unix_ms > now.saturating_add(MAX_CLOCK_SKEW_MS)
|| health.expires_at_unix_ms < now
|| !health.capabilities.contains(&input.capability_id)
|| verify_signature(
&node.protocol_public_key,
&health.signature,
canonical_health_bytes(&health).as_bytes(),
"NODE_HEALTH",
)
.is_err()
{
continue;
}
eligible.push((node, health));
}
eligible.sort_by(
|(left, _), (right, _)| match left.priority.cmp(&right.priority) {
Ordering::Equal => left.node_id.cmp(&right.node_id),
ordering => ordering,
},
);
let (node, health) = eligible
.into_iter()
.next()
.ok_or("HOLOLAKE_NO_HEALTHY_CAPABILITY_ROUTE")?;
let receipt_id = sha256_hex(
format!(
"{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}",
registry.registry_id,
registry.version,
node.node_id,
input.human_id,
input.persona_id,
input.domain_id,
input.capability_id,
health.sequence
)
.as_bytes(),
);
Ok(CapabilityRouteReceipt {
schema: ROUTE_SCHEMA,
state: "ROUTE_SELECTED",
registry_id: registry.registry_id,
registry_version: registry.version,
node_id: node.node_id.clone(),
domain_id: node.domain_id.clone(),
endpoint: node.endpoint.clone(),
capability_id: input.capability_id,
human_id: input.human_id,
persona_id: input.persona_id,
health_sequence: health.sequence,
health_expires_at_unix_ms: health.expires_at_unix_ms,
write_authority_granted: false,
receipt_id,
})
}
fn validate_registry(registry: &DynamicNodeRegistry) -> Result<(), String> {
if registry.schema != REGISTRY_SCHEMA
|| registry.version == 0
|| registry.nodes.is_empty()
|| registry.nodes.len() > MAX_NODES
{
return Err("HOLOLAKE_DYNAMIC_REGISTRY_INVALID".into());
}
validate_identifier(&registry.registry_id, "REGISTRY")?;
validate_identifier(&registry.signer_id, "SIGNER")?;
let mut node_ids = BTreeSet::new();
for node in &registry.nodes {
validate_identifier(&node.node_id, "NODE")?;
validate_identifier(&node.domain_id, "DOMAIN")?;
validate_identifiers(&node.human_ids, "HUMAN")?;
validate_identifiers(&node.persona_ids, "PERSONA")?;
validate_capabilities(&node.capabilities)?;
if !node_ids.insert(node.node_id.as_str()) {
return Err("HOLOLAKE_DYNAMIC_REGISTRY_DUPLICATE_NODE".into());
}
let endpoint = tauri::Url::parse(&node.endpoint)
.map_err(|error| format!("HOLOLAKE_NODE_ENDPOINT_INVALID: {error}"))?;
if endpoint.scheme() != "https"
|| endpoint.host_str().is_none()
|| !endpoint.username().is_empty()
|| endpoint.password().is_some()
|| endpoint.query().is_some()
|| endpoint.fragment().is_some()
{
return Err("HOLOLAKE_NODE_ENDPOINT_MUST_USE_HTTPS".into());
}
decode_exact(&node.protocol_public_key, 32, "NODE_PROTOCOL_PUBLIC_KEY")?;
}
Ok(())
}
fn validate_capabilities(values: &[String]) -> Result<(), String> {
validate_identifiers(values, "CAPABILITY")
}
fn validate_identifiers(values: &[String], kind: &str) -> Result<(), String> {
if values.is_empty() || values.len() > MAX_BINDINGS_PER_NODE {
return Err(format!("HOLOLAKE_{kind}_BINDING_REQUIRED"));
}
let mut unique = BTreeSet::new();
for value in values {
validate_identifier(value, kind)?;
if !unique.insert(value.as_str()) {
return Err(format!("HOLOLAKE_{kind}_BINDING_DUPLICATE"));
}
}
Ok(())
}
fn validate_identifier(value: &str, kind: &str) -> Result<(), String> {
if value.is_empty()
|| value.len() > MAX_IDENTIFIER_BYTES
|| !value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
{
return Err(format!("HOLOLAKE_{kind}_ID_INVALID"));
}
Ok(())
}
fn canonical_registry_bytes(registry: &DynamicNodeRegistry) -> String {
let mut lines = vec![
registry.schema.clone(),
registry.registry_id.clone(),
registry.version.to_string(),
registry.signer_id.clone(),
];
for node in &registry.nodes {
lines.extend([
node.node_id.clone(),
node.domain_id.clone(),
node.endpoint.clone(),
node.protocol_public_key.clone(),
node.human_ids.join(","),
node.persona_ids.join(","),
node.capabilities.join(","),
node.priority.to_string(),
node.enabled.to_string(),
]);
}
lines.join("\n")
}
fn canonical_health_bytes(health: &SignedNodeHealth) -> String {
[
health.schema.clone(),
health.registry_version.to_string(),
health.node_id.clone(),
health.observed_at_unix_ms.to_string(),
health.expires_at_unix_ms.to_string(),
health.sequence.to_string(),
match health.status {
HealthStatus::Ready => "READY",
HealthStatus::Degraded => "DEGRADED",
HealthStatus::Offline => "OFFLINE",
}
.into(),
health.capabilities.join(","),
]
.join("\n")
}
fn verify_signature(
public_key: &str,
signature: &str,
bytes: &[u8],
kind: &str,
) -> Result<(), String> {
let public_key = decode_exact(public_key, 32, &format!("{kind}_PUBLIC_KEY"))?;
let signature = decode_exact(signature, 64, &format!("{kind}_SIGNATURE"))?;
UnparsedPublicKey::new(&ED25519, public_key)
.verify(bytes, &signature)
.map_err(|_| format!("HOLOLAKE_{kind}_SIGNATURE_INVALID"))
}
fn trusted_registry_key(raw: &str, signer_id: &str) -> Result<String, String> {
let trust: DynamicRoutingTrust = serde_json::from_str(raw)
.map_err(|error| format!("HOLOLAKE_DYNAMIC_ROUTING_TRUST_INVALID: {error}"))?;
if trust.schema != "hololake.dynamic-routing-trust/v1" {
return Err("HOLOLAKE_DYNAMIC_ROUTING_TRUST_SCHEMA_UNSUPPORTED".into());
}
match trust.state {
RoutingTrustState::UnprovisionedFailClosed => {
if !trust.signers.is_empty() {
return Err("HOLOLAKE_DYNAMIC_ROUTING_TRUST_INVALID".into());
}
Err("HOLOLAKE_DYNAMIC_ROUTING_TRUST_UNPROVISIONED".into())
}
RoutingTrustState::Provisioned => {
if trust.signers.is_empty() {
return Err("HOLOLAKE_DYNAMIC_ROUTING_TRUST_INVALID".into());
}
let mut ids = BTreeSet::new();
for signer in trust.signers {
validate_identifier(&signer.signer_id, "SIGNER")?;
decode_exact(&signer.public_key, 32, "REGISTRY_PUBLIC_KEY")?;
if !ids.insert(signer.signer_id.clone()) {
return Err("HOLOLAKE_DYNAMIC_ROUTING_TRUST_DUPLICATE_SIGNER".into());
}
if signer.signer_id == signer_id {
return Ok(signer.public_key);
}
}
Err("HOLOLAKE_DYNAMIC_ROUTING_SIGNER_NOT_TRUSTED".into())
}
}
}
fn decode_exact(value: &str, expected: usize, kind: &str) -> Result<Vec<u8>, String> {
let decoded = BASE64
.decode(value)
.map_err(|_| format!("HOLOLAKE_{kind}_INVALID"))?;
if decoded.len() != expected {
return Err(format!("HOLOLAKE_{kind}_INVALID"));
}
Ok(decoded)
}
fn write_json_atomic<T: Serialize>(path: &Path, value: &T, kind: &str) -> Result<(), String> {
let parent = path
.parent()
.ok_or("HOLOLAKE_DYNAMIC_ROUTING_PATH_INVALID")?;
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_DYNAMIC_ROUTING_STORAGE_UNAVAILABLE: {error}"))?;
let temporary = parent.join(format!(".{}.tmp", Uuid::new_v4()));
let bytes = serde_json::to_vec_pretty(value)
.map_err(|error| format!("HOLOLAKE_{kind}_INVALID: {error}"))?;
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&temporary)
.map_err(|error| format!("HOLOLAKE_{kind}_WRITE_FAILED: {error}"))?;
file.write_all(&bytes)
.and_then(|_| file.sync_all())
.map_err(|error| format!("HOLOLAKE_{kind}_WRITE_FAILED: {error}"))?;
fs::rename(&temporary, path).map_err(|error| format!("HOLOLAKE_{kind}_WRITE_FAILED: {error}"))
}
fn read_json<T: for<'de> Deserialize<'de>>(path: &Path, kind: &str) -> Result<T, String> {
let bytes = fs::read(path).map_err(|error| {
if error.kind() == std::io::ErrorKind::NotFound {
format!("HOLOLAKE_{kind}_NOT_FOUND")
} else {
format!("HOLOLAKE_{kind}_UNREADABLE: {error}")
}
})?;
serde_json::from_slice(&bytes).map_err(|error| format!("HOLOLAKE_{kind}_INVALID: {error}"))
}
fn lock_file(path: &Path, kind: &str) -> Result<std::fs::File, String> {
let parent = path
.parent()
.ok_or("HOLOLAKE_DYNAMIC_ROUTING_PATH_INVALID")?;
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_DYNAMIC_ROUTING_STORAGE_UNAVAILABLE: {error}"))?;
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)
.map_err(|error| format!("HOLOLAKE_{kind}_LOCK_FAILED: {error}"))?;
file.lock_exclusive()
.map_err(|error| format!("HOLOLAKE_{kind}_LOCK_FAILED: {error}"))?;
Ok(file)
}
fn sha256_hex(value: &[u8]) -> String {
ring::digest::digest(&ring::digest::SHA256, value)
.as_ref()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use ring::rand::SystemRandom;
use ring::signature::{Ed25519KeyPair, KeyPair};
use tempfile::TempDir;
fn key_pair() -> Ed25519KeyPair {
let pkcs8 = Ed25519KeyPair::generate_pkcs8(&SystemRandom::new()).unwrap();
Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap()
}
fn node(node_id: &str, key: &Ed25519KeyPair, priority: u32) -> DynamicNodeRecord {
DynamicNodeRecord {
node_id: node_id.into(),
domain_id: "DOM-FIFTH-0001".into(),
endpoint: format!("https://{node_id}.example.test/hololake"),
protocol_public_key: BASE64.encode(key.public_key().as_ref()),
human_ids: vec!["human-BS-0001".into()],
persona_ids: vec!["ICE-P-ZY001".into()],
capabilities: vec!["GH-PNCC-READ".into()],
priority,
enabled: true,
}
}
fn registry(nodes: Vec<DynamicNodeRecord>, signer: &Ed25519KeyPair) -> DynamicNodeRegistry {
let mut registry = DynamicNodeRegistry {
schema: REGISTRY_SCHEMA.into(),
registry_id: "registry-1".into(),
version: 1,
nodes,
signer_id: "jd-controller-release-registry".into(),
signature: String::new(),
};
registry.signature = BASE64.encode(
signer
.sign(canonical_registry_bytes(&registry).as_bytes())
.as_ref(),
);
registry
}
fn health(node_id: &str, signer: &Ed25519KeyPair, now: u64) -> SignedNodeHealth {
let mut health = SignedNodeHealth {
schema: HEALTH_SCHEMA.into(),
registry_version: 1,
node_id: node_id.into(),
observed_at_unix_ms: now,
expires_at_unix_ms: now + 60_000,
sequence: 1,
status: HealthStatus::Ready,
capabilities: vec!["GH-PNCC-READ".into()],
signature: String::new(),
};
health.signature = BASE64.encode(
signer
.sign(canonical_health_bytes(&health).as_bytes())
.as_ref(),
);
health
}
fn input() -> ResolveCapabilityRouteInput {
ResolveCapabilityRouteInput {
human_id: "human-BS-0001".into(),
persona_id: "ICE-P-ZY001".into(),
domain_id: "DOM-FIFTH-0001".into(),
capability_id: "GH-PNCC-READ".into(),
}
}
#[test]
fn installs_a_signed_dynamic_registry_and_selects_by_priority() {
let temp = TempDir::new().unwrap();
let registry_signer = key_pair();
let node_a = key_pair();
let node_b = key_pair();
let registry = registry(
vec![node("node-b", &node_b, 20), node("node-a", &node_a, 10)],
&registry_signer,
);
install_registry_at(
temp.path(),
registry,
&BASE64.encode(registry_signer.public_key().as_ref()),
)
.unwrap();
let now = 1_000_000;
record_health_at(temp.path(), health("node-a", &node_a, now)).unwrap();
record_health_at(temp.path(), health("node-b", &node_b, now)).unwrap();
let route = resolve_at(temp.path(), input(), now + 1).unwrap();
assert_eq!(route.node_id, "node-a");
assert!(!route.write_authority_granted);
}
#[test]
fn rejects_unsigned_registry_and_replayed_health() {
let temp = TempDir::new().unwrap();
let registry_signer = key_pair();
let node_key = key_pair();
let mut invalid = registry(vec![node("node-a", &node_key, 10)], &registry_signer);
invalid.signature = BASE64.encode([0_u8; 64]);
assert!(install_registry_at(
temp.path(),
invalid,
&BASE64.encode(registry_signer.public_key().as_ref())
)
.is_err());
install_registry_at(
temp.path(),
registry(vec![node("node-a", &node_key, 10)], &registry_signer),
&BASE64.encode(registry_signer.public_key().as_ref()),
)
.unwrap();
let current = health("node-a", &node_key, 1_000_000);
record_health_at(temp.path(), current.clone()).unwrap();
assert_eq!(
record_health_at(temp.path(), current).unwrap_err(),
"HOLOLAKE_NODE_HEALTH_REPLAYED"
);
}
#[test]
fn stale_unsigned_or_wrong_binding_nodes_are_not_routable() {
let temp = TempDir::new().unwrap();
let registry_signer = key_pair();
let node_key = key_pair();
install_registry_at(
temp.path(),
registry(vec![node("node-a", &node_key, 10)], &registry_signer),
&BASE64.encode(registry_signer.public_key().as_ref()),
)
.unwrap();
record_health_at(temp.path(), health("node-a", &node_key, 1_000_000)).unwrap();
assert_eq!(
resolve_at(temp.path(), input(), 2_000_000).unwrap_err(),
"HOLOLAKE_NO_HEALTHY_CAPABILITY_ROUTE"
);
let mut wrong_persona = input();
wrong_persona.persona_id = "ICE-P-OTHER".into();
assert_eq!(
resolve_at(temp.path(), wrong_persona, 1_000_001).unwrap_err(),
"HOLOLAKE_NO_HEALTHY_CAPABILITY_ROUTE"
);
}
#[test]
fn embedded_unprovisioned_trust_fails_closed() {
assert_eq!(
trusted_registry_key(EMBEDDED_ROUTING_TRUST, "any-signer").unwrap_err(),
"HOLOLAKE_DYNAMIC_ROUTING_TRUST_UNPROVISIONED"
);
}
}

1
product-source/hololake-native-desktop/src-tauri/src/lib.rs
View file

@ -1,5 +1,6 @@
mod direct_local_broker;
mod direct_local_session;
mod dynamic_capability_routing;
mod local_development_bridge;
mod release_trust;