diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index 9cbb4444ab..375365a143 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -155,12 +155,12 @@ fn identity_from_env() -> Option { } /// Build the no-redirect HTTP client used for authenticated relay media -/// fetches (download / copy). +/// fetches (download / copy) and owner-private host transport. /// /// This client is a security boundary, not a convenience: it carries a minted /// media `Authorization` header, so it MUST NOT follow redirects. A relay 3xx -/// to an off-origin or private host would otherwise forward that header across -/// origins (a redirect-hop SSRF). `redirect::Policy::none()` returns the 3xx +/// must not forward bodies or make redirect-hop requests, regardless of header +/// stripping. `redirect::Policy::none()` returns the 3xx /// verbatim so the caller can reject it. /// /// Returned as a `Result` so the fail-closed invariant is testable — callers diff --git a/desktop/src-tauri/src/commands/host_execution.rs b/desktop/src-tauri/src/commands/host_execution.rs index 5ff16f5191..c1ab823337 100644 --- a/desktop/src-tauri/src/commands/host_execution.rs +++ b/desktop/src-tauri/src/commands/host_execution.rs @@ -88,7 +88,7 @@ pub async fn execute_host_command( let catalog = super::discover_acp_providers(app.clone(), Some(true)) .await .unwrap_or_default(); - let registrations = crate::relay::query_relay_at_with_keys(&state, + let registrations = crate::relay::query_private_host_at_with_keys(&state, &crate::relay::relay_http_base_url(&relay), &[serde_json::json!({"kinds": [50000], "ids": [registration_id], "#p": [expected_owner], "limit": 2})], &owner, None).await.map_err(|_| "cannot revalidate execution registration")?; diff --git a/desktop/src-tauri/src/commands/host_start.rs b/desktop/src-tauri/src/commands/host_start.rs index 8072750654..68e48b3cd4 100644 --- a/desktop/src-tauri/src/commands/host_start.rs +++ b/desktop/src-tauri/src/commands/host_start.rs @@ -1,7 +1,7 @@ //! Owner-operated Desktop Start transport. No host-key login and no source //! identity/config/file transfer. Command/receipt payloads are encrypted; routing //! envelopes and local retry bookkeeping contain no launch secrets. -use super::host_start_store::{retry_pending, Pending, Store}; +use super::host_start_store::{prepare_receipt, retry_pending, Pending, Store}; use crate::app_state::AppState; use buzz_core_pkg::{ host, @@ -28,7 +28,7 @@ pub(super) async fn current_registration( relay: &str, id: &str, ) -> Result { - let events = crate::relay::query_relay_at_with_keys(state, &crate::relay::relay_http_base_url(relay), + let events = crate::relay::query_private_host_at_with_keys(state, &crate::relay::relay_http_base_url(relay), &[serde_json::json!({"kinds":[50000], "ids":[id], "#p":[owner.public_key().to_hex()], "limit":2})], owner, None) .await.map_err(|_| "cannot verify destination registration")?; let [reg] = events.as_slice() else { @@ -170,7 +170,7 @@ pub(super) async fn history( ) -> Result, String> { let mut result = Vec::new(); loop { - let page = crate::relay::query_relay_at_with_keys( + let page = crate::relay::query_private_host_at_with_keys( state, &crate::relay::relay_http_base_url(relay), &[filter.clone()], @@ -229,7 +229,7 @@ async fn publish( let response = state .media_fetch_client .post(url) - .timeout(std::time::Duration::from_secs(15)) + .timeout(crate::relay::PRIVATE_HOST_REQUEST_TIMEOUT) .header("Authorization", auth) .header("Content-Type", "application/json") .body(bytes) @@ -367,12 +367,37 @@ pub async fn pump_host_start( ); store.save()?; } + // The durable publication owner fsyncs each prepared envelope before send. // One revoked/unpublishable operation must not starve another placement. - retry_pending(&mut store.journal.received, true, |pending| { - publish_pending(&state, &owner, &relay, pending, true) - }) - .await; - store.save()?; + store + .retry_receipts( + |mut pending| { + let (state, owner, host_keys, relay, receipts, expected_owner) = ( + &state, + &owner, + &host_keys, + &relay, + &receipts, + &expected_owner, + ); + async move { + current_registration(state, owner, relay, &pending.registration.id.to_hex()) + .await?; + scope(state, expected_owner, relay)?; + prepare_receipt( + &mut pending, + owner, + host_keys, + relay, + receipts, + Timestamp::now().as_secs(), + )?; + Ok(pending) + } + }, + |pending| publish_pending(&state, &owner, &relay, pending, true), + ) + .await?; retry_pending(&mut store.journal.sent, false, |pending| { publish_pending(&state, &owner, &relay, pending, false) }) @@ -638,6 +663,7 @@ async fn publish_pending( &pending.command }; current_registration(state, owner, relay, &pending.registration.id.to_hex()).await?; + scope(state, &owner.public_key().to_hex(), relay)?; publish(state, owner, relay, &pending.registration, event).await } diff --git a/desktop/src-tauri/src/commands/host_start_recovery_tests.rs b/desktop/src-tauri/src/commands/host_start_recovery_tests.rs new file mode 100644 index 0000000000..4a09fb94bb --- /dev/null +++ b/desktop/src-tauri/src/commands/host_start_recovery_tests.rs @@ -0,0 +1,428 @@ +//! Durable outbox/restart tests with a controllable relay admission clock. +//! The relay predicate mirrors ingest's +/-900 seconds; no DB/service is used. +use super::*; +use buzz_core_pkg::{ + host, + host_execution::{self, Action, Command, Outcome, Receipt}, +}; +use nostr::{JsonUtil, Keys}; + +fn observed(owner: &Keys, executor: &Keys) -> (Pending, Receipt) { + let registration = host::registration(owner, executor.public_key(), 99).unwrap(); + let request = Command { + v: 1, + operation: "ab".repeat(16), + relay: "wss://relay.example".into(), + agent: Keys::generate().public_key().to_hex(), + expires_at: 400, + action: Action::Start { + runtime: "buzz-agent".into(), + revision: "cd".repeat(32), + }, + }; + let command = host_execution::command(owner, ®istration, &request, 100).unwrap(); + let result = Receipt { + v: 1, + command: command.id.to_hex(), + run: request.run().into(), + request, + observed_at: 101, + outcome: Outcome::Spawned, + }; + let receipt = host_execution::receipt(executor, ®istration, &result, 101).unwrap(); + ( + Pending { + registration, + command, + receipt: Some(receipt), + published: false, + error: None, + supersedes: None, + }, + result, + ) +} + +// Exact timestamp condition of handlers/ingest.rs, plus real signature/routing +// checks. Kept here rather than altering relay admission to accommodate retries. +fn admit(p: &Pending, owner: &Keys, now: u64) -> Result<(), String> { + let event = p.receipt.as_ref().ok_or("missing")?; + host_execution::validate_transport(event, &p.registration, owner.public_key())?; + if now.abs_diff(event.created_at.as_secs()) > 900 { + return Err("timestamp drift".into()); + } + Ok(()) +} + +#[tokio::test] +async fn late_never_accepted_receipt_survives_restart_without_new_execution() { + let dir = tempfile::tempdir().unwrap(); + let owner = Keys::generate(); + let executor = Keys::generate(); + let (pending, observation) = observed(&owner, &executor); + let original = pending.receipt.clone().unwrap(); + let command = pending.command.as_json(); + let id = pending.command.id.to_hex(); + let relay = observation.request.relay.clone(); + let mut store = Store::open_dir(dir.path(), "owner", &relay).unwrap(); + store.journal.received.insert(id.clone(), pending); + store.save().unwrap(); + retry_pending(&mut store.journal.received, true, |_| async { + Err("never reached relay".into()) + }) + .await; + store.save().unwrap(); + drop(store); + + // More than the admission window AND the command lifetime, after restart. + let now = 101 + 901; + let mut store = Store::open_dir(dir.path(), "owner", &relay).unwrap(); + assert_eq!( + admit(&store.journal.received[&id], &owner, now), + Err("timestamp drift".into()) + ); + assert!( + host_execution::decrypt_command( + &executor, + &store.journal.received[&id].registration, + &store.journal.received[&id].command, + &relay, + now + ) + .is_err(), + "expired intent cannot execute" + ); + prepare_receipt( + store.journal.received.get_mut(&id).unwrap(), + &owner, + &executor, + &relay, + &[], + now, + ) + .unwrap(); + store.save().unwrap(); + let renewed = store.journal.received[&id].receipt.clone().unwrap(); + assert_ne!(renewed.id, original.id); + assert_eq!(renewed.created_at.as_secs(), now); + assert_eq!( + renewed.content, original.content, + "same encrypted proof, no new observation" + ); + assert_eq!(renewed.tags, original.tags); + assert_eq!(store.journal.received[&id].command.as_json(), command); + drop(store); // crash after preparing, before send + + let mut store = Store::open_dir(dir.path(), "owner", &relay).unwrap(); + let pending = store.journal.received.get_mut(&id).unwrap(); + prepare_receipt(pending, &owner, &executor, &relay, &[], now + 1).unwrap(); + assert_eq!( + pending.receipt.as_ref().unwrap().as_json(), + renewed.as_json() + ); + retry_pending(&mut store.journal.received, true, |p| { + let owner = &owner; + async move { admit(&p, owner, now + 1) } + }) + .await; + assert!(store.journal.received[&id].published); + let pending = &store.journal.received[&id]; + // What the source learns from accepted history is the original exact result. + assert_eq!( + host_execution::decrypt_receipt( + &owner, + &pending.registration, + &renewed, + &pending.command, + &observation.request + ) + .unwrap(), + observation + ); + store.save().unwrap(); + drop(store); + assert!( + Store::open_dir(dir.path(), "owner", &relay) + .unwrap() + .journal + .received[&id] + .published + ); +} + +#[tokio::test] +async fn accepted_but_ack_lost_uses_history_without_renewing_proof() { + let dir = tempfile::tempdir().unwrap(); + let owner = Keys::generate(); + let executor = Keys::generate(); + let (pending, observation) = observed(&owner, &executor); + let original = pending.receipt.clone().unwrap(); + let id = pending.command.id.to_hex(); + let relay = &observation.request.relay; + let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap(); + store.journal.received.insert(id.clone(), pending); + store.save().unwrap(); + store + .retry_receipts( + |p| std::future::ready(Ok(p)), + |p| { + let owner = &owner; + async move { + admit(&p, owner, 101)?; + Err("accepted, but ACK lost".into()) + } + }, + ) + .await + .unwrap(); + drop(store); + let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap(); + store + .retry_receipts( + |mut p| { + let result = prepare_receipt( + &mut p, + &owner, + &executor, + relay, + std::slice::from_ref(&original), + 2000, + ) + .map(|()| p); + std::future::ready(result) + }, + |_| async { panic!("accepted history must suppress another publication") }, + ) + .await + .unwrap(); + assert!(store.journal.received[&id].published); + assert_eq!( + store.journal.received[&id] + .receipt + .as_ref() + .unwrap() + .as_json(), + original.as_json() + ); +} + +#[test] +fn history_of_an_earlier_envelope_also_resolves_ambiguous_delivery() { + let owner = Keys::generate(); + let executor = Keys::generate(); + let (mut pending, observation) = observed(&owner, &executor); + let original = pending.receipt.clone().unwrap(); + prepare_receipt( + &mut pending, + &owner, + &executor, + &observation.request.relay, + &[], + 2000, + ) + .unwrap(); + let renewed = pending.receipt.clone().unwrap(); + prepare_receipt( + &mut pending, + &owner, + &executor, + &observation.request.relay, + &[original], + 3000, + ) + .unwrap(); + assert!(pending.published); + assert_eq!(pending.receipt.unwrap(), renewed); +} + +#[test] +fn recovery_rejects_wrong_signer_scope_registration_missing_and_tampered_evidence() { + let owner = Keys::generate(); + let executor = Keys::generate(); + let (pending, observation) = observed(&owner, &executor); + let relay = &observation.request.relay; + assert!(prepare_receipt( + &mut pending.clone(), + &owner, + &Keys::generate(), + relay, + &[], + 2000 + ) + .is_err()); + assert!(prepare_receipt( + &mut pending.clone(), + &Keys::generate(), + &executor, + relay, + &[], + 2000 + ) + .is_err()); + assert!(prepare_receipt( + &mut pending.clone(), + &owner, + &executor, + "wss://elsewhere.example", + &[], + 2000 + ) + .is_err()); + let mut changed = pending.clone(); + changed.registration = host::registration(&owner, executor.public_key(), 100).unwrap(); + assert!(prepare_receipt(&mut changed, &owner, &executor, relay, &[], 2000).is_err()); + let mut changed = pending.clone(); + changed.receipt = None; + assert!(prepare_receipt(&mut changed, &owner, &executor, relay, &[], 2000).is_err()); + let mut changed = pending; + changed.receipt.as_mut().unwrap().content.push('x'); + assert!(prepare_receipt(&mut changed, &owner, &executor, relay, &[], 2000).is_err()); +} + +#[test] +fn forged_history_cannot_confirm_delivery_and_invalid_lifetime_cannot_be_renewed() { + let owner = Keys::generate(); + let executor = Keys::generate(); + let (mut pending, mut observation) = observed(&owner, &executor); + let original = pending.receipt.clone().unwrap(); + let forged = nostr::EventBuilder::new(original.kind, original.content.clone()) + .tags(original.tags.clone()) + .allow_self_tagging() + .sign_with_keys(&Keys::generate()) + .unwrap(); + prepare_receipt( + &mut pending, + &owner, + &executor, + &observation.request.relay, + &[forged], + 1002, + ) + .unwrap(); + assert!(!pending.published); + assert_eq!(pending.receipt.as_ref().unwrap().created_at.as_secs(), 1002); + observation.request.expires_at = 401; // original creation + COMMAND_TTL + 1 + let content = nostr::nips::nip44::encrypt( + owner.secret_key(), + &executor.public_key(), + serde_json::to_string(&observation.request).unwrap(), + nostr::nips::nip44::Version::V2, + ) + .unwrap(); + pending.command = nostr::EventBuilder::new(pending.command.kind, content) + .tags(pending.command.tags.clone()) + .allow_self_tagging() + .custom_created_at(nostr::Timestamp::from(100)) + .sign_with_keys(&owner) + .unwrap(); + observation.command = pending.command.id.to_hex(); + pending.receipt = + Some(host_execution::receipt(&executor, &pending.registration, &observation, 101).unwrap()); + assert!(prepare_receipt( + &mut pending, + &owner, + &executor, + &observation.request.relay, + &[], + 1002 + ) + .is_err()); +} + +#[tokio::test] +async fn publication_owner_persists_before_send_and_recovers_late_unaccepted_after_restart() { + let dir = tempfile::tempdir().unwrap(); + let owner = Keys::generate(); + let executor = Keys::generate(); + let (pending, observation) = observed(&owner, &executor); + let relay = &observation.request.relay; + let id = pending.command.id.to_hex(); + let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap(); + store.journal.received.insert(id.clone(), pending.clone()); + // An earlier revoked entry must not starve this one. + store.journal.received.insert("a-revoked".into(), pending); + store.save().unwrap(); + let disk = store.path.clone(); + let prepare = |mut p: Pending, now| { + prepare_receipt(&mut p, &owner, &executor, relay, &[], now)?; + Ok(p) + }; + store + .retry_receipts( + |p| std::future::ready(prepare(p, 101)), + |_| async { Err("never accepted: disconnected".into()) }, + ) + .await + .unwrap(); + drop(store); + let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap(); + store + .journal + .received + .get_mut("a-revoked") + .unwrap() + .registration = host::registration(&owner, executor.public_key(), 100).unwrap(); + let attempts = std::sync::atomic::AtomicUsize::new(0); + store + .retry_receipts( + |p| std::future::ready(prepare(p, 1002)), + |p| { + let (disk, id, owner, attempts) = (&disk, &id, &owner, &attempts); + async move { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let durable: Journal = + serde_json::from_slice(&fs::read(disk).unwrap()).unwrap(); + assert_eq!( + durable.received[id].receipt, p.receipt, + "fsync before publication" + ); + admit(&p, owner, 1002) + } + }, + ) + .await + .unwrap(); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1); + assert!(store.journal.received[&id].published); + assert!(!store.journal.received["a-revoked"].published); + assert!(store.journal.received["a-revoked"].error.is_some()); + let p = &store.journal.received[&id]; + assert_eq!( + host_execution::decrypt_receipt( + &owner, + &p.registration, + p.receipt.as_ref().unwrap(), + &p.command, + &observation.request + ) + .unwrap(), + observation + ); +} + +#[tokio::test] +async fn failed_outbox_save_never_sends_a_renewed_envelope() { + let dir = tempfile::tempdir().unwrap(); + let owner = Keys::generate(); + let executor = Keys::generate(); + let (pending, observation) = observed(&owner, &executor); + let relay = &observation.request.relay; + let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap(); + store + .journal + .received + .insert(pending.command.id.to_hex(), pending); + store.save().unwrap(); + store.path = dir.path().into(); // cannot atomically replace a directory + let result = store + .retry_receipts( + |mut p| { + let result = + prepare_receipt(&mut p, &owner, &executor, relay, &[], 1002).map(|()| p); + std::future::ready(result) + }, + |_| async { panic!("publication must not precede durable save") }, + ) + .await; + assert!(result.is_err()); +} diff --git a/desktop/src-tauri/src/commands/host_start_store.rs b/desktop/src-tauri/src/commands/host_start_store.rs index 6c87618f6b..e966dbb1eb 100644 --- a/desktop/src-tauri/src/commands/host_start_store.rs +++ b/desktop/src-tauri/src/commands/host_start_store.rs @@ -1,4 +1,5 @@ -//! Encrypted, immutable transport outbox. Execution authority remains in the +//! Encrypted transport outbox with immutable command and observation payloads. +//! Only a receipt transport timestamp/signature may be renewed. Authority remains in the //! per-placement journal; transport retry must never create a new operation. use nostr::Event; use serde::{Deserialize, Serialize}; @@ -77,6 +78,53 @@ impl Store { }) } + /// Prepare/authenticate one durable result, fsync its envelope, then publish. + /// No later entry can age an earlier prepared envelope while doing network + /// reads. A failed entry remains visible and never starves other placements. + pub async fn retry_receipts< + Prepared: std::future::Future> + Send, + Published: std::future::Future> + Send, + >( + &mut self, + mut prepare: impl FnMut(Pending) -> Prepared, + mut publish: impl FnMut(Pending) -> Published, + ) -> Result<(), String> { + let ids: Vec<_> = self.journal.received.keys().cloned().collect(); + for id in ids { + let Some(pending) = self.journal.received.get(&id).cloned() else { + continue; + }; + if pending.published { + continue; + } + match prepare(pending.clone()).await { + Ok(mut ready) => { + // Even a crash inside publish leaves exactly these signed + // bytes durable. Keep the original observation ciphertext. + self.journal.received.insert(id.clone(), ready.clone()); + self.save()?; + if !ready.published { + match publish(ready.clone()).await { + Ok(()) => { + ready.published = true; + ready.error = None; + } + Err(error) => ready.error = Some(error), + } + } + self.journal.received.insert(id, ready); + } + Err(error) => { + let mut failed = pending; + failed.error = Some(error); + self.journal.received.insert(id, failed); + } + } + self.save()?; + } + Ok(()) + } + pub fn save(&self) -> Result<(), String> { if self.journal.sent.len() + self.journal.received.len() + self.journal.moves.len() > 4096 { return Err("Start outbox requires archival".into()); @@ -93,9 +141,82 @@ impl Store { } } +/// Recover delivery of an already durable observation, never execution. The caller +/// saves the journal BEFORE sending any renewed envelope and still revalidates +/// current registration/owner authority at publication. Keep ciphertext byte-exact: +/// even ambiguous acceptance can only yield the same proof, not a new observation. +pub(super) fn prepare_receipt( + pending: &mut Pending, + owner: &nostr::Keys, + host: &nostr::Keys, + relay: &str, + history: &[Event], + now: u64, +) -> Result<(), String> { + use buzz_core_pkg::host_execution; + if pending.published { + return Ok(()); + } + // Authenticate the original command at its signed time, including its bounded + // lifetime. This is read-only evidence recovery, not admission of an expired + // intent to the executor. Wrong host/community/registration fail closed. + let request = host_execution::decrypt_command( + host, + &pending.registration, + &pending.command, + relay, + pending.command.created_at.as_secs(), + )?; + let event = pending + .receipt + .as_ref() + .ok_or("missing durable Start receipt")?; + host_execution::decrypt_receipt( + owner, + &pending.registration, + event, + &pending.command, + &request, + )?; + if event.created_at.as_secs() > now.saturating_add(30) { + return Err("receipt timestamp is in the future".into()); + } + // ACK loss is not non-acceptance. Match immutable ciphertext/routing, not only + // event ID: an earlier envelope may have committed before a later retry. + if history.iter().any(|accepted| { + accepted.content == event.content + && accepted.tags == event.tags + && host_execution::decrypt_receipt( + owner, + &pending.registration, + accepted, + &pending.command, + &request, + ) + .is_ok() + }) { + pending.published = true; + pending.error = None; + return Ok(()); + } + // Renew before the relay's +/-900s admission window, leaving time for I/O. + // Recent retries reuse exact bytes; observation/request/run remain unchanged. + if now.saturating_sub(event.created_at.as_secs()) >= 600 { + pending.receipt = Some( + nostr::EventBuilder::new(event.kind, event.content.clone()) + .allow_self_tagging() + .tags(event.tags.clone()) + .custom_created_at(nostr::Timestamp::from(now)) + .sign_with_keys(host) + .map_err(|_| "receipt transport signing failed")?, + ); + } + Ok(()) +} + /// Attempt every eligible entry independently. Failed publication remains pending /// and visible; an ACK is recorded only for that exact event. Caller fsyncs before -/// returning. Dropped ACK/restart consequently resends immutable bytes. +/// returning. Receipt preparation is persisted separately before this send loop. pub(super) async fn retry_pending> + Send>( entries: &mut BTreeMap, receipts: bool, @@ -221,3 +342,7 @@ mod tests { } } } + +#[cfg(test)] +#[path = "host_start_recovery_tests.rs"] +mod recovery_tests; diff --git a/desktop/src-tauri/src/commands/hosts.rs b/desktop/src-tauri/src/commands/hosts.rs index 851a455966..7c1974e280 100644 --- a/desktop/src-tauri/src/commands/hosts.rs +++ b/desktop/src-tauri/src/commands/hosts.rs @@ -471,7 +471,7 @@ pub async fn get_host_history_page( { return Err("Invalid host history scope".into()); } - let events = crate::relay::query_relay_at_with_keys( + let events = crate::relay::query_private_host_at_with_keys( &state, &crate::relay::relay_http_base_url(&relay_url), &[filter], diff --git a/desktop/src-tauri/src/managed_agents/execution.rs b/desktop/src-tauri/src/managed_agents/execution.rs index 331b64da46..04d24b4cd8 100644 --- a/desktop/src-tauri/src/managed_agents/execution.rs +++ b/desktop/src-tauri/src/managed_agents/execution.rs @@ -43,9 +43,7 @@ pub(crate) fn local_execution_config( app: &AppHandle, record: &ManagedAgentRecord, ) -> Result { - if record.backend != BackendKind::Local { - return Err("destination agent is not locally provisioned".into()); - } + local_execution_prerequisites(record)?; let personas = load_personas(app)?; let teams = load_teams(app)?; let global = load_global_agent_config(app)?; @@ -76,6 +74,18 @@ pub(crate) fn local_execution_config( }) } +// Shared by inspection/advertisement and the launch preflight. Only key +// availability is checked, exactly as at spawn; no key parsing/export is needed. +fn local_execution_prerequisites(record: &ManagedAgentRecord) -> Result<(), String> { + if record.backend != BackendKind::Local { + return Err("destination agent is not locally provisioned".into()); + } + if let Some(error) = super::storage::spawn_key_refusal(record) { + return Err(error); + } + Ok(()) +} + /// Remote execution requires an explicit verified owner attestation, even for /// legacy local records. A local key in the store is not a remote launch grant. pub(crate) fn execution_agent_owner( @@ -437,6 +447,24 @@ fn exact_generation_matches(actual: &str, expected: &str) -> bool { #[cfg(test)] mod tests { use super::*; + #[test] + fn keyless_record_is_not_advertised_as_locally_provisioned() { + let mut record: ManagedAgentRecord = serde_json::from_value(serde_json::json!({ + "pubkey": "synthetic-agent", "name": "test-agent", "private_key_nsec": "", + "relay_url": "wss://relay.example", "acp_command": "buzz-acp", + "agent_command": "buzz-agent", "agent_args": [], "mcp_command": "", + "turn_timeout_seconds": 320, "created_at": "", "updated_at": "" + })) + .unwrap(); + assert_eq!( + local_execution_prerequisites(&record).unwrap_err(), + super::super::storage::spawn_key_refusal(&record).unwrap() + ); + // Availability only: this preflight must not inspect or export key bytes. + record.private_key_nsec = "synthetic-nonempty-placeholder".into(); + assert!(local_execution_prerequisites(&record).is_ok()); + } + #[test] fn stop_fences_successor_and_malformed_or_legacy_generation() { assert!(exact_generation_matches(&"aa".repeat(16), &"aa".repeat(16))); diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs index f408ef2afd..1415617484 100644 --- a/desktop/src-tauri/src/relay.rs +++ b/desktop/src-tauri/src/relay.rs @@ -350,6 +350,9 @@ pub async fn relay_error_message(response: reqwest::Response) -> String { format!("relay returned {status}") } +mod private_host; +pub(crate) use private_host::{query_private_host_at_with_keys, PRIVATE_HOST_REQUEST_TIMEOUT}; + // ── HTTP bridge: POST /query ──────────────────────────────────────────────── /// Execute a one-shot query via the relay's HTTP bridge (`POST /query`). diff --git a/desktop/src-tauri/src/relay/private_host.rs b/desktop/src-tauri/src/relay/private_host.rs new file mode 100644 index 0000000000..e69891bdd2 --- /dev/null +++ b/desktop/src-tauri/src/relay/private_host.rs @@ -0,0 +1,41 @@ +//! Selected-owner private host inventory, authorization and execution history. +//! These POST bodies must never follow redirects, even when sensitive HTTP +//! headers would be stripped: the filters themselves disclose private metadata. +use super::*; + +/// Per-request deadline includes response-body consumption, not rate-limit wait. +/// Private host reads/writes deliberately use a shorter budget than WS history. +pub(crate) const PRIVATE_HOST_REQUEST_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(15); + +/// Read private host records with the current unlocked owner's authority, using +/// the same no-redirect client and deadline as host execution publication. +pub(crate) async fn query_private_host_at_with_keys( + state: &AppState, + api_base_url: &str, + filters: &[serde_json::Value], + keys: &Keys, + auth_tag: Option<&str>, +) -> Result, String> { + crate::relay_admission::wait_for_rate_limit().await; + assert_expected_signer( + Some(&keys.public_key().to_hex()), + &state.signing_keys()?.public_key().to_hex(), + )?; + let url = format!("{}/query", api_base_url.trim_end_matches('/')); + let body = + serde_json::to_vec(filters).map_err(|e| format!("filter serialization failed: {e}"))?; + let auth = build_nip98_auth_header_for_keys(keys, &Method::POST, &url, &body)?; + send_query_request( + &state.media_fetch_client, + &url, + &auth, + auth_tag, + body, + PRIVATE_HOST_REQUEST_TIMEOUT, + ) + .await +} + +#[cfg(test)] +mod tests; diff --git a/desktop/src-tauri/src/relay/private_host/tests.rs b/desktop/src-tauri/src/relay/private_host/tests.rs new file mode 100644 index 0000000000..0601671631 --- /dev/null +++ b/desktop/src-tauri/src/relay/private_host/tests.rs @@ -0,0 +1,187 @@ +use super::*; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; + +async fn read_request(stream: &mut TcpStream) -> (String, Vec) { + let mut bytes = Vec::new(); + loop { + let mut buf = [0; 4096]; + let n = stream.read(&mut buf).await.unwrap(); + assert_ne!(n, 0); + bytes.extend_from_slice(&buf[..n]); + if let Some(end) = bytes.windows(4).position(|b| b == b"\r\n\r\n") { + let headers = String::from_utf8(bytes[..end].to_vec()).unwrap(); + let len: usize = headers + .lines() + .find_map(|line| { + let (key, value) = line.split_once(':')?; + key.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse().unwrap()) + }) + .unwrap(); + if bytes.len() >= end + 4 + len { + return (headers, bytes[end + 4..end + 4 + len].to_vec()); + } + } + } +} + +fn state_and_keys() -> (AppState, Keys) { + let state = crate::app_state::build_app_state(); + let keys = Keys::generate(); + *state.keys.lock().unwrap() = keys.clone(); + (state, keys) +} + +#[tokio::test] +async fn private_host_redirects_never_contact_target_or_disclose_filters() { + let _serial = crate::relay_admission::TEST_SERIAL.lock().await; + crate::relay_admission::reset_rate_limit_gate(); + for status in [307, 308] { + let origin = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let target = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", origin.local_addr().unwrap()); + let target_url = format!("http://{}/private-leak", target.local_addr().unwrap()); + let (state, keys) = state_and_keys(); + let filters = + [serde_json::json!({"kinds":[50002],"#p":[keys.public_key().to_hex()],"limit":1000})]; + let expected_body = serde_json::to_vec(&filters).unwrap(); + let server = tokio::spawn(async move { + let (mut stream, _) = origin.accept().await.unwrap(); + let (_, body) = read_request(&mut stream).await; + assert_eq!(body, expected_body); + stream.write_all(format!("HTTP/1.1 {status} Redirect\r\nLocation: {target_url}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap(); + }); + let result = tokio::time::timeout( + Duration::from_secs(3), + query_private_host_at_with_keys(&state, &base, &filters, &keys, None), + ) + .await + .unwrap(); + assert!(result.unwrap_err().contains(&status.to_string())); + server.await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(100), target.accept()) + .await + .is_err(), + "redirect target must receive no request, including no filter body" + ); + } +} + +#[tokio::test] +async fn private_host_configured_auth_is_bound_to_exact_url_and_body() { + let _serial = crate::relay_admission::TEST_SERIAL.lock().await; + crate::relay_admission::reset_rate_limit_gate(); + let origin = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", origin.local_addr().unwrap()); + let url = format!("{base}/query"); + let (state, keys) = state_and_keys(); + let signer = keys.public_key(); + let server = tokio::spawn(async move { + let (mut stream, _) = origin.accept().await.unwrap(); + let (headers, body) = read_request(&mut stream).await; + let header = |name: &str| { + headers + .lines() + .find_map(|line| { + let (key, value) = line.split_once(':')?; + key.eq_ignore_ascii_case(name).then(|| value.trim()) + }) + .unwrap() + }; + assert_eq!(header("x-auth-tag"), "synthetic-configured-auth"); + let encoded = header("authorization").strip_prefix("Nostr ").unwrap(); + let bytes = base64::engine::general_purpose::STANDARD + .decode(encoded) + .unwrap(); + let auth = nostr::Event::from_json(bytes).unwrap(); + buzz_core_pkg::verify_event(&auth).unwrap(); + assert_eq!(auth.pubkey, signer); + assert!(auth.tags.iter().any(|t| t.as_slice() == ["u", &url])); + assert!(auth.tags.iter().any(|t| t.as_slice() == ["method", "POST"])); + assert!(auth + .tags + .iter() + .any(|t| t.as_slice() == ["payload", &hex::encode(Sha256::digest(&body))])); + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n[]").await.unwrap(); + }); + assert!(query_private_host_at_with_keys( + &state, + &base, + &[serde_json::json!({"kinds":[50000],"#p":[signer.to_hex()]})], + &keys, + Some("synthetic-configured-auth") + ) + .await + .unwrap() + .is_empty()); + server.await.unwrap(); +} + +#[tokio::test] +async fn private_host_deadline_covers_stalled_headers_and_body() { + // Exercise the actual 15s policy (no test-only shorter timeout). + let _serial = crate::relay_admission::TEST_SERIAL.lock().await; + crate::relay_admission::reset_rate_limit_gate(); + assert_eq!(PRIVATE_HOST_REQUEST_TIMEOUT, Duration::from_secs(15)); + for send_headers in [false, true] { + let origin = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", origin.local_addr().unwrap()); + let (state, keys) = state_and_keys(); + let server = tokio::spawn(async move { + let (mut stream, _) = origin.accept().await.unwrap(); + read_request(&mut stream).await; + if send_headers { + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 50\r\n\r\n[").await.unwrap(); + } + std::future::pending::<()>().await; + }); + let start = std::time::Instant::now(); + let result = tokio::time::timeout( + Duration::from_secs(20), + query_private_host_at_with_keys( + &state, + &base, + &[serde_json::json!({"kinds":[50002]})], + &keys, + None, + ), + ) + .await + .unwrap(); + assert_eq!(result.unwrap_err(), "relay unreachable: request timed out"); + assert!(start.elapsed() >= Duration::from_secs(14)); + server.abort(); + let _ = server.await; + } +} + +#[tokio::test] +async fn private_host_locked_or_changed_identity_has_no_egress() { + let _serial = crate::relay_admission::TEST_SERIAL.lock().await; + crate::relay_admission::reset_rate_limit_gate(); + let origin = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", origin.local_addr().unwrap()); + let (state, keys) = state_and_keys(); + for gate in [&state.keyring_locked, &state.identity_lost] { + gate.store(true, std::sync::atomic::Ordering::Release); + assert!( + query_private_host_at_with_keys(&state, &base, &[], &keys, None) + .await + .is_err() + ); + gate.store(false, std::sync::atomic::Ordering::Release); + } + assert!( + query_private_host_at_with_keys(&state, &base, &[], &Keys::generate(), None) + .await + .is_err() + ); + assert!( + tokio::time::timeout(Duration::from_millis(100), origin.accept()) + .await + .is_err() + ); +} diff --git a/docs/host-execution.md b/docs/host-execution.md index b0b69983d7..2fb34014ae 100644 --- a/docs/host-execution.md +++ b/docs/host-execution.md @@ -4,7 +4,8 @@ The owner-operated native Start outbox, receiver and Hosts picker are implemente behind the **non-default `remote-start-preview` Cargo feature**. Default builds advertise `accepts_start: false` and reject queuing. Even a preview build advertises Start only with a successful same-owner/community receiver pump in the last 15s -and at least one destination-local provisioned, compatible configuration. A +and at least one destination-local provisioned, compatible configuration with an +available agent key (the same availability refusal as launch). A runtime catalog entry alone is not launch capability. This is not a release or physical-host/provider certification. @@ -64,7 +65,24 @@ immutable ledger replay and before creating any new intent or side effect. Futur timestamps are rejected. Expiration never creates a replacement operation. The native outbox uses restricted atomic writes, directory fsync and OS file locks. -It retains signed commands and receipts through ACK loss/reconnect/restart. +It retains signed commands and receipt ciphertext through ACK loss/reconnect/restart. +Before retrying a durable receipt, the destination checks accepted history. If the +result never reached the relay and its envelope is at least 10 minutes old, it +re-signs the **same encrypted observation bytes and routing tags** with a fresh +transport timestamp and fsyncs before sending. Original command ID, operation, +run, outcome and observation time do not change. Recent retries reuse exact +signed bytes; accepted history of any earlier envelope resolves ACK loss without +creating another observation. Only a saved, authenticated result may be renewed; +this never executes an expired intent or opens a new ledger operation. Current +registration and unlocked owner/community authority are still required. + +Owner-private host registration/inventory and execution history reads use the +same native no-redirect client as Start publication. Their 15-second per-request +deadline includes response-body consumption (rate-limit admission wait is +separate). Redirects are errors, including 307/308: neither private filter bodies +nor authentication are sent to a redirect target. Unrelated query transports +retain their existing policy. + Publication errors are recorded per entry, remain visible and retry independently; a revoked old registration does not starve another operation. Corrupt bindings, missing receipts and invalid/cyclic supersession chains fail closed. Transport ACK