diff --git a/apps/maple-agent/Cargo.lock b/apps/maple-agent/Cargo.lock index 29169cbad..e0c64b203 100644 --- a/apps/maple-agent/Cargo.lock +++ b/apps/maple-agent/Cargo.lock @@ -5766,7 +5766,7 @@ dependencies = [ [[package]] name = "maple-agent" -version = "0.0.8" +version = "0.0.9" dependencies = [ "agent-client-protocol", "anyhow", @@ -5818,7 +5818,7 @@ dependencies = [ [[package]] name = "maple-agent-app" -version = "0.0.8" +version = "0.0.9" dependencies = [ "axum", "base64 0.22.1", @@ -5874,7 +5874,7 @@ dependencies = [ [[package]] name = "maple-agent-updater" -version = "0.0.8" +version = "0.0.9" dependencies = [ "anyhow", "base64 0.22.1", @@ -5902,7 +5902,7 @@ dependencies = [ [[package]] name = "maple-billing" -version = "0.0.8" +version = "0.0.9" dependencies = [ "reqwest 0.13.4", "serde", @@ -7680,7 +7680,7 @@ dependencies = [ [[package]] name = "pi-agent-core" -version = "0.0.8" +version = "0.0.9" dependencies = [ "async-trait", "futures-util", @@ -7693,7 +7693,7 @@ dependencies = [ [[package]] name = "pi-ai" -version = "0.0.8" +version = "0.0.9" dependencies = [ "async-trait", "bytes", @@ -7710,7 +7710,7 @@ dependencies = [ [[package]] name = "pi-coding-agent" -version = "0.0.8" +version = "0.0.9" dependencies = [ "async-trait", "base64 0.22.1", @@ -10103,7 +10103,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "store-bench" -version = "0.0.8" +version = "0.0.9" dependencies = [ "anyhow", "maple-agent", diff --git a/apps/maple-agent/Cargo.toml b/apps/maple-agent/Cargo.toml index 56bd90534..7dfce95ee 100644 --- a/apps/maple-agent/Cargo.toml +++ b/apps/maple-agent/Cargo.toml @@ -4,7 +4,7 @@ members = ["crates/maple-agent", "crates/maple-billing", "crates/maple-agent-upd [workspace.package] edition = "2024" -version = "0.0.8" +version = "0.0.9" license = "MIT" [workspace.dependencies] diff --git a/apps/maple-agent/app/src/desktop.rs b/apps/maple-agent/app/src/desktop.rs index 0bcbe3e9e..3c58aec40 100644 --- a/apps/maple-agent/app/src/desktop.rs +++ b/apps/maple-agent/app/src/desktop.rs @@ -3245,6 +3245,7 @@ pub fn run() { let expected_account = account.clone(); let bridge = update_bridge.clone(); let update_status = remote_updater.remote_status(); + let check_updater = remote_updater.clone(); let task = status_backend.spawn(async move { tokio::join!( backend.remote_setting::( @@ -3252,7 +3253,12 @@ pub fn run() { () ), backend.local_control::("store_status", ()), - bridge.poll(backend.clone(), expected_account, update_status), + bridge.poll( + backend.clone(), + expected_account, + update_status, + check_updater + ), ) }); if let Ok((remote, store, update_command)) = task.await { diff --git a/apps/maple-agent/app/src/desktop/remote_updates.rs b/apps/maple-agent/app/src/desktop/remote_updates.rs index f08b3bc1b..4f71b6359 100644 --- a/apps/maple-agent/app/src/desktop/remote_updates.rs +++ b/apps/maple-agent/app/src/desktop/remote_updates.rs @@ -11,7 +11,8 @@ use crate::{ backend::AgentBackend, remote::update::{ReleaseIdentity, RemoteUpdateStatus}, remote_updates::{ - AttachRequest, CompleteRequest, DesktopCommand, DesktopLease, PublishRequest, TakeRequest, + AttachRequest, CompleteRequest, DesktopCommand, DesktopLease, DesktopRefreshCommand, + PublishRequest, RefreshCompleteRequest, RefreshFailure, TakeRequest, }, update::{Controller, Snapshot, State}, }; @@ -37,6 +38,13 @@ pub(super) struct Pending { bridge: Bridge, } +struct PendingRefresh { + account: String, + lease: String, + command: DesktopRefreshCommand, + bridge: Bridge, +} + impl Bridge { /// Called by the existing five-second local status pump. No new timer or /// long-polling host request keeps an account/admission guard alive. @@ -45,18 +53,104 @@ impl Bridge { backend: Arc, account: Option, status: RemoteUpdateStatus, + updater: Controller, ) -> Option { let account = account.filter(|_| !backend.is_retired_after_update()); - self.poll_with( - account, - status, + let caller = backend.clone(); + let pending = self + .poll_with( + account.clone(), + status, + Duration::from_secs(5), + move |method, params| { + let backend = caller.clone(); + async move { backend.remote_setting(method, params).await } + }, + ) + .await; + if pending.is_none() + && let Some(refresh) = self + .take_refresh_with(account, { + let backend = backend.clone(); + move |method, params| { + let backend = backend.clone(); + async move { backend.remote_setting(method, params).await } + } + }) + .await + { + let work_backend = backend.clone(); + let download_backend = backend.clone(); + drop(backend.spawn(async move { + let expected = refresh.account.clone(); + let check_account = expected.clone(); + let check_backend = work_backend.clone(); + let check_updater = updater.clone(); + let completed = refresh + .run_with( + move |method, params| { + let backend = work_backend.clone(); + async move { backend.remote_setting(method, params).await } + }, + async move { + if check_backend.is_retired_after_update() + || check_backend.saved_user_id().as_deref() + != Some(check_account.as_str()) + || check_backend.store_state() + != maple_agent::store::StoreState::Open + { + return Err(RefreshFailure::Unavailable); + } + check_updater.refresh_remote().await + }, + ) + .await; + if completed + && download_backend.saved_user_id().as_deref() == Some(expected.as_str()) + && !download_backend.is_retired_after_update() + { + // Follow the execution desktop's saved download preference; + // this never installs or restarts either computer. + updater.download_if_automatic().await; + } + })); + } + pending + } + + async fn take_refresh_with( + &self, + account: Option, + mut call: C, + ) -> Option + where + C: FnMut(&'static str, serde_json::Value) -> F, + F: Future>, + { + let registration = self.0.lock().await; + let current = registration + .as_ref() + .filter(|current| Some(¤t.account) == account.as_ref())?; + let command: Option = tokio::time::timeout( Duration::from_secs(5), - move |method, params| { - let backend = backend.clone(); - async move { backend.remote_setting(method, params).await } - }, + request( + &mut call, + "remote_update_refresh_take", + TakeRequest { + expected_account: current.account.clone(), + lease: current.lease.id.clone(), + }, + ), ) .await + .ok()? + .ok()?; + command.map(|command| PendingRefresh { + account: current.account.clone(), + lease: current.lease.id.clone(), + command, + bridge: self.clone(), + }) } async fn poll_with( @@ -162,6 +256,47 @@ impl Bridge { } } +impl PendingRefresh { + async fn owned(&self) -> bool { + self.bridge.0.lock().await.as_ref().is_some_and(|current| { + current.account == self.account && current.lease.id == self.lease + }) + } + + async fn run_with(self, mut call: C, check: Check) -> bool + where + C: FnMut(&'static str, serde_json::Value) -> F, + F: Future>, + Check: Future>, + { + if !self.owned().await { + return false; + } + let result = check.await; + if !self.owned().await { + return false; + } + let checked = result.is_ok(); + matches!( + tokio::time::timeout( + Duration::from_secs(5), + request::<(), _, _>( + &mut call, + "remote_update_refresh_complete", + RefreshCompleteRequest { + expected_account: self.account, + lease: self.lease, + id: self.command.id, + result, + }, + ) + ) + .await, + Ok(Ok(())) + ) && checked + } +} + async fn request( call: &mut C, method: &'static str, @@ -326,6 +461,136 @@ mod tests { } } + async fn refresh_call( + broker: Arc, + method: &'static str, + params: serde_json::Value, + ) -> Result { + match method { + "remote_update_refresh_take" => serde_json::to_value( + broker + .take_refresh(serde_json::from_value(params).unwrap()) + .await?, + ) + .map_err(|_| "fixture response".into()), + "remote_update_refresh_complete" => { + broker + .complete_refresh(serde_json::from_value(params).unwrap()) + .await?; + Ok(serde_json::Value::Null) + } + _ => panic!("unexpected refresh RPC {method}"), + } + } + + #[tokio::test] + async fn queued_refresh_invokes_check_once_and_returns_its_fresh_result() { + use std::sync::atomic::{AtomicUsize, Ordering}; + for outcome in [ + Ok(RemoteUpdateStatus { + state: UpdateState::Current, + ..Default::default() + }), + Ok(RemoteUpdateStatus { + state: UpdateState::Available, + offered: Some(target()), + ..Default::default() + }), + Err(RefreshFailure::CheckFailed), + Err(RefreshFailure::Busy), + ] { + let directory = tempfile::tempdir().unwrap(); + let store = Arc::new( + maple_agent::store::Store::open(maple_agent::store::StorePaths::new( + directory.path(), + )) + .unwrap(), + ); + let broker = Arc::new(crate::remote_updates::Broker::new(Some(store))); + let lease = broker.attach("alice".into()).await.unwrap(); + let bridge = Bridge::default(); + *bridge.0.lock().await = Some(Registration { + account: "alice".into(), + lease, + published: Some(ready_status()), + completion: None, + }); + let waiting = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .request_refresh("alice".into(), "paired-peer".into()) + .await + } + }); + let pending = tokio::time::timeout(Duration::from_secs(2), async { + loop { + let caller = broker.clone(); + if let Some(pending) = bridge + .take_refresh_with(Some("alice".into()), move |method, params| { + refresh_call(caller.clone(), method, params) + }) + .await + { + break pending; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(!waiting.is_finished()); + let calls = AtomicUsize::new(0); + let expected = outcome.clone(); + let caller = broker.clone(); + let completed = pending + .run_with( + move |method, params| refresh_call(caller.clone(), method, params), + async { + calls.fetch_add(1, Ordering::SeqCst); + outcome + }, + ) + .await; + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(completed, expected.is_ok()); + let reply = waiting.await.unwrap(); + match expected { + Ok(status) => assert_eq!(reply.unwrap(), status), + Err(error) => assert_eq!(reply.unwrap_err(), error.message()), + } + } + } + + #[tokio::test] + async fn changed_desktop_owner_does_not_invoke_or_complete_an_old_refresh() { + let bridge = Bridge::default(); + let pending = PendingRefresh { + account: "old-account".into(), + lease: "old-lease".into(), + command: DesktopRefreshCommand { + id: "old-check".into(), + }, + bridge: bridge.clone(), + }; + *bridge.0.lock().await = Some(Registration { + account: "new-account".into(), + lease: DesktopLease { + id: "new-lease".into(), + }, + published: None, + completion: None, + }); + assert!( + !pending + .run_with( + |_, _| async { panic!("stale completion must not be sent") }, + async { panic!("stale check must not run") }, + ) + .await + ); + } + enum CompletionDelivery { LoseAcknowledgement, TimeoutBeforeWrite, diff --git a/apps/maple-agent/app/src/host.rs b/apps/maple-agent/app/src/host.rs index efda2e957..147954967 100644 --- a/apps/maple-agent/app/src/host.rs +++ b/apps/maple-agent/app/src/host.rs @@ -1415,6 +1415,8 @@ impl OwnerContext { | "remote_update_publish" | "remote_update_take" | "remote_update_complete" + | "remote_update_refresh_take" + | "remote_update_refresh_complete" ) { let account = request .params @@ -1455,6 +1457,26 @@ impl OwnerContext { .await?, ) .map_err(|_| "Could not encode desktop updater response".to_owned()), + "remote_update_refresh_take" => serde_json::to_value( + backend + .remote_updates + .take_refresh( + serde_json::from_value(request.params.clone()) + .map_err(|_| "Invalid desktop update check poll")?, + ) + .await?, + ) + .map_err(|_| "Could not encode desktop updater response".to_owned()), + "remote_update_refresh_complete" => { + backend + .remote_updates + .complete_refresh( + serde_json::from_value(request.params.clone()) + .map_err(|_| "Invalid desktop update check completion")?, + ) + .await?; + Ok(serde_json::Value::Null) + } _ => { backend .remote_updates diff --git a/apps/maple-agent/app/src/remote/control.rs b/apps/maple-agent/app/src/remote/control.rs index 26d974ac1..5ec5efb08 100644 --- a/apps/maple-agent/app/src/remote/control.rs +++ b/apps/maple-agent/app/src/remote/control.rs @@ -262,20 +262,39 @@ async fn serve_request( return serve_install(node, peer, &mut tx, request).await; } let id = request.id; - let result = if request.method == wire::Method::PeerInfo { + let result = if matches!( + request.method, + wire::Method::PeerInfo | wire::Method::RemoteUpdateRefresh + ) { #[derive(serde::Deserialize)] #[serde(deny_unknown_fields)] struct Empty {} if request.send_id.is_some() || serde_json::from_value::(request.params).is_err() { Err("Invalid remote information request".into()) } else { - let info = peer_information( + let update = if request.method == wire::Method::RemoteUpdateRefresh { + // This response is tied to a new check claimed by the current + // execution desktop, never to its cached status heartbeat. + node.backend + .remote_updates + .request_refresh(node.account.clone(), peer.to_string()) + .await + } else { node.backend .remote_updates .status(&node.account, &peer.to_string()) - .await, - )?; - serde_json::to_value(info).map_err(|_| "Remote information unavailable".into()) + .await + }; + if request.method == wire::Method::RemoteUpdateRefresh { + match update { + Ok(status) => serde_json::to_value(peer_information(Ok(status))?) + .map_err(|_| "Remote information unavailable".into()), + Err(error) => Err(error), + } + } else { + serde_json::to_value(peer_information(update)?) + .map_err(|_| "Remote information unavailable".into()) + } } } else if request.method == wire::Method::Subscribe { set_subscription(&subscription, &node.account, request.params) @@ -376,9 +395,30 @@ async fn serve_install( .await; } }; - command.request.validate()?; - let authority = RemoteInstallAuthority::capture(&node, peer, command.expected_grant)?; - authority.ensure_current().await?; + // These refusals precede Broker admission. Reply explicitly so the client + // does not confuse a known rejection with a lost installation receipt. + if command.request.validate().is_err() { + return install_refused(tx, id, "Invalid remote update request").await; + } + let authority = match RemoteInstallAuthority::capture(&node, peer, command.expected_grant) { + Ok(authority) => authority, + Err(_) => { + return install_refused( + tx, + id, + "Remote update authorization changed. Reopen this computer.", + ) + .await; + } + }; + if authority.ensure_current().await.is_err() { + return install_refused( + tx, + id, + "Remote update authorization changed. Reopen this computer.", + ) + .await; + } let mut pending = PendingInstallReply { broker: node.backend.remote_updates.clone(), authority, @@ -389,11 +429,22 @@ async fn serve_install( .broker .request_install(pending.authority.clone(), command.request) .await; - pending.authority.ensure_current().await?; let admitted = accepted.is_ok(); - let result = accepted.and_then(|value| { - serde_json::to_value(value).map_err(|_| "Remote update receipt unavailable".into()) - }); + if admitted { + pending.authority.ensure_current().await?; + } + let result = match accepted { + Ok(value) => { + serde_json::to_value(value).map_err(|_| "Remote update receipt unavailable".into()) + } + Err(error) => Err(if error.contains("in progress") || error.contains("busy") { + "The remote computer is busy".into() + } else if error.contains("ready") || error.contains("Open Maple") { + "The remote computer is not ready to install this update".into() + } else { + "The remote computer refused the update before accepting it".into() + }), + }; streams::reply_with_limit( tx, &Message::Response { id, result }, @@ -411,6 +462,22 @@ async fn serve_install( Ok(()) } +async fn install_refused( + tx: &mut SendStream, + id: u64, + message: &'static str, +) -> Result<(), String> { + streams::reply_with_limit( + tx, + &Message::Response { + id, + result: Err(message.to_owned()), + }, + wire::MAX_MANAGEMENT_REPLY, + ) + .await +} + fn set_subscription( subscription: &RwLock, account: &str, diff --git a/apps/maple-agent/app/src/remote/manager.rs b/apps/maple-agent/app/src/remote/manager.rs index c8beaa132..93555f896 100644 --- a/apps/maple-agent/app/src/remote/manager.rs +++ b/apps/maple-agent/app/src/remote/manager.rs @@ -24,6 +24,93 @@ use tokio::sync::{Mutex as AsyncMutex, Semaphore, broadcast}; use tokio_util::sync::CancellationToken; use uuid::Uuid; +const HOST_GRANT_RECOVERY_LIMIT: usize = 8; +const HOST_GRANT_PROBE_TIMEOUT: Duration = Duration::from_secs(3); +const HOST_GRANT_RECOVERY_TIMEOUT: Duration = Duration::from_secs(15); +const PAIRING_CONFIRMATION_REQUIRED: &str = "Pair this computer again before updating it remotely. Its saved pairing confirmation is unavailable."; + +fn encode_install_reply(reply: super::update::InstallReply) -> Result { + serde_json::to_value(reply).map_err(|_| "Could not display remote update receipt".into()) +} + +fn remote_refresh_failure(error: &str) -> String { + use crate::remote_updates::RefreshFailure; + [ + RefreshFailure::Busy, + RefreshFailure::Disabled, + RefreshFailure::Unavailable, + RefreshFailure::CheckFailed, + RefreshFailure::TimedOut, + ] + .into_iter() + .map(RefreshFailure::message) + .find(|message| *message == error) + .unwrap_or(RefreshFailure::CheckFailed.message()) + .to_owned() +} + +fn confirmed_host_grant_reply( + window: [u8; 16], + confirmed: Result<[u8; 16], wire::Refusal>, +) -> Result { + match confirmed { + Ok(grant) if grant == window => Ok(true), + Err(refusal) + if refusal.terminal && refusal.message == "No current grant matches this pairing" => + { + Ok(false) + } + Err(refusal) => Err(refusal.into_error()), + Ok(_) => Err("Invalid grant acknowledgement".into()), + } +} + +/// Attempt history is only a bounded candidate list, never authority. A +/// successful authenticated ConfirmGrant response supplies the missing proof. +async fn recover_confirmed_host_grant( + attempts: &[[u8; 16]], + probe_timeout: Duration, + total_timeout: Duration, + mut probe: P, +) -> Result<[u8; 16], String> +where + P: FnMut([u8; 16]) -> F, + F: std::future::Future>, +{ + tokio::time::timeout(total_timeout, async { + for window in attempts.iter().rev().take(HOST_GRANT_RECOVERY_LIMIT) { + match tokio::time::timeout(probe_timeout, probe(*window)).await { + Ok(Ok(true)) => return Ok(*window), + Ok(Ok(false)) => {} + Ok(Err(error)) => return Err(error), + Err(_) => return Err(PAIRING_CONFIRMATION_REQUIRED.into()), + } + } + Err(PAIRING_CONFIRMATION_REQUIRED.into()) + }) + .await + .map_err(|_| PAIRING_CONFIRMATION_REQUIRED.to_owned())? +} + +async fn retain_observed_peer_info( + mut info: Option, + known: bool, + deadline: tokio::time::Instant, + recovery: F, +) -> Option +where + F: std::future::Future>, +{ + if let Some(info) = &mut info { + let confirmed = + known || matches!(tokio::time::timeout_at(deadline, recovery).await, Ok(Ok(_))); + if !confirmed && info.update.blocked.is_none() { + info.update.blocked = Some(super::update::BlockedReason::PairingConfirmationRequired); + } + } + info +} + async fn query_peer_information( connect: impl std::future::Future>, probe: impl FnOnce(T) -> F, @@ -360,30 +447,80 @@ impl Manager { let encode = |value| { serde_json::to_value(value).map_err(|_| "Could not encode remote settings".to_string()) }; - if request.method == "remote_peer_info" { + if matches!( + request.method.as_str(), + "remote_peer_info" | "remote_refresh_update" + ) { + let refresh = request.method == "remote_refresh_update"; let (account, peer): (String, String) = serde_json::from_value(request.params) .map_err(|_| "Invalid remote update request")?; let _account_guard = self.backend.account_gate.read().await; self.backend.authorize_account(&account)?; - let info = query_peer_information( + let limit = Duration::from_secs(if refresh { 45 } else { 20 }); + let deadline = tokio::time::Instant::now() + limit; + let mut recovery = None; + let mut info = query_peer_information( self.connect(&peer), |link| { + recovery = Some((link.node.clone(), link.peer)); let account = &account; async move { if link.node.account != *account { return Err("The signed-in account changed".into()); } - Ok(super::streams::peer_info(&link.connection).await) + let observed = super::streams::peer_info(&link.connection).await; + let info = if refresh { + let observed = observed.ok_or("Could not read this computer’s update capabilities")?; + if !observed.supports_remote_refresh() { + return Err("This Maple version cannot check for updates remotely. Check for Updates directly on that computer.".into()); + } + let response = super::streams::call(&link.connection, wire::Command { + id: 3, + method: wire::Method::RemoteUpdateRefresh, + send_id: None, + params: serde_json::json!({}), + }).await.map_err(|_| "The remote update check could not be completed")?; + let protocol::Message::Response { id: 3, result } = response else { + return Err("The remote update check returned an invalid response".into()); + }; + let value = result.map_err(|error| remote_refresh_failure(&error))?; + let checked: super::update::PeerInfo = serde_json::from_value(value) + .map_err(|_| "The remote update check returned an invalid response")?; + checked.validate()?; + Some(checked) + } else { + observed + }; + Ok(info) } }, - Duration::from_secs(5), + limit, ) .await?; + // Retain the independently read status before attempting metadata + // recovery. Slow confirmation must never erase a known version or + // turn a completed signed-feed check into an unavailable response. + if let Some((node, peer)) = recovery.filter(|_| info.is_some()) { + let known = node.store.read().ok().is_some_and(|store| { + store + .peers() + .get(&peer.to_string()) + .is_some_and(|saved| saved.can_connect && saved.host_grant.is_some()) + }); + info = + retain_observed_peer_info(info, known, deadline, node.ensure_host_grant(peer)) + .await; + } self.backend.authorize_account(&account)?; + if refresh && info.is_none() { + return Err("The remote update check did not finish. Its previous status is not a fresh update check.".into()); + } return serde_json::to_value(info) .map_err(|_| "Could not display remote update status".into()); } if request.method == "remote_install_update" { + use super::update::{InstallRefusal as Refusal, InstallReply}; + let refused = |reason| Ok::<_, String>(InstallReply::Refused { reason }); let (account, peer, expected_grant, install): ( String, String, @@ -391,52 +528,96 @@ impl Manager { super::update::InstallRequest, ) = serde_json::from_value(request.params) .map_err(|_| "Invalid remote update request")?; - install.validate()?; + if install.validate().is_err() { + return encode_install_reply(InstallReply::Refused { + reason: Refusal::Unsupported, + }); + } let _account_guard = self.backend.account_gate.read().await; - self.backend.authorize_account(&account)?; - let response = tokio::time::timeout(Duration::from_secs(15), async { - let link = self.connect(&peer).await?; - self.backend.authorize_account(&account)?; - if link.node.account != account { return Err("The signed-in account changed".into()); } - let paired = link.node.store.read().map_err(|_| "Remote grant unavailable")?.peers() - .get(&link.peer.to_string()).is_some_and(|value| value.can_connect && value.grant == expected_grant); - if !paired { return Err("Remote update authorization changed. Reopen this computer.".into()); } - let info = super::streams::peer_info(&link.connection).await - .filter(super::update::PeerInfo::supports_remote_updates) - .ok_or("This computer does not support remote updates")?; + if self.backend.authorize_account(&account).is_err() { + return encode_install_reply(InstallReply::Refused { + reason: Refusal::AccountChanged, + }); + } + let submitted = AtomicBool::new(false); + let outcome = tokio::time::timeout(Duration::from_secs(15), async { + let link = match self.connect(&peer).await { + Ok(link) => link, + Err(error) if wire::is_protocol_incompatibility_refusal(&error) => return refused(Refusal::Unsupported), + Err(_) => return refused(Refusal::Unavailable), + }; + if self.backend.authorize_account(&account).is_err() || link.node.account != account { + return refused(Refusal::AccountChanged); + } + let paired = match link.node.store.read() { + Ok(store) => store.peers().get(&link.peer.to_string()).is_some_and(|value| value.can_connect && value.host_grant == Some(expected_grant)), + Err(_) => return refused(Refusal::Unavailable), + }; + if !paired { return refused(Refusal::AuthorizationChanged); } + let Some(info) = super::streams::peer_info(&link.connection).await else { + return refused(Refusal::Unavailable); + }; + if !info.supports_remote_updates() { return refused(Refusal::Unsupported); } if info.update.offered.as_ref() != Some(&install.target) { - return Err("The remote update changed. Refresh before trying again.".into()); + return refused(Refusal::UpdateChanged); } if info.update.state != super::update::UpdateState::Ready || info.update.blocked.is_some() { - return Err("The remote computer is not ready to install this update.".into()); + return refused(if info.update.blocked == Some(super::update::BlockedReason::Busy) { Refusal::Busy } else { Refusal::NotReady }); } - self.backend.authorize_account(&account)?; - link.node.ensure_valid().await?; - if !link.node.store.read().map_err(|_| "Remote grant unavailable")?.peers() - .get(&link.peer.to_string()).is_some_and(|value| value.can_connect && value.grant == expected_grant) { - return Err("Remote update authorization changed. Reopen this computer.".into()); + if self.backend.authorize_account(&account).is_err() { + return refused(Refusal::AccountChanged); + } + if link.node.ensure_valid().await.is_err() { return refused(Refusal::AuthorizationChanged); } + let paired = match link.node.store.read() { + Ok(store) => store.peers().get(&link.peer.to_string()).is_some_and(|value| value.can_connect && value.host_grant == Some(expected_grant)), + Err(_) => return refused(Refusal::Unavailable), + }; + if !paired { + return refused(Refusal::AuthorizationChanged); } + // From this point a lost/invalid reply can follow admission. + // Never turn that ambiguity into a definite refusal or retry. + let params = match serde_json::to_value(super::update::InstallCommand { + expected_grant, request: install.clone(), + }) { + Ok(params) => params, + Err(_) => return refused(Refusal::Unsupported), + }; + submitted.store(true, Ordering::Relaxed); let message = super::streams::call(&link.connection, wire::Command { id: 2, method: wire::Method::RemoteUpdateInstall, send_id: None, - params: serde_json::to_value(super::update::InstallCommand { - expected_grant, request: install.clone(), - }).map_err(|_| "Invalid remote update request")?, + params, }).await.map_err(|_| "Remote update outcome is uncertain. Refresh its status; do not submit another update.")?; let protocol::Message::Response { result, .. } = message else { return Err("Remote update outcome is uncertain. Refresh its status; do not submit another update.".into()); }; - let accepted: super::update::InstallAccepted = serde_json::from_value(result?) + let value = match result { + Ok(value) => value, + // The execution host replies Err only when admission failed. + // Do not copy its arbitrary private error into the client UI. + Err(_) => return refused(Refusal::Unavailable), + }; + let accepted: super::update::InstallAccepted = serde_json::from_value(value) .map_err(|_| "Remote update outcome is uncertain. Refresh its status; do not submit another update.")?; if accepted.operation_id != install.operation_id || accepted.target != install.target { return Err("Remote update outcome is uncertain. Refresh its status; do not submit another update.".into()); } - Ok::<_, String>(accepted) - }).await.map_err(|_| "Remote update outcome is uncertain. Refresh its status; do not submit another update.")??; - self.backend.authorize_account(&account)?; - return serde_json::to_value(response) - .map_err(|_| "Could not display remote update receipt".into()); + Ok::<_, String>(InstallReply::Accepted { receipt: accepted }) + }).await; + let response = match outcome { + Ok(response) => response?, + Err(_) if !submitted.load(Ordering::Relaxed) => InstallReply::Refused { reason: Refusal::Unavailable }, + Err(_) => return Err("Remote update outcome is uncertain. Refresh its status; do not submit another update.".into()), + }; + if self.backend.authorize_account(&account).is_err() { + return match response { + InstallReply::Refused { .. } => encode_install_reply(InstallReply::Refused { reason: Refusal::AccountChanged }), + InstallReply::Accepted { .. } => Err("Remote update outcome is uncertain. Refresh its status; do not submit another update.".into()), + }; + } + return encode_install_reply(response); } if request.method == "remote_pairing_window" { let current = self.node.lock().await.clone(); @@ -1005,6 +1186,76 @@ impl Node { .await .map_err(|_| "Pairing acknowledgement is still pending; reconnect to confirm it")? } + /// Repair only confirmed outgoing metadata in a released record. The caller + /// holds its local account gate and has already admitted a normal control + /// connection to this exact endpoint. No pairing proof or install is replayed. + async fn ensure_host_grant(&self, peer: EndpointId) -> Result<[u8; 16], String> { + let generation = self.pairing_generation.load(Ordering::SeqCst); + let expected = { + let store = self.store.read().map_err(|_| "Remote grant unavailable")?; + let saved = store + .peers() + .get(&peer.to_string()) + .filter(|saved| saved.can_connect) + .ok_or(PAIRING_CONFIRMATION_REQUIRED)?; + if let Some(grant) = saved.host_grant { + return Ok(grant); + } + store.host_grant_recovery(peer)? + }; + let window = recover_confirmed_host_grant( + &expected.attempts, + HOST_GRANT_PROBE_TIMEOUT, + HOST_GRANT_RECOVERY_TIMEOUT, + |window| self.confirm_saved_host_grant(peer, window), + ) + .await?; + self.ensure_valid().await?; + let mut store = write_policy(&self.store).await?; + self.backend.authorize_account(&self.account)?; + if self.pairing_generation.load(Ordering::SeqCst) != generation { + return Err(PAIRING_CONFIRMATION_REQUIRED.into()); + } + // The durable edit checks directional authority and the attempt snapshot: + // ordinary re-pairing does not necessarily change pairing_generation. + store.recover_host_grant(peer, &expected, window)?; + Ok(window) + } + + async fn confirm_saved_host_grant( + &self, + peer: EndpointId, + window: [u8; 16], + ) -> Result { + self.ensure_valid().await?; + let (address, _) = self.destination(peer).await?; + let connection = self + .endpoint + .connect(address, wire::ALPN) + .await + .map_err(|_| PAIRING_CONFIRMATION_REQUIRED)?; + if connection.remote_id() != peer { + return Err("Remote identity changed".into()); + } + let (mut tx, mut rx) = connection + .open_bi() + .await + .map_err(|_| PAIRING_CONFIRMATION_REQUIRED)?; + wire::write( + &mut tx, + &self.hello(wire::Mode::ConfirmGrant { window }), + wire::MAX_PREFACE, + ) + .await + .map_err(|error| error.to_string())?; + let confirmed: Result<[u8; 16], wire::Refusal> = wire::read(&mut rx, wire::MAX_PREFACE) + .await + .map_err(|error| error.to_string())?; + // Only the exact legacy negative acknowledgement permits another + // candidate. Context, protocol and identity failures stop recovery. + confirmed_host_grant_reply(window, confirmed) + } + pub(super) async fn accept_pairing( &self, peer: EndpointId, @@ -1089,6 +1340,7 @@ impl Node { Peer { name: "Paired controller".into(), grant: window, + host_grant: None, controls_this_computer: true, can_connect: false, pending_confirmation: None, @@ -1279,6 +1531,175 @@ fn validate_devices(devices: &[RemoteDevice]) -> Result<(), String> { mod tests { use super::*; + #[tokio::test(start_paused = true)] + async fn a_slow_legacy_confirmation_cannot_erase_an_observed_ready_release() { + use super::super::update::{ + BlockedReason, PeerInfo, ProtocolInfo, RemoteUpdateStatus, RunningApplication, + UpdateState, + }; + let observed = PeerInfo { + protocol: ProtocolInfo::current(), + running: RunningApplication::current(), + update: RemoteUpdateStatus { + state: UpdateState::Ready, + ..Default::default() + }, + }; + let returned = retain_observed_peer_info( + Some(observed.clone()), + false, + tokio::time::Instant::now() + Duration::from_secs(2), + std::future::pending::>(), + ) + .await + .unwrap(); + assert_eq!(returned.running, observed.running); + assert_eq!(returned.update.state, UpdateState::Ready); + assert_eq!( + returned.update.blocked, + Some(BlockedReason::PairingConfirmationRequired) + ); + + let mut protected = observed; + protected.update.blocked = Some(BlockedReason::PasswordProtected); + let returned = retain_observed_peer_info( + Some(protected.clone()), + false, + tokio::time::Instant::now(), + std::future::ready(Err(PAIRING_CONFIRMATION_REQUIRED.into())), + ) + .await; + assert_eq!(returned, Some(protected)); + } + + #[test] + fn remote_refresh_preserves_controlled_failure_categories_without_peer_text() { + use crate::remote_updates::RefreshFailure; + for failure in [ + RefreshFailure::Busy, + RefreshFailure::Disabled, + RefreshFailure::Unavailable, + RefreshFailure::CheckFailed, + RefreshFailure::TimedOut, + ] { + assert_eq!(remote_refresh_failure(failure.message()), failure.message()); + } + assert_eq!( + remote_refresh_failure("synthetic private remote path or account detail"), + RefreshFailure::CheckFailed.message(), + ); + } + + #[tokio::test] + async fn legacy_grant_recovery_needs_an_exact_confirmation_and_has_a_bounded_tail() { + let attempts: Vec<_> = (1..=12).map(|n| [n; 16]).collect(); + let mut probes = Vec::new(); + let recovered = recover_confirmed_host_grant( + &attempts, + Duration::from_secs(1), + Duration::from_secs(2), + |candidate| { + probes.push(candidate); + std::future::ready(confirmed_host_grant_reply( + candidate, + if candidate == [10; 16] { + Ok(candidate) + } else { + Err(wire::Refusal::terminal( + "No current grant matches this pairing", + )) + }, + )) + }, + ) + .await + .unwrap(); + assert_eq!(recovered, [10; 16]); + assert_eq!(probes, [[12; 16], [11; 16], [10; 16]]); + + probes.clear(); + let missing = recover_confirmed_host_grant( + &attempts, + Duration::from_secs(1), + Duration::from_secs(2), + |candidate| { + probes.push(candidate); + std::future::ready(Ok(candidate == [1; 16])) + }, + ) + .await; + assert_eq!(missing.unwrap_err(), PAIRING_CONFIRMATION_REQUIRED); + assert_eq!(probes.len(), HOST_GRANT_RECOVERY_LIMIT); + assert_eq!(probes.last(), Some(&[5; 16])); + } + + #[tokio::test] + async fn malformed_or_context_refused_confirmation_cannot_fall_through_to_another_attempt() { + for response in [ + Ok([9; 16]), + Err(wire::Refusal::terminal( + "Remote account or deployment mismatch", + )), + Err(wire::Refusal::terminal( + wire::Error::Incompatible.to_string(), + )), + ] { + let mut response = Some(response); + let mut calls = 0; + let result = recover_confirmed_host_grant( + &[[1; 16], [2; 16]], + Duration::from_secs(1), + Duration::from_secs(2), + |candidate| { + calls += 1; + std::future::ready(confirmed_host_grant_reply( + candidate, + response.take().unwrap(), + )) + }, + ) + .await; + assert!(result.is_err()); + assert_eq!(calls, 1); + } + } + + #[tokio::test(start_paused = true)] + async fn legacy_grant_recovery_times_out_without_trying_another_candidate() { + let started = tokio::time::Instant::now(); + let mut calls = 0; + let result = recover_confirmed_host_grant( + &[[1; 16], [2; 16]], + Duration::from_secs(3), + Duration::from_secs(15), + |_| { + calls += 1; + std::future::pending::>() + }, + ) + .await; + assert_eq!(result.unwrap_err(), PAIRING_CONFIRMATION_REQUIRED); + assert_eq!(calls, 1); + assert_eq!(started.elapsed(), Duration::from_secs(3)); + } + + #[tokio::test(start_paused = true)] + async fn legacy_grant_recovery_total_budget_bounds_a_slow_negative_history() { + let started = tokio::time::Instant::now(); + let result = recover_confirmed_host_grant( + &[[1; 16]; 12], + Duration::from_secs(3), + Duration::from_secs(5), + |_| async { + tokio::time::sleep(Duration::from_secs(2)).await; + Ok(false) + }, + ) + .await; + assert_eq!(result.unwrap_err(), PAIRING_CONFIRMATION_REQUIRED); + assert_eq!(started.elapsed(), Duration::from_secs(5)); + } + fn policy_fixture() -> (tempfile::TempDir, Arc>) { let root = tempfile::tempdir().unwrap(); let database = Arc::new( diff --git a/apps/maple-agent/app/src/remote/mod.rs b/apps/maple-agent/app/src/remote/mod.rs index 85baf82e9..7682bea86 100644 --- a/apps/maple-agent/app/src/remote/mod.rs +++ b/apps/maple-agent/app/src/remote/mod.rs @@ -1,6 +1,8 @@ //! Desktop remote access is owned by the profile's local daemon. mod network; mod pairing; +#[cfg(test)] +mod pairing_lifecycle_tests; mod store; pub(crate) mod streams; pub(crate) mod update; diff --git a/apps/maple-agent/app/src/remote/pairing_lifecycle_tests.rs b/apps/maple-agent/app/src/remote/pairing_lifecycle_tests.rs new file mode 100644 index 000000000..f11fa172c --- /dev/null +++ b/apps/maple-agent/app/src/remote/pairing_lifecycle_tests.rs @@ -0,0 +1,321 @@ +//! Compositional pairing regression: real CPace and ConfirmGrant frames over +//! loopback Iroh, two independent durable stores, restart, then the production +//! install-grant capture policy. Enrollment/session admission is a fixture +//! boundary; this is not a signed-payload installation or production-login test. + +use super::{ + network, pairing, + store::{Peer, Store}, + update::capture_install_grant, + wire, +}; +use iroh::{Endpoint, EndpointId, RelayMode, endpoint::Connection}; +use maple_agent::store::{Store as Database, StorePaths}; +use std::{sync::Arc, time::Duration}; + +const ACCOUNT: &str = "pairing-lifecycle-fixture"; +const API: &str = "http://127.0.0.1:1"; +const PROJECT: &str = "pairing-lifecycle-project"; +const CHANNEL: &str = "unpackaged"; + +struct Computer { + root: tempfile::TempDir, + database: Arc, + policy: Store, + endpoint: Endpoint, +} + +impl Computer { + async fn new() -> Self { + let root = tempfile::tempdir().unwrap(); + let database = Arc::new(Database::open(StorePaths::new(root.path())).unwrap()); + database.upsert_account(ACCOUNT, 1).unwrap(); + let mut policy = Store::open(database.clone(), API, PROJECT, CHANNEL, ACCOUNT).unwrap(); + policy.set_enabled(true).unwrap(); + let endpoint = network::bind(policy.key(), RelayMode::Disabled, false, true) + .await + .unwrap(); + Self { + root, + database, + policy, + endpoint, + } + } + + async fn reopen(self) -> Self { + let Self { + root, + database, + policy, + endpoint, + } = self; + let identity = endpoint.id(); + endpoint.close().await; + drop(policy); + database.close().unwrap(); + drop(database); + let database = Arc::new(Database::open(StorePaths::new(root.path())).unwrap()); + assert!(Store::exists(&database, API, PROJECT, CHANNEL, ACCOUNT).unwrap()); + let policy = Store::open(database.clone(), API, PROJECT, CHANNEL, ACCOUNT).unwrap(); + let endpoint = network::bind(policy.key(), RelayMode::Disabled, false, true) + .await + .unwrap(); + assert_eq!( + endpoint.id(), + identity, + "the endpoint key must survive restart" + ); + Self { + root, + database, + policy, + endpoint, + } + } + + async fn close(self) { + self.endpoint.close().await; + drop(self.policy); + self.database.close().unwrap(); + } +} + +fn hello(mode: wire::Mode) -> wire::Hello { + wire::Hello { + version: wire::VERSION, + minimum_version: wire::VERSION, + account: ACCOUNT.into(), + project: PROJECT.into(), + deployment: "local".into(), + channel: CHANNEL.into(), + mode, + } +} + +fn check_context(hello: &wire::Hello) { + assert_eq!(hello.account, ACCOUNT); + assert_eq!(hello.project, PROJECT); + assert_eq!(hello.deployment, "local"); + assert_eq!(hello.channel, CHANNEL); +} + +async fn connections(a: &Endpoint, b: &Endpoint) -> (Connection, Connection) { + let (client, server) = tokio::join!( + async { a.connect(b.addr(), wire::ALPN).await.unwrap() }, + async { b.accept().await.unwrap().await.unwrap() }, + ); + assert_eq!(client.remote_id(), b.id()); + assert_eq!(server.remote_id(), a.id()); + (client, server) +} + +/// Only enrollment admission is supplied by the fixture. Both CPace proofs and +/// all persistence transitions use the production implementations. +async fn pair(controller: &mut Computer, execution: &mut Computer, lose_ack: bool) -> [u8; 16] { + let window = *uuid::Uuid::new_v4().as_bytes(); + let controller_id = controller.endpoint.id(); + let execution_id = execution.endpoint.id(); + let controller_key = *controller_id.as_bytes(); + let execution_key = *execution_id.as_bytes(); + let context = pairing::Context { + controller: &controller_key, + execution: &execution_key, + account: ACCOUNT, + project: PROJECT, + deployment: "local", + channel: CHANNEL, + window: &window, + }; + let proof = + pairing::Controller::new(pairing::Code::parse("ABCDEFG2").unwrap(), &context).unwrap(); + let (client, server) = connections(&controller.endpoint, &execution.endpoint).await; + let outgoing = async { + let (mut tx, mut rx) = client.open_bi().await.unwrap(); + wire::write( + &mut tx, + &hello(wire::Mode::Pair { + window, + admission: "fixture-only-enrollment-admission".into(), + }), + wire::MAX_PREFACE, + ) + .await + .unwrap(); + controller + .policy + .reserve_attempt(execution_id, window) + .unwrap(); + wire::write(&mut tx, &proof.message(), wire::MAX_PREFACE) + .await + .unwrap(); + let answer: Result<(Vec, Vec), wire::Refusal> = + wire::read(&mut rx, wire::MAX_PREFACE).await.unwrap(); + let (message, host_tag) = answer.unwrap(); + let tag = proof.confirm(message, &host_tag).unwrap(); + controller + .policy + .remember_host_proof(execution_id, window, "Execution computer".into()) + .unwrap(); + wire::write(&mut tx, &tag.as_slice(), wire::MAX_PREFACE) + .await + .unwrap(); + let acknowledgement: Result, wire::Error> = + wire::read(&mut rx, wire::MAX_PREFACE).await; + if lose_ack { + assert!(acknowledgement.is_err()); + } else { + let grant = acknowledgement.unwrap().unwrap(); + assert_eq!(grant, window); + controller.policy.confirm_host(execution_id, grant).unwrap(); + } + }; + let incoming = async { + let (mut tx, mut rx) = server.accept_bi().await.unwrap(); + let request = wire::read_hello(&mut rx).await.unwrap(); + check_context(&request); + assert!( + matches!(request.mode, wire::Mode::Pair { window: received, .. } if received == window) + ); + let message = wire::read(&mut rx, wire::MAX_PREFACE).await.unwrap(); + let proof = + pairing::Execution::new(pairing::Code::parse("ABCDEFG2").unwrap(), &context, message) + .unwrap(); + wire::write( + &mut tx, + &Ok::<_, wire::Refusal>((proof.message(), proof.confirmation().as_slice())), + wire::MAX_PREFACE, + ) + .await + .unwrap(); + let tag: Vec = wire::read(&mut rx, wire::MAX_PREFACE).await.unwrap(); + proof.verify(&tag).unwrap(); + execution + .policy + .grant( + controller_id, + Peer { + name: "Paired controller".into(), + grant: window, + host_grant: None, + controls_this_computer: true, + can_connect: false, + pending_confirmation: None, + }, + ) + .unwrap(); + if lose_ack { + // The host committed its grant; only the acknowledgement is lost. + tx.reset(91u32.into()).unwrap(); + } else { + wire::write(&mut tx, &Ok::<_, wire::Refusal>(window), wire::MAX_PREFACE) + .await + .unwrap(); + tx.finish().unwrap(); + } + }; + tokio::join!(outgoing, incoming); + window +} + +async fn confirm_after_restart(controller: &mut Computer, execution: &Computer) { + let controller_id = controller.endpoint.id(); + let execution_id = execution.endpoint.id(); + let pending = &controller.policy.peers()[&execution_id.to_string()]; + assert!(!pending.can_connect); + assert!(pending.host_grant.is_none()); + let window = pending + .pending_confirmation + .expect("durable proof intent survives lost acknowledgement"); + let (client, server) = connections(&controller.endpoint, &execution.endpoint).await; + let outgoing = async { + let (mut tx, mut rx) = client.open_bi().await.unwrap(); + wire::write( + &mut tx, + &hello(wire::Mode::ConfirmGrant { window }), + wire::MAX_PREFACE, + ) + .await + .unwrap(); + let confirmed: Result<[u8; 16], wire::Refusal> = + wire::read(&mut rx, wire::MAX_PREFACE).await.unwrap(); + let confirmed = confirmed.unwrap(); + assert_eq!(confirmed, window); + controller + .policy + .confirm_host(execution_id, confirmed) + .unwrap(); + }; + let incoming = async { + let (mut tx, mut rx) = server.accept_bi().await.unwrap(); + let request = wire::read_hello(&mut rx).await.unwrap(); + check_context(&request); + let wire::Mode::ConfirmGrant { window } = request.mode else { + panic!("expected confirmation") + }; + let allowed = execution + .policy + .peers() + .get(&controller_id.to_string()) + .is_some_and(|peer| peer.controls_this_computer && peer.grant == window); + let confirmed = if allowed { + Ok(window) + } else { + Err(wire::Refusal::terminal( + "No current grant matches this pairing", + )) + }; + wire::write(&mut tx, &confirmed, wire::MAX_PREFACE) + .await + .unwrap(); + tx.finish().unwrap(); + }; + tokio::join!(outgoing, incoming); +} + +fn outgoing_grant(controller: &Computer, execution: EndpointId) -> [u8; 16] { + let peer = &controller.policy.peers()[&execution.to_string()]; + assert!(peer.can_connect); + assert!(peer.pending_confirmation.is_none()); + peer.host_grant + .expect("pairing must durably retain the confirmed outgoing authority") +} + +#[tokio::test] +async fn cpace_confirm_restart_and_mutual_pairing_feed_the_real_install_grant_policy() { + let exercise = async { + let mut a = Computer::new().await; + let mut b = Computer::new().await; + let b_window = pair(&mut a, &mut b, true).await; + a = a.reopen().await; + b = b.reopen().await; + confirm_after_restart(&mut a, &b).await; + a = a.reopen().await; + b = b.reopen().await; + let a_to_b = outgoing_grant(&a, b.endpoint.id()); + assert_eq!(a_to_b, b_window); + assert!(capture_install_grant(&b.policy, a.endpoint.id(), a_to_b).is_ok()); + assert!(capture_install_grant(&b.policy, a.endpoint.id(), [0; 16]).is_err()); + + // Reversing the roles must not overwrite the already confirmed A→B + // authority with B→A's different incoming pairing generation. + let a_window = pair(&mut b, &mut a, false).await; + assert_ne!(a_window, b_window); + a = a.reopen().await; + b = b.reopen().await; + let a_to_b = outgoing_grant(&a, b.endpoint.id()); + let b_to_a = outgoing_grant(&b, a.endpoint.id()); + assert_eq!(a_to_b, b_window); + assert_eq!(b_to_a, a_window); + assert!(capture_install_grant(&b.policy, a.endpoint.id(), a_to_b).is_ok()); + assert!(capture_install_grant(&a.policy, b.endpoint.id(), b_to_a).is_ok()); + assert!(capture_install_grant(&b.policy, a.endpoint.id(), b_to_a).is_err()); + assert!(capture_install_grant(&a.policy, b.endpoint.id(), a_to_b).is_err()); + b.policy.revoke(a.endpoint.id()).unwrap(); + assert!(capture_install_grant(&b.policy, a.endpoint.id(), a_to_b).is_err()); + tokio::join!(a.close(), b.close()); + }; + tokio::time::timeout(Duration::from_secs(30), exercise) + .await + .expect("pairing lifecycle timed out"); +} diff --git a/apps/maple-agent/app/src/remote/store.rs b/apps/maple-agent/app/src/remote/store.rs index 03a869b89..6aafc9662 100644 --- a/apps/maple-agent/app/src/remote/store.rs +++ b/apps/maple-agent/app/src/remote/store.rs @@ -58,6 +58,13 @@ pub struct Store { record: RemoteIdentity, } +/// A durable legacy record, captured before asking the authenticated host to +/// confirm its outgoing grant. It is compared again inside the storage edit. +pub(super) struct HostGrantRecovery { + pub peer: Peer, + pub attempts: Vec<[u8; 16]>, +} + impl Store { fn scope( database: &Database, @@ -193,6 +200,41 @@ impl Store { window, )) } + pub(super) fn host_grant_recovery(&self, key: EndpointId) -> Result { + let key = key.to_string(); + let peer = self + .record + .peers + .get(&key) + .filter(|peer| { + peer.can_connect && peer.host_grant.is_none() && peer.pending_confirmation.is_none() + }) + .ok_or("Pair this computer again before updating it remotely")?; + Ok(HostGrantRecovery { + peer: peer.clone(), + attempts: self + .record + .attempted_windows + .get(&key) + .cloned() + .unwrap_or_default(), + }) + } + pub(super) fn recover_host_grant( + &mut self, + key: EndpointId, + expected: &HostGrantRecovery, + window: [u8; 16], + ) -> Result<(), String> { + self.refresh(self.database.recover_remote_host_grant( + self.account, + &self.namespace, + &key.to_string(), + &expected.peer, + &expected.attempts, + window, + )) + } /// The transaction commits before any code-derived proof is sent. pub fn reserve_attempt(&mut self, key: EndpointId, window: [u8; 16]) -> Result<(), String> { self.refresh(self.database.reserve_remote_pairing_attempt( @@ -307,11 +349,16 @@ mod tests { let mut expected = Peer { name: "Maple computer".into(), grant: [1; 16], + host_grant: None, controls_this_computer: true, can_connect: false, - pending_confirmation: Some([2; 16]), + pending_confirmation: None, }; store.grant(peer, expected.clone()).unwrap(); + store + .remember_host_proof(peer, [2; 16], "Maple computer".into()) + .unwrap(); + expected.pending_confirmation = Some([2; 16]); store .refresh_peer_names([(key.as_str(), "Office Mac"), (&unknown, "Unknown")]) .unwrap(); @@ -345,6 +392,7 @@ mod tests { Peer { name: "Test".into(), grant: [1; 16], + host_grant: None, controls_this_computer: true, can_connect: false, pending_confirmation: None, @@ -381,16 +429,22 @@ mod tests { .remember_host_proof(peer, [1; 16], "B".into()) .unwrap(); assert!(!store.peers()[&peer.to_string()].can_connect); + assert_eq!(store.peers()[&peer.to_string()].host_grant, None); drop(store); let mut store = open(); assert!(store.reserve_attempt(peer, [1; 16]).is_err()); + assert_eq!(store.peers()[&peer.to_string()].host_grant, None); store.confirm_host(peer, [1; 16]).unwrap(); + drop(store); + let mut store = open(); + assert_eq!(store.peers()[&peer.to_string()].host_grant, Some([1; 16])); store .grant( peer, Peer { name: "B".into(), grant: [2; 16], + host_grant: None, controls_this_computer: true, can_connect: false, pending_confirmation: None, @@ -399,6 +453,7 @@ mod tests { .unwrap(); assert!(store.permits(peer)); assert!(store.peers()[&peer.to_string()].can_connect); + assert_eq!(store.peers()[&peer.to_string()].host_grant, Some([1; 16])); store .remember_host_proof(peer, [3; 16], "B".into()) .unwrap(); diff --git a/apps/maple-agent/app/src/remote/streams.rs b/apps/maple-agent/app/src/remote/streams.rs index 330388e30..fb184a464 100644 --- a/apps/maple-agent/app/src/remote/streams.rs +++ b/apps/maple-agent/app/src/remote/streams.rs @@ -24,6 +24,9 @@ const STREAM_RECEIVE_WINDOW: u32 = 512 * 1024; const CONNECTION_RECEIVE_WINDOW: u32 = 32 * 1024 * 1024; const _: () = assert!(CONNECTION_RECEIVE_WINDOW > MAX_BIDI_STREAMS * STREAM_RECEIVE_WINDOW); pub const IO_TIMEOUT: Duration = Duration::from_secs(10); +// The execution broker may wait 40 seconds for its desktop check. Leave response +// margin inside the manager's overall refresh deadline of 45 seconds. +const REFRESH_REPLY_TIMEOUT: Duration = Duration::from_secs(42); const DISCOVERY_TIMEOUT: Duration = Duration::from_secs(5); const FOLDERS_UNCONFIRMED: &str = "Could not confirm folder browsing support. Try again, or update Maple on the remote computer."; @@ -222,6 +225,7 @@ fn body_limit(method: Method) -> usize { method, Method::PeerInfo | Method::RemoteUpdateInstall + | Method::RemoteUpdateRefresh | Method::UploadAttachment | Method::AttachmentUploadStatus | Method::CancelAttachmentUpload @@ -362,13 +366,12 @@ pub async fn call(connection: &Connection, command: Command) -> Result message, Ok(message @ Message::Closed { .. }) => message, @@ -389,6 +392,28 @@ pub async fn call(connection: &Connection, command: Command) -> Result( + method: Method, + budget: Duration, + response: F, +) -> Result +where + F: std::future::Future, +{ + if method == Method::RemoteUpdateRefresh || method.urgent() { + tokio::time::timeout(budget, response).await.map_err(|_| { + if method == Method::RemoteUpdateRefresh { + "Remote update check reply timed out" + } else { + "Remote control reply timed out" + } + .to_owned() + }) + } else { + Ok(response.await) + } +} + pub(super) async fn finish_reply(rx: &mut RecvStream, deadline: Duration) -> Result<(), String> { let mut trailing = [0u8; 1]; let count = tokio::time::timeout(deadline, tokio::io::AsyncReadExt::read(rx, &mut trailing)) diff --git a/apps/maple-agent/app/src/remote/streams/tests.rs b/apps/maple-agent/app/src/remote/streams/tests.rs index 2947867ca..72203e5e3 100644 --- a/apps/maple-agent/app/src/remote/streams/tests.rs +++ b/apps/maple-agent/app/src/remote/streams/tests.rs @@ -118,9 +118,118 @@ async fn optional_probe_has_an_overall_deadline_even_before_first_reply_byte() { b.close().await; } +#[tokio::test] +async fn refresh_reply_has_a_bounded_check_budget_without_relaxing_urgent_controls() { + // Exercise the same response-await path as call() over real loopback QUIC. + // Scale only the deadline, so CI need not stall for 42 wall-clock seconds. + assert_eq!(IO_TIMEOUT, Duration::from_secs(10)); + assert_eq!(REFRESH_REPLY_TIMEOUT, Duration::from_secs(42)); + let ordinary = Duration::from_millis(100); + for (method, budget, success) in [ + (Method::RemoteUpdateRefresh, Duration::from_secs(5), true), + (Method::CancelRun, ordinary, false), + (Method::RemoteUpdateRefresh, ordinary, false), + ] { + let (a, b, client, server) = endpoints().await; + let release = Arc::new(tokio::sync::Notify::new()); + let (received, ready) = tokio::sync::oneshot::channel(); + let receiver = tokio::spawn({ + let release = release.clone(); + async move { + let capacity = Capacity::new(); + let (header, mut tx, mut rx, _slot) = receive(&server, &capacity).await; + assert_eq!(header.method, method); + header.body(&mut rx).await.unwrap(); + received.send(()).unwrap(); + release.notified().await; + reply_with_limit( + &mut tx, + &Message::Response { + id: 1, + result: Ok(json!("fresh check completed")), + }, + method.reply_limit(), + ) + .await + } + }); + let exercise = async { + let (mut tx, mut rx) = client.open_bi().await.unwrap(); + wire::write( + &mut tx, + &Header { + id: 1, + method, + send_id: None, + }, + wire::MAX_PREFACE, + ) + .await + .unwrap(); + wire::write(&mut tx, &json!({}), body_limit(method)) + .await + .unwrap(); + tx.finish().unwrap(); + ready.await.unwrap(); + let delayed_reply = success.then(|| { + tokio::spawn({ + let release = release.clone(); + async move { + tokio::time::sleep(ordinary * 3).await; + release.notify_one(); + } + }) + }); + let response = await_response( + method, + budget, + wire::read_progress::( + &mut rx, + method.reply_limit(), + TRANSFER_TIMEOUT, + true, + ), + ) + .await; + if success { + assert!( + matches!(response.unwrap().unwrap(), Message::Response { id:1, result:Ok(value) } if value == json!("fresh check completed")) + ); + finish_reply(&mut rx, IO_TIMEOUT).await.unwrap(); + delayed_reply.unwrap().await.unwrap(); + receiver.await.unwrap().unwrap(); + } else { + let Err(error) = response else { + panic!("the withheld reply must time out"); + }; + assert_eq!( + error, + if method == Method::RemoteUpdateRefresh { + "Remote update check reply timed out" + } else { + "Remote control reply timed out" + } + ); + // The failed caller rejects its stream. Release the fixture + // afterward; a delivery error to that receiver is expected. + rx.stop(17u32.into()).unwrap(); + release.notify_one(); + let _ = receiver.await.unwrap(); + } + }; + let result = tokio::time::timeout(Duration::from_secs(15), exercise).await; + tokio::join!(a.close(), b.close()); + result.expect("bounded loopback reply test stalled"); + } +} + #[test] fn management_messages_have_small_independent_limits() { - for method in [Method::PeerInfo, Method::RemoteUpdateInstall] { + for method in [ + Method::PeerInfo, + Method::RemoteUpdateInstall, + Method::RemoteUpdateRefresh, + ] { assert_eq!(body_limit(method), 4096); assert_eq!(method.reply_limit(), 16 * 1024); } diff --git a/apps/maple-agent/app/src/remote/update.rs b/apps/maple-agent/app/src/remote/update.rs index ea86592b2..5d920f383 100644 --- a/apps/maple-agent/app/src/remote/update.rs +++ b/apps/maple-agent/app/src/remote/update.rs @@ -6,9 +6,10 @@ use serde::{Deserialize, Deserializer, Serialize}; use std::sync::{Arc, Weak}; use uuid::Uuid; -pub const PROTOCOL_VERSION: &str = "7.0.0"; +pub const PROTOCOL_VERSION: &str = "7.1.0"; pub const PEER_INFO_CAPABILITY: &str = "peer_info_v1"; pub const INSTALL_CAPABILITY: &str = "remote_update_install_v1"; +pub const REFRESH_CAPABILITY: &str = "remote_update_refresh_v1"; pub const PROJECT_FOLDERS_CAPABILITY: &str = "project_folders_v1"; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -25,6 +26,7 @@ impl ProtocolInfo { capabilities: vec![ PEER_INFO_CAPABILITY.into(), INSTALL_CAPABILITY.into(), + REFRESH_CAPABILITY.into(), PROJECT_FOLDERS_CAPABILITY.into(), ], } @@ -154,6 +156,7 @@ pub enum BlockedReason { Busy, DesktopUnavailable, CoordinatorUnavailable, + PairingConfirmationRequired, #[serde(other)] Unknown, } @@ -175,6 +178,9 @@ impl BlockedReason { Self::CoordinatorUnavailable => { "Remote update controls need attention on this computer. Update it locally." } + Self::PairingConfirmationRequired => { + "Pair this computer again before updating it remotely. Its saved pairing confirmation is unavailable." + } Self::Unknown => "The remote computer cannot install this update remotely.", } } @@ -226,6 +232,10 @@ impl PeerInfo { self.protocol.supports(INSTALL_CAPABILITY) } + pub fn supports_remote_refresh(&self) -> bool { + self.protocol.supports(REFRESH_CAPABILITY) + } + pub fn supports_project_folders(&self) -> bool { self.protocol.supports(PROJECT_FOLDERS_CAPABILITY) } @@ -335,6 +345,54 @@ pub struct InstallAccepted { pub target: ReleaseIdentity, } +/// Local controller result. The remote wire receipt remains InstallAccepted; +/// definite refusal must not be presented as an unconfirmed installation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "result", rename_all = "snake_case")] +pub enum InstallReply { + Accepted { receipt: InstallAccepted }, + Refused { reason: InstallRefusal }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum InstallRefusal { + AuthorizationChanged, + UpdateChanged, + NotReady, + Unsupported, + AccountChanged, + Busy, + Unavailable, +} + +impl InstallRefusal { + #[cfg(feature = "desktop")] + pub fn message(self) -> &'static str { + match self { + Self::AuthorizationChanged => { + "The pairing confirmation changed. Refresh this computer’s status, or pair it again before updating it." + } + Self::UpdateChanged => { + "The remote update changed. Refresh this computer’s status before continuing." + } + Self::NotReady => { + "The remote computer is not ready to install this update. Refresh its status before continuing." + } + Self::Unsupported => { + "This computer does not support this remote update. Update Maple directly on it." + } + Self::AccountChanged => { + "The signed-in account changed. Reopen Remote settings before continuing." + } + Self::Busy => "The remote computer is busy. Refresh its status after it finishes.", + Self::Unavailable => { + "The update request was not accepted. Refresh this computer’s status or check it directly." + } + } + } +} + #[derive(Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub(super) struct InstallCommand { @@ -350,26 +408,36 @@ pub(crate) struct RemoteInstallAuthority { grant: [u8; 16], } +/// The incoming, durable pairing policy used at capture and final consumption. +/// Authentication/session and lease checks remain with their existing owners. +pub(super) fn capture_install_grant( + store: &super::store::Store, + peer: EndpointId, + expected_grant: [u8; 16], +) -> Result<(), String> { + if !store.permits(peer) + || !store + .peers() + .get(&peer.to_string()) + .is_some_and(|value| value.controls_this_computer && value.grant == expected_grant) + { + return Err("Remote update authorization changed. Reopen this computer.".into()); + } + Ok(()) +} + impl RemoteInstallAuthority { pub(super) fn capture( node: &Arc, peer: EndpointId, expected_grant: [u8; 16], ) -> Result { - let peer = peer.to_string(); let store = node.store.read().map_err(|_| "Remote grant unavailable")?; - if !store.permits(peer.parse().map_err(|_| "Remote grant unavailable")?) - || !store - .peers() - .get(&peer) - .is_some_and(|value| value.controls_this_computer && value.grant == expected_grant) - { - return Err("Remote update authorization changed. Reopen this computer.".into()); - } + capture_install_grant(&store, peer, expected_grant)?; Ok(Self { node: Arc::downgrade(node), account: node.account.clone(), - peer, + peer: peer.to_string(), grant: expected_grant, }) } @@ -404,14 +472,10 @@ impl RemoteInstallAuthority { let store = node.store.read().map_err(|_| "Remote grant unavailable")?; if node.stopped.is_cancelled() || node.backend.saved_user_id().as_deref() != Some(&self.account) - || !store.permits(peer) - || !store - .peers() - .get(&self.peer) - .is_some_and(|value| value.grant == self.grant) { return Err("Remote update authorization changed. Reopen this computer.".into()); } + capture_install_grant(&store, peer, self.grant)?; effect() } } @@ -445,6 +509,53 @@ mod tests { } } + #[test] + fn a_cached_peer_status_is_not_remote_refresh_support() { + let mut info = current_peer_for_compatibility(); + assert!(info.supports_remote_refresh()); + info.protocol.version = "7.0.0".into(); + info.protocol + .capabilities + .retain(|value| value != REFRESH_CAPABILITY); + assert!(info.supports_remote_updates()); + assert!(!info.supports_remote_refresh()); + info.protocol.capabilities.push(REFRESH_CAPABILITY.into()); + assert!(info.supports_remote_refresh()); + info.protocol.version = "6.1.0".into(); + assert!(!info.supports_remote_refresh()); + } + + #[test] + fn a_definite_local_refusal_cannot_decode_as_an_accepted_wire_receipt() { + let refused = InstallReply::Refused { + reason: InstallRefusal::AuthorizationChanged, + }; + let encoded = serde_json::to_value(&refused).unwrap(); + assert!(serde_json::from_value::(encoded.clone()).is_err()); + assert_eq!( + serde_json::from_value::(encoded).unwrap(), + refused + ); + let accepted = InstallAccepted { + operation_id: "11111111-1111-4111-8111-111111111111".into(), + target: target(), + }; + assert!( + serde_json::from_value::(serde_json::to_value(&accepted).unwrap()) + .is_err() + ); + assert_eq!( + serde_json::from_value::( + serde_json::to_value(InstallReply::Accepted { + receipt: accepted.clone(), + }) + .unwrap() + ) + .unwrap(), + InstallReply::Accepted { receipt: accepted }, + ); + } + #[test] fn install_requests_reject_unknown_nested_fields_and_noncanonical_identities() { let value = @@ -491,6 +602,7 @@ mod tests { u64::from(super::super::wire::VERSION) ); assert!(current.supports(INSTALL_CAPABILITY)); + assert!(current.supports(REFRESH_CAPABILITY)); assert!(current.supports(PROJECT_FOLDERS_CAPABILITY)); } @@ -747,7 +859,7 @@ mod tests { ); let legacy: released_v6_1::PeerInfo = serde_json::from_slice(&bytes).unwrap(); assert!(legacy.valid()); - assert_eq!(legacy.protocol.version, "7.0.0"); + assert_eq!(legacy.protocol.version, PROTOCOL_VERSION); assert!( !legacy.supports_remote_updates(), "no cross-major management bridge" diff --git a/apps/maple-agent/app/src/remote/uploads/tests.rs b/apps/maple-agent/app/src/remote/uploads/tests.rs index ee1837786..7585cd993 100644 --- a/apps/maple-agent/app/src/remote/uploads/tests.rs +++ b/apps/maple-agent/app/src/remote/uploads/tests.rs @@ -483,6 +483,7 @@ fn revocation_disable_and_repair_do_not_publish_an_upload_admitted_under_an_old_ grant: [1; 16], controls_this_computer: true, can_connect: false, + host_grant: None, pending_confirmation: None, }; policy.set_enabled(true).unwrap(); @@ -604,6 +605,7 @@ fn receiver_fixture( grant: [1; 16], controls_this_computer: true, can_connect: false, + host_grant: None, pending_confirmation: None, }, ) diff --git a/apps/maple-agent/app/src/remote/wire.rs b/apps/maple-agent/app/src/remote/wire.rs index 8ed2c6f16..760ce6750 100644 --- a/apps/maple-agent/app/src/remote/wire.rs +++ b/apps/maple-agent/app/src/remote/wire.rs @@ -111,6 +111,7 @@ impl Hello { pub enum Method { PeerInfo, RemoteUpdateInstall, + RemoteUpdateRefresh, NavigationBootstrap, ListTasks, ListProjects, @@ -196,6 +197,7 @@ impl Method { self, Self::PeerInfo | Self::RemoteUpdateInstall + | Self::RemoteUpdateRefresh | Self::UploadAttachment | Self::AttachmentUploadStatus | Self::CancelAttachmentUpload @@ -271,7 +273,6 @@ const INCOMPATIBLE_PROTOCOL_MESSAGE: &str = /// The GUI receives terminal refusals through the host's string error boundary. /// Only the explicit protocol refusal establishes that an update is required; /// malformed frames, network failures and revoked authority do not. -#[cfg(any(feature = "desktop", test))] pub(crate) fn is_protocol_incompatibility_refusal(message: &str) -> bool { message.strip_prefix("Remote access ended: ") == Some(INCOMPATIBLE_PROTOCOL_MESSAGE) } diff --git a/apps/maple-agent/app/src/remote_updates.rs b/apps/maple-agent/app/src/remote_updates.rs index 374eafd16..6d04e5ac1 100644 --- a/apps/maple-agent/app/src/remote_updates.rs +++ b/apps/maple-agent/app/src/remote_updates.rs @@ -15,12 +15,64 @@ use std::{ sync::{Arc, Mutex}, time::{Duration, Instant}, }; +use tokio::sync::watch; use uuid::Uuid; const LEASE_TIME: Duration = Duration::from_secs(20); const MAX_RECEIPTS: usize = 512; const MAX_JOURNAL: u64 = 1024 * 1024; const ACK_LIFETIME: Duration = Duration::from_secs(60); +const REFRESH_TIMEOUT: Duration = Duration::from_secs(40); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum RefreshFailure { + Busy, + Disabled, + Unavailable, + CheckFailed, + TimedOut, +} +impl RefreshFailure { + pub(crate) fn message(self) -> &'static str { + match self { + Self::Busy => { + "The remote updater is busy. Wait for its current operation, then check again." + } + Self::Disabled => "Updates are unavailable for this remote installation.", + Self::Unavailable => "Open Maple on the remote computer before checking for updates.", + Self::CheckFailed => { + "The remote computer could not check its signed update feed. Its last reported status is not a fresh check." + } + Self::TimedOut => { + "The remote update check timed out. Its last reported status is not a fresh check." + } + } + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) struct DesktopRefreshCommand { + pub id: String, +} +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct RefreshCompleteRequest { + pub expected_account: String, + pub lease: String, + pub id: String, + pub result: Result, +} + +#[derive(Clone)] +struct Refresh { + id: String, + account: String, + lease: String, + expires: Instant, + claimed: bool, + completion: watch::Sender>>, +} #[derive(Clone, Debug, Serialize, Deserialize)] pub(crate) struct DesktopLease { @@ -83,6 +135,8 @@ struct State { receipts: BTreeMap, lease: Option, authorities: BTreeMap, + // Ephemeral: a restart never recreates or retries a remote check. + refresh: Option, } pub(crate) struct Broker { @@ -185,6 +239,151 @@ impl Broker { .await } + pub(crate) async fn request_refresh( + &self, + account: String, + peer: String, + ) -> Result { + let (completion, mut response) = watch::channel(None); + let expected_account = account.clone(); + let (id, owner) = self + .access(move |state, _| { + let owner = state + .lease + .as_ref() + .filter(|lease| { + lease.account == expected_account && lease.expires > Instant::now() + }) + .ok_or_else(|| RefreshFailure::Unavailable.message().to_owned())?; + if state.refresh.is_some() + || state.receipts.values().any(|receipt| { + receipt.lease == owner.id + && matches!( + receipt.operation.state, + OperationState::PendingAck + | OperationState::Queued + | OperationState::Installing + ) + }) + { + return Err(RefreshFailure::Busy.message().into()); + } + let id = Uuid::new_v4().to_string(); + let owner = owner.id.clone(); + state.refresh = Some(Refresh { + id: id.clone(), + account: expected_account, + lease: owner.clone(), + expires: Instant::now() + REFRESH_TIMEOUT, + claimed: false, + completion, + }); + Ok((id, owner)) + }) + .await?; + let result = tokio::time::timeout(REFRESH_TIMEOUT, async { + loop { + let completed = response.borrow().clone(); + if let Some(result) = completed { + return result; + } + if response.changed().await.is_err() { + return Err(RefreshFailure::Unavailable); + } + } + }) + .await + .unwrap_or(Err(RefreshFailure::TimedOut)); + // Consume only this generation under the same desktop/account owner. + // A heartbeat or a replacement desktop cannot manufacture completion. + self.access(move |state, store| { + let owned = state.lease.as_ref().is_some_and(|lease| { + lease.id == owner && lease.account == account && lease.expires > Instant::now() + }); + if !owned { + return Ok(Err(RefreshFailure::Unavailable)); + } + if !state + .refresh + .as_ref() + .is_some_and(|refresh| refresh.id == id && refresh.lease == owner) + { + return Ok(Err(result.err().unwrap_or(RefreshFailure::Unavailable))); + } + state.refresh = None; + Ok(result.map(|mut status| { + apply_store_policy(&mut status, store); + let account = scope_binding("account", &account); + let peer = scope_binding("peer", &peer); + status.operation = state + .receipts + .values() + .filter(|receipt| { + receipt.account_binding == account && receipt.peer_binding == peer + }) + .max_by_key(|receipt| receipt.sequence) + .map(|receipt| receipt.operation.clone()); + status + })) + }) + .await? + .map_err(|failure| failure.message().to_owned()) + } + + pub(crate) async fn take_refresh( + &self, + request: TakeRequest, + ) -> Result, String> { + self.access(move |state, _| { + lease(state, &request.expected_account, &request.lease)?; + let Some(refresh) = state.refresh.as_mut().filter(|refresh| { + refresh.account == request.expected_account + && refresh.lease == request.lease + && !refresh.claimed + }) else { + return Ok(None); + }; + refresh.claimed = true; + Ok(Some(DesktopRefreshCommand { + id: refresh.id.clone(), + })) + }) + .await + } + + pub(crate) async fn complete_refresh( + &self, + request: RefreshCompleteRequest, + ) -> Result<(), String> { + self.access(move |state, _| { + lease(state, &request.expected_account, &request.lease)?; + let refresh = state + .refresh + .as_ref() + .filter(|refresh| { + refresh.id == request.id + && refresh.account == request.expected_account + && refresh.lease == request.lease + && refresh.claimed + }) + .ok_or("Remote update check expired or changed owner")?; + if refresh.completion.borrow().is_some() { + return Err("Remote update check already completed".into()); + } + let mut result = request.result; + if let Ok(status) = &mut result { + if let Some(offered) = &status.offered { + offered.validate()?; + } + status.operation = None; + state.lease.as_mut().expect("checked lease").status = status.clone(); + } + refresh.completion.send_replace(Some(result)); + Ok(()) + }) + .await + } + pub(crate) async fn take( &self, request: TakeRequest, @@ -259,14 +458,7 @@ impl Broker { }) .map(|l| l.status.clone()) .unwrap_or_else(unavailable); - match store.map(|s| s.lock_status()) { - Some(Ok(lock)) if lock.mode == StoreMode::Password => { - status.blocked = Some(BlockedReason::PasswordProtected) - } - Some(Ok(lock)) if lock.state == StoreState::Open => {} - Some(_) => status.blocked = Some(BlockedReason::Busy), - None => status.blocked = Some(BlockedReason::DesktopUnavailable), - } + apply_store_policy(&mut status, store); status.operation = state .receipts .values() @@ -314,6 +506,9 @@ impl Broker { "Remote update receipt limit reached; update this computer manually".into(), ); } + if state.refresh.is_some() { + return Err("A remote update check is in progress".into()); + } let lease = state .lease .as_ref() @@ -574,6 +769,14 @@ fn scope_binding(domain: &str, value: &str) -> String { } fn expire_requests(state: &mut State) { let now = Instant::now(); + if state.refresh.as_ref().is_some_and(|refresh| { + refresh.expires <= now + || !state.lease.as_ref().is_some_and(|lease| { + lease.id == refresh.lease && lease.account == refresh.account && lease.expires > now + }) + }) { + state.refresh = None; + } let milliseconds = now_ms(); for (id, receipt) in &mut state.receipts { let active = state.lease.as_ref().is_some_and(|l| { @@ -610,6 +813,16 @@ fn plain_store(store: Option<&Store>) -> Result { .store_id .ok_or_else(|| "Remote local data identity is unavailable".into()) } +fn apply_store_policy(status: &mut RemoteUpdateStatus, store: Option<&Store>) { + match store.map(Store::lock_status) { + Some(Ok(lock)) if lock.mode == StoreMode::Password => { + status.blocked = Some(BlockedReason::PasswordProtected); + } + Some(Ok(lock)) if lock.state == StoreState::Open => {} + Some(_) => status.blocked = Some(BlockedReason::Busy), + None => status.blocked = Some(BlockedReason::DesktopUnavailable), + } +} fn unavailable() -> RemoteUpdateStatus { RemoteUpdateStatus { state: UpdateState::Unknown, @@ -784,6 +997,179 @@ mod tests { let broker = Broker::new(Some(store.clone())); (directory, store, broker) } + + async fn claim_refresh(broker: &Broker, lease: &DesktopLease) -> DesktopRefreshCommand { + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let Some(command) = broker + .take_refresh(TakeRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + }) + .await + .unwrap() + { + return command; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("queued refresh must become claimable") + } + + #[tokio::test] + async fn fresh_refresh_requires_one_claimed_generation_not_a_cached_publication() { + let (_directory, _store, broker) = fixture(); + let broker = Arc::new(broker); + let lease = desktop(&broker).await; + let waiting = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .request_refresh("alice".into(), "paired-peer".into()) + .await + } + }); + let command = claim_refresh(&broker, &lease).await; + broker + .publish(PublishRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + status: RemoteUpdateStatus { + state: UpdateState::Current, + ..Default::default() + }, + }) + .await + .unwrap(); + assert!( + !waiting.is_finished(), + "a heartbeat is not a fresh-check completion" + ); + assert!( + broker + .take_refresh(TakeRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + }) + .await + .unwrap() + .is_none(), + "a check is claimed only once" + ); + assert!( + broker + .request_refresh("alice".into(), "paired-peer".into()) + .await + .unwrap_err() + .contains("busy") + ); + assert!( + broker + .complete_refresh(RefreshCompleteRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + id: Uuid::new_v4().to_string(), + result: Ok(RemoteUpdateStatus::default()), + }) + .await + .is_err() + ); + assert!(!waiting.is_finished()); + let fresh = RemoteUpdateStatus { + state: UpdateState::Available, + offered: Some(target()), + ..Default::default() + }; + broker + .complete_refresh(RefreshCompleteRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + id: command.id, + result: Ok(fresh.clone()), + }) + .await + .unwrap(); + assert_eq!(waiting.await.unwrap().unwrap(), fresh); + assert!(broker.state.lock().unwrap().receipts.is_empty()); + } + + #[tokio::test] + async fn failed_or_disconnected_refresh_never_returns_the_old_offer() { + for disconnect in [false, true] { + let (_directory, _store, broker) = fixture(); + let broker = Arc::new(broker); + let lease = desktop(&broker).await; + let waiting = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .request_refresh("alice".into(), "paired-peer".into()) + .await + } + }); + let command = claim_refresh(&broker, &lease).await; + if disconnect { + broker + .access(|state, _| { + state.lease.as_mut().unwrap().expires = + Instant::now() - Duration::from_secs(1); + Ok(()) + }) + .await + .unwrap(); + let _ = broker.status("alice", "paired-peer").await.unwrap(); + } else { + broker + .complete_refresh(RefreshCompleteRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + id: command.id, + result: Err(RefreshFailure::CheckFailed), + }) + .await + .unwrap(); + } + let result = tokio::time::timeout(Duration::from_secs(2), waiting) + .await + .unwrap() + .unwrap(); + assert!(result.is_err()); + } + } + + #[tokio::test] + async fn refresh_rechecks_store_policy_after_the_desktop_check() { + let (_directory, store, broker) = fixture(); + let broker = Arc::new(broker); + let lease = desktop(&broker).await; + let waiting = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .request_refresh("alice".into(), "paired-peer".into()) + .await + } + }); + let command = claim_refresh(&broker, &lease).await; + store.close().unwrap(); + broker + .complete_refresh(RefreshCompleteRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + id: command.id, + result: Ok(RemoteUpdateStatus { + state: UpdateState::Ready, + offered: Some(target()), + ..Default::default() + }), + }) + .await + .unwrap(); + let status = waiting.await.unwrap().unwrap(); + assert_eq!(status.blocked, Some(BlockedReason::Busy)); + } fn target() -> ReleaseIdentity { ReleaseIdentity { version: "0.1.0".into(), @@ -1225,8 +1611,34 @@ mod tests { }) .await .unwrap(); - let broker = Broker::new(Some(store.clone())); + let broker = Arc::new(Broker::new(Some(store.clone()))); let lease = desktop(&broker).await; + let check = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .request_refresh("alice".into(), "paired-peer".into()) + .await + } + }); + let command = claim_refresh(&broker, &lease).await; + broker + .complete_refresh(RefreshCompleteRequest { + expected_account: "alice".into(), + lease: lease.id.clone(), + id: command.id, + result: Ok(RemoteUpdateStatus { + state: UpdateState::Ready, + offered: Some(target()), + ..Default::default() + }), + }) + .await + .unwrap(); + assert_eq!( + check.await.unwrap().unwrap().blocked, + Some(BlockedReason::PasswordProtected) + ); receipt(&broker, &lease, OperationState::Queued).await; assert_eq!(store.state(), StoreState::Open); assert!( diff --git a/apps/maple-agent/app/src/ui/remote.rs b/apps/maple-agent/app/src/ui/remote.rs index 8ac2350ee..a29cc856c 100644 --- a/apps/maple-agent/app/src/ui/remote.rs +++ b/apps/maple-agent/app/src/ui/remote.rs @@ -4,7 +4,7 @@ use crate::{ backend::AgentBackend, remote::{ manager::{PairingWindow, Status}, - update::{InstallAccepted, InstallRequest, OperationState, PeerInfo, UpdateState}, + update::{InstallReply, InstallRequest, OperationState, PeerInfo, UpdateState}, }, ui::{task, text_input::TextInput, theme, widgets}, }; @@ -13,6 +13,7 @@ use maple_sdk::remote::RemoteDevice; use std::{ cell::RefCell, collections::HashMap, + future::Future, sync::Arc, time::{Duration, Instant}, }; @@ -57,6 +58,35 @@ enum InstallPhase { Failed, } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum PeerInfoQuery { + #[default] + Read, + Check, +} + +async fn request_peer_info( + query: PeerInfoQuery, + account: String, + peer: String, + call: F, +) -> Result, String> +where + F: FnOnce(&'static str, serde_json::Value) -> Fut, + Fut: Future>, +{ + let method = match query { + PeerInfoQuery::Read => "remote_peer_info", + PeerInfoQuery::Check => "remote_refresh_update", + }; + let value = call(method, serde_json::json!([account, peer])).await?; + match query { + PeerInfoQuery::Read => serde_json::from_value(value), + PeerInfoQuery::Check => serde_json::from_value(value).map(Some), + } + .map_err(|_| "Could not read remote update status".to_owned()) +} + struct InstallAttempt { request: InstallRequest, phase: InstallPhase, @@ -67,6 +97,8 @@ struct InstallAttempt { struct ComputerUpdate { generation: u64, phase: PeerInfoPhase, + query: PeerInfoQuery, + check_required: bool, info: Option, running_label: String, update_label: String, @@ -154,8 +186,10 @@ impl ComputerUpdate { let Some(info) = &self.info else { self.running_label = "Version unavailable".into(); self.update_label = match self.phase { - PeerInfoPhase::Loading => "Reading update status…", + PeerInfoPhase::Loading if self.query == PeerInfoQuery::Check => "Checking for updates on this computer…", + PeerInfoPhase::Loading => "Reading reported update status…", PeerInfoPhase::Unsupported => "Remote update status is unavailable. Use Open projects for ordinary remote access.", + PeerInfoPhase::Unreachable if self.query == PeerInfoQuery::Check => "Could not check for updates. Check that this computer is online, then refresh.", PeerInfoPhase::Unreachable => "Could not read update status. Check that this computer is online, then refresh.", PeerInfoPhase::Incompatible => PROTOCOL_UPDATE_REQUIRED, _ => "Refresh to read this computer’s version and update status.", @@ -174,15 +208,33 @@ impl ComputerUpdate { None => format!("{prefix} v{}", info.running.version), }; self.update_label = if self.phase == PeerInfoPhase::Loading { - "Refreshing update status…".into() + if self.query == PeerInfoQuery::Check { + "Checking for updates on this computer…" + } else { + "Reading reported update status…" + } + .into() } else if self.phase == PeerInfoPhase::Unreachable { - "Could not refresh update status. The last reported version is shown above.".into() + if self.query == PeerInfoQuery::Check && !info.supports_remote_refresh() { + "This version cannot check for updates remotely. Check on that computer; its last reported status is shown above." + } else if self.query == PeerInfoQuery::Check { + "Could not check for updates. The last reported version is shown above." + } else { + "Could not read update status. The last reported version is shown above." + }.into() } else if self.phase == PeerInfoPhase::Incompatible { PROTOCOL_UPDATE_REQUIRED.into() - } else if let Some(offered) = &info.update.offered { - format!("v{} · {}", offered.version, info.update.state.label()) } else { - info.update.state.label().into() + let status = if let Some(offered) = &info.update.offered { + format!("v{} · {}", offered.version, info.update.state.label()) + } else { + info.update.state.label().into() + }; + if self.phase == PeerInfoPhase::Loaded && self.query == PeerInfoQuery::Check { + format!("Update check: {status}") + } else { + format!("Last reported update: {status}") + } }; // Cached observations remain useful as history, but must not look like // a current offer, policy decision or version comparison after failure. @@ -218,8 +270,16 @@ impl ComputerUpdate { } InstallPhase::Failed => false, }); - self.install_available = - self.phase == PeerInfoPhase::Loaded && !pending && blocked.is_none(); + self.install_available = self.phase == PeerInfoPhase::Loaded + && !pending + && !self.check_required + && blocked.is_none(); + if self.check_required && self.blocked_label.is_none() { + self.blocked_label = Some( + "Refresh this computer’s update status before reviewing another update request." + .into(), + ); + } } fn reconcile_operation(&mut self) { @@ -436,8 +496,25 @@ impl RemoteScreen { .map(|peer| peer.grant) } + fn install_grant(&self, peer: &str) -> Option<[u8; 16]> { + self.status + .as_ref()? + .peers + .get(peer) + .filter(|peer| peer.can_connect) + .and_then(|peer| peer.host_grant) + } + + fn can_install(&self, peer: &str) -> bool { + self.install_grant(peer).is_some() + && self + .computer_updates + .get(peer) + .is_some_and(|update| update.install_available) + } + fn confirmation_is_current(&self, confirmation: &InstallConfirmation) -> bool { - self.peer_grant(&confirmation.peer) == Some(confirmation.grant) + self.install_grant(&confirmation.peer) == Some(confirmation.grant) && self .computer_updates .get(&confirmation.peer) @@ -463,6 +540,14 @@ impl RemoteScreen { } fn refresh_peer_info(&mut self, peer: String, cx: &mut Context) { + self.query_peer_info(peer, PeerInfoQuery::Read, cx); + } + + fn check_peer_update(&mut self, peer: String, cx: &mut Context) { + self.query_peer_info(peer, PeerInfoQuery::Check, cx); + } + + fn query_peer_info(&mut self, peer: String, query: PeerInfoQuery, cx: &mut Context) { let Some(grant) = self.peer_grant(&peer) else { return; }; @@ -478,6 +563,7 @@ impl RemoteScreen { let update = self.computer_updates.entry(peer.clone()).or_default(); update.generation = generation; update.phase = PeerInfoPhase::Loading; + update.query = query; update.refresh_presentation(); self.invalidate_confirmation(); let backend = self.backend.clone(); @@ -486,13 +572,16 @@ impl RemoteScreen { let bridge = task::call( &self.backend, async move { - backend - .remote_setting::>("remote_peer_info", (account, target)) - .await + request_peer_info(query, account, target, move |method, params| async move { + backend + .remote_setting::(method, params) + .await + }) + .await }, cx, move |screen, result, cx| { - screen.apply_peer_info(&peer, grant, generation, result); + screen.apply_peer_info_result(&peer, grant, generation, query, result); cx.notify(); }, ); @@ -500,12 +589,24 @@ impl RemoteScreen { cx.notify(); } + #[cfg(test)] fn apply_peer_info( &mut self, peer: &str, grant: [u8; 16], generation: u64, result: Result, String>, + ) { + self.apply_peer_info_result(peer, grant, generation, PeerInfoQuery::Read, result); + } + + fn apply_peer_info_result( + &mut self, + peer: &str, + grant: [u8; 16], + generation: u64, + query: PeerInfoQuery, + result: Result, String>, ) { if self.peer_grant(peer) != Some(grant) { return; @@ -516,9 +617,13 @@ impl RemoteScreen { if update.generation != generation { return; } + update.query = query; match result { Ok(Some(info)) => { update.phase = PeerInfoPhase::Loaded; + if query == PeerInfoQuery::Check { + update.check_required = false; + } update.info = Some(info); update.reconcile_operation(); } @@ -542,7 +647,7 @@ impl RemoteScreen { if self.install_confirmation.is_some() { return; } - let Some(grant) = self.peer_grant(&peer) else { + let Some(grant) = self.install_grant(&peer) else { return; }; let Some(update) = self @@ -655,7 +760,7 @@ impl RemoteScreen { &self.backend, async move { backend - .remote_setting::( + .remote_setting::( "remote_install_update", (intent.account, intent.peer, intent.grant, intent.request), ) @@ -676,9 +781,9 @@ impl RemoteScreen { peer: &str, grant: [u8; 16], request: &InstallRequest, - result: Result, + result: Result, ) { - if self.peer_grant(peer) != Some(grant) { + if self.install_grant(peer) != Some(grant) { return; } let Some(update) = self.computer_updates.get_mut(peer) else { @@ -689,15 +794,23 @@ impl RemoteScreen { }) else { return; }; - if let Ok(reply) = result - && reply.operation_id == request.operation_id - && reply.target == request.target - { - attempt.phase = InstallPhase::Accepted; - attempt.label = "Update accepted. Remote access will disconnect while this computer restarts. Refresh to check when it returns.".into(); - } else { - attempt.phase = InstallPhase::Uncertain; - attempt.label = UNCONFIRMED_INSTALL.into(); + match result { + Ok(InstallReply::Accepted { receipt }) + if receipt.operation_id == request.operation_id + && receipt.target == request.target => + { + attempt.phase = InstallPhase::Accepted; + attempt.label = "Update accepted. Remote access will disconnect while this computer restarts. Refresh to check when it returns.".into(); + } + Ok(InstallReply::Refused { reason }) => { + attempt.phase = InstallPhase::Failed; + attempt.label = format!("Update request declined: {}", reason.message()); + update.check_required = true; + } + _ => { + attempt.phase = InstallPhase::Uncertain; + attempt.label = UNCONFIRMED_INSTALL.into(); + } } update.refresh_presentation(); } @@ -743,15 +856,40 @@ impl RemoteScreen { { self.clear_selection(cx); } - self.computer_updates.retain(|key, _| { - status.peers.get(key).is_some_and(|peer| { - peer.can_connect - && self - .status - .as_ref() - .and_then(|old| old.peers.get(key)) - .is_some_and(|old| old.can_connect && old.grant == peer.grant) - }) + self.computer_updates.retain(|key, update| { + let old = self.status.as_ref().and_then(|old| old.peers.get(key)); + let peer = status.peers.get(key); + if old.zip(peer).is_some_and(|(old, peer)| { + old.can_connect + && peer.can_connect + && old.grant == peer.grant + && old.host_grant == peer.host_grant + }) { + return true; + } + // Neither re-pairing direction settles an outstanding install. + // Retain its nonce through revocation/recovery, but invalidate all + // observations and callbacks made under the previous bindings. + let Some(attempt) = update.attempt.as_mut().filter(|attempt| { + matches!( + attempt.phase, + InstallPhase::Sending | InstallPhase::Accepted | InstallPhase::Uncertain + ) + }) else { + return false; + }; + let outgoing_changed = old.zip(peer).is_none_or(|(old, peer)| { + !old.can_connect || !peer.can_connect || old.host_grant != peer.host_grant + }); + if outgoing_changed && attempt.phase == InstallPhase::Sending { + attempt.phase = InstallPhase::Uncertain; + attempt.label = UNCONFIRMED_INSTALL.into(); + } + update.generation = update.generation.wrapping_add(1); + update.phase = PeerInfoPhase::Unchecked; + update.query = PeerInfoQuery::Read; + update.refresh_presentation(); + true }); self.status = Some(status); self.invalidate_confirmation(); @@ -1299,7 +1437,24 @@ impl RemoteScreen { fn render_computer_update(&self, peer: &str, cx: &mut Context) -> Div { let update = self.computer_updates.get(peer); let loading = update.is_some_and(|update| update.phase == PeerInfoPhase::Loading); - let can_install = update.is_some_and(|update| update.install_available); + let can_install = self.can_install(peer); + let blocked_label = update + .and_then(|update| update.blocked_label.clone()) + .or_else(|| { + (self.install_grant(peer).is_none() + && update.is_some_and(|update| { + update.phase == PeerInfoPhase::Loaded + && update + .info + .as_ref() + .is_some_and(PeerInfo::supports_remote_updates) + })) + .then(|| { + crate::remote::update::BlockedReason::PairingConfirmationRequired + .message() + .to_owned() + }) + }); let has_offer = update.is_some_and(|update| { update .info @@ -1352,17 +1507,14 @@ impl RemoteScreen { ) }, ) - .when_some( - update.and_then(|update| update.blocked_label.as_ref()), - |row, reason| { - row.child( - div() - .text_sm() - .text_color(gpui::rgb(theme::text_muted())) - .child(reason.clone()), - ) - }, - ) + .when_some(blocked_label.as_ref(), |row, reason| { + row.child( + div() + .text_sm() + .text_color(gpui::rgb(theme::text_muted())) + .child(reason.clone()), + ) + }) .when_some( update.and_then(|update| update.attempt.as_ref()), |row, attempt| row.child(div().text_sm().child(attempt.label.clone())), @@ -1378,16 +1530,21 @@ impl RemoteScreen { let peer = peer.to_owned(); move || format!("refresh-update-{peer}") }) - .aria_label("Refresh this computer’s version and update status") + .aria_label("Check this computer for updates and refresh its status") .child(if loading { - "Reading status…" + if update.is_some_and(|update| update.query == PeerInfoQuery::Check) + { + "Checking for updates…" + } else { + "Reading status…" + } } else { "Refresh update status" }) .when(loading, |button| button.opacity(0.5)) .when(!loading, |button| { button.on_click(cx.listener(move |screen, _, _, cx| { - screen.refresh_peer_info(refresh.clone(), cx) + screen.check_peer_update(refresh.clone(), cx) })) }), ) diff --git a/apps/maple-agent/app/src/ui/remote/tests.rs b/apps/maple-agent/app/src/ui/remote/tests.rs index d620b83b7..ad9424c6e 100644 --- a/apps/maple-agent/app/src/ui/remote/tests.rs +++ b/apps/maple-agent/app/src/ui/remote/tests.rs @@ -270,7 +270,7 @@ fn seed_update(screen: &mut RemoteScreen, key: &str, grant: [u8; 16]) { screen.status.as_mut().unwrap().peers.insert( key.into(), serde_json::from_value(serde_json::json!({ - "name": format!("Office {key}"), "grant": grant, "can_connect": true, + "name": format!("Office {key}"), "grant": grant, "host_grant": grant, "can_connect": true, "controls_this_computer": false, "pending_confirmation": null })) .unwrap(), @@ -798,7 +798,7 @@ fn stale_grant_or_offer_invalidates_remote_update_consent(cx: &mut TestAppContex .peers .get_mut("a") .unwrap() - .grant = [2; 16]; + .host_grant = Some([2; 16]); assert!( screen.take_install_intent().is_none(), "submission rechecks pairing even before a refresh callback" @@ -811,7 +811,7 @@ fn stale_grant_or_offer_invalidates_remote_update_consent(cx: &mut TestAppContex .peers .get_mut("a") .unwrap() - .grant = [1; 16]; + .host_grant = Some([1; 16]); }); cx.update(|window, cx| { screen.update(cx, |screen, cx| { @@ -986,7 +986,10 @@ fn current_and_idle_without_offers_do_not_show_download_instructions() { ..Default::default() }; update.refresh_presentation(); - assert_eq!(update.update_label, state.label()); + assert_eq!( + update.update_label, + format!("Last reported update: {}", state.label()) + ); assert!(update.blocked_label.is_none()); assert!( !update.install_available, @@ -1036,7 +1039,7 @@ fn confirmed_remote_outcome_stays_historical_after_later_upgrade() { let attempt = update.attempt.as_ref().unwrap(); assert_eq!(attempt.phase, InstallPhase::Succeeded); assert_eq!(attempt.label, confirmed); - assert_eq!(update.update_label, "Up to date"); + assert_eq!(update.update_label, "Last reported update: Up to date"); assert!(update.blocked_label.is_none()); assert!(!update.install_available); } @@ -1096,3 +1099,519 @@ fn only_matching_success_receipt_can_finish_a_nonterminal_request() { ); } } + +#[tokio::test] +async fn status_reads_and_explicit_checks_invoke_different_backend_routes() { + let mut calls = Vec::new(); + for (query, method) in [ + (PeerInfoQuery::Read, "remote_peer_info"), + (PeerInfoQuery::Check, "remote_refresh_update"), + ] { + let mut info = ready_peer(); + info.update.offered.as_mut().unwrap().version = if query == PeerInfoQuery::Read { + "0.0.7" + } else { + "0.0.8" + } + .into(); + let response = serde_json::to_value(&info).unwrap(); + let reported = request_peer_info( + query, + "account-a".into(), + "remote-a".into(), + |called, params| { + calls.push((called, params)); + async move { Ok(response) } + }, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(reported, info); + assert_eq!( + calls.last().unwrap(), + &(method, serde_json::json!(["account-a", "remote-a"])) + ); + } + assert_eq!(calls.len(), 2); +} + +#[tokio::test] +async fn fresh_check_requires_a_report_and_does_not_accept_legacy_null_as_success() { + let read = request_peer_info( + PeerInfoQuery::Read, + "account".into(), + "peer".into(), + |method, _| async move { + assert_eq!(method, "remote_peer_info"); + Ok(serde_json::Value::Null) + }, + ) + .await + .unwrap(); + assert!(read.is_none()); + let checked = request_peer_info( + PeerInfoQuery::Check, + "account".into(), + "peer".into(), + |method, _| async move { + assert_eq!(method, "remote_refresh_update"); + Ok(serde_json::Value::Null) + }, + ) + .await; + assert_eq!(checked.unwrap_err(), "Could not read remote update status"); +} + +#[test] +fn reported_status_is_not_presented_as_a_fresh_update_check() { + for state in [ + UpdateState::Idle, + UpdateState::Checking, + UpdateState::Current, + UpdateState::Available, + UpdateState::Downloading, + UpdateState::Ready, + UpdateState::Error, + ] { + let mut info = ready_peer(); + info.update.state = state; + if matches!( + state, + UpdateState::Idle | UpdateState::Checking | UpdateState::Current + ) { + info.update.offered = None; + } + let mut update = ComputerUpdate { + phase: PeerInfoPhase::Loaded, + info: Some(info), + ..Default::default() + }; + update.refresh_presentation(); + assert!(update.update_label.starts_with("Last reported update:")); + update.query = PeerInfoQuery::Check; + update.refresh_presentation(); + assert!(update.update_label.starts_with("Update check:")); + update.phase = PeerInfoPhase::Unreachable; + update.refresh_presentation(); + assert!(update.running_label.starts_with("Last reported")); + assert!(!update.update_label.starts_with("Update check:")); + assert!(!update.install_available); + } + let mut cold = ComputerUpdate { + query: PeerInfoQuery::Check, + phase: PeerInfoPhase::Loading, + ..Default::default() + }; + cold.refresh_presentation(); + assert_eq!(cold.running_label, "Version unavailable"); + assert!(cold.update_label.contains("Checking for updates")); + cold.phase = PeerInfoPhase::Unreachable; + cold.refresh_presentation(); + assert!(cold.update_label.contains("Could not check")); +} + +#[gpui::test] +fn missing_outgoing_grant_blocks_consent_but_still_allows_status_reads(cx: &mut TestAppContext) { + let (screen, cx) = show_update(cx); + screen.update(cx, |screen, _| { + screen + .status + .as_mut() + .unwrap() + .peers + .get_mut("a") + .unwrap() + .host_grant = None; + assert_eq!(screen.peer_grant("a"), Some([1; 16])); + assert!(screen.install_grant("a").is_none()); + assert!(!screen.can_install("a")); + }); + cx.update(|window, cx| { + screen.update(cx, |screen, cx| { + screen.ask_remote_install("a".into(), window, cx); + assert!(screen.install_confirmation.is_none()); + screen.refresh_peer_info("a".into(), cx); + assert_eq!(screen.computer_updates["a"].phase, PeerInfoPhase::Loading); + assert_eq!(screen.computer_updates["a"].query, PeerInfoQuery::Read); + }) + }); +} + +#[gpui::test] +fn explicit_refresh_button_requests_a_check_without_an_outgoing_install_grant( + cx: &mut TestAppContext, +) { + let (screen, cx) = show_update(cx); + screen.update(cx, |screen, cx| { + screen + .status + .as_mut() + .unwrap() + .peers + .get_mut("a") + .unwrap() + .host_grant = None; + cx.notify(); + }); + cx.run_until_parked(); + screen.read_with(cx, |screen, _| { + assert_eq!(screen.computer_updates["a"].phase, PeerInfoPhase::Loaded); + assert_eq!(screen.computer_updates["a"].query, PeerInfoQuery::Read); + assert!(screen.install_grant("a").is_none()); + }); + let refresh = cx.debug_bounds("refresh-update-a").unwrap(); + cx.simulate_click(refresh.center(), gpui::Modifiers::default()); + cx.run_until_parked(); + screen.read_with(cx, |screen, _| { + assert_eq!(screen.computer_updates["a"].query, PeerInfoQuery::Check); + assert!(screen.install_confirmation.is_none()); + }); +} + +#[gpui::test] +fn failed_check_keeps_only_historical_values_and_sanitized_legacy_or_offline_messages( + cx: &mut TestAppContext, +) { + let screen = screen(cx); + screen.update(cx, |screen, _| { + for legacy in [false, true] { + seed_update(screen, "a", [1; 16]); + if legacy { + screen + .computer_updates + .get_mut("a") + .unwrap() + .info + .as_mut() + .unwrap() + .protocol + .capabilities + .retain(|cap| cap != crate::remote::update::REFRESH_CAPABILITY); + } + screen.apply_peer_info_result( + "a", + [1; 16], + 1, + PeerInfoQuery::Check, + Err("private-canary https://user:token@example.invalid/private".into()), + ); + let update = &screen.computer_updates["a"]; + assert!(update.running_label.starts_with("Last reported")); + assert!(!update.install_available); + assert!(!update.update_label.contains("private-canary")); + assert!(!update.update_label.contains("token")); + assert_eq!( + update + .update_label + .contains("cannot check for updates remotely"), + legacy + ); + assert!(screen.status.as_ref().unwrap().peers["a"].can_connect); + } + }); +} + +#[gpui::test] +fn definite_install_refusal_is_not_uncertain_and_requires_a_new_check(cx: &mut TestAppContext) { + use crate::remote::update::InstallRefusal; + let screen = screen(cx); + screen.update(cx, |screen, _| { + for reason in [ + InstallRefusal::AuthorizationChanged, + InstallRefusal::UpdateChanged, + InstallRefusal::NotReady, + InstallRefusal::Unsupported, + InstallRefusal::AccountChanged, + InstallRefusal::Busy, + InstallRefusal::Unavailable, + ] { + seed_update(screen, "a", [1; 16]); + let request = InstallRequest { + operation_id: uuid::Uuid::new_v4().to_string(), + target: target_release(), + }; + screen.computer_updates.get_mut("a").unwrap().attempt = Some(InstallAttempt { + request: request.clone(), + phase: InstallPhase::Sending, + label: String::new(), + }); + screen.apply_install_reply( + "a", + [1; 16], + &request, + Ok(InstallReply::Refused { reason }), + ); + let update = &screen.computer_updates["a"]; + let attempt = update.attempt.as_ref().unwrap(); + assert_eq!(attempt.phase, InstallPhase::Failed); + assert!(attempt.label.contains(reason.message())); + assert!(!attempt.label.contains("Do not resend")); + assert!(!screen.can_install("a")); + screen.apply_peer_info("a", [1; 16], 1, Ok(Some(ready_peer()))); + assert!( + !screen.can_install("a"), + "reading a cached offer is not a new check" + ); + screen.apply_peer_info_result( + "a", + [1; 16], + 1, + PeerInfoQuery::Check, + Ok(Some(ready_peer())), + ); + assert!(screen.can_install("a")); + } + }); +} + +#[gpui::test] +fn typed_acceptance_requires_the_original_operation_and_target(cx: &mut TestAppContext) { + use crate::remote::update::InstallAccepted; + let screen = screen(cx); + screen.update(cx, |screen, _| { + for matches in [false, true] { + seed_update(screen, "a", [1; 16]); + let request = InstallRequest { + operation_id: uuid::Uuid::new_v4().to_string(), + target: target_release(), + }; + screen.computer_updates.get_mut("a").unwrap().attempt = Some(InstallAttempt { + request: request.clone(), + phase: InstallPhase::Sending, + label: String::new(), + }); + let receipt = InstallAccepted { + operation_id: if matches { + request.operation_id.clone() + } else { + uuid::Uuid::new_v4().to_string() + }, + target: request.target.clone(), + }; + screen.apply_install_reply( + "a", + [1; 16], + &request, + Ok(InstallReply::Accepted { receipt }), + ); + let attempt = screen.computer_updates["a"].attempt.as_ref().unwrap(); + assert_eq!( + attempt.phase, + if matches { + InstallPhase::Accepted + } else { + InstallPhase::Uncertain + } + ); + assert_eq!(attempt.request, request); + assert!(!screen.can_install("a")); + } + }); +} + +#[gpui::test] +fn incoming_repair_does_not_erase_an_uncertain_outgoing_install(cx: &mut TestAppContext) { + let screen = screen(cx); + screen.update(cx, |screen, cx| { + seed_update(screen, "a", [1; 16]); + let request = InstallRequest { + operation_id: uuid::Uuid::new_v4().to_string(), + target: target_release(), + }; + screen.computer_updates.get_mut("a").unwrap().attempt = Some(InstallAttempt { + request: request.clone(), + phase: InstallPhase::Uncertain, + label: UNCONFIRMED_INSTALL.into(), + }); + screen.computer_updates.get_mut("a").unwrap().query = PeerInfoQuery::Check; + let mut status = screen.status.clone().unwrap(); + status.peers.get_mut("a").unwrap().grant = [9; 16]; + screen.update_status(status, cx); + assert_eq!(screen.install_grant("a"), Some([1; 16])); + let update = &screen.computer_updates["a"]; + assert_eq!(update.attempt.as_ref().unwrap().request, request); + assert_eq!(update.phase, PeerInfoPhase::Unchecked); + assert_eq!(update.query, PeerInfoQuery::Read); + assert!(update.update_label.starts_with("Last reported update:")); + assert!(!screen.can_install("a")); + let generation = update.generation; + screen.apply_peer_info_result( + "a", + [1; 16], + 1, + PeerInfoQuery::Check, + Ok(Some(ready_peer())), + ); + assert_eq!(screen.computer_updates["a"].phase, PeerInfoPhase::Unchecked); + screen.apply_peer_info_result( + "a", + [9; 16], + generation, + PeerInfoQuery::Check, + Ok(Some(ready_peer())), + ); + let attempt = screen.computer_updates["a"].attempt.as_ref().unwrap(); + assert_eq!(attempt.request, request); + assert_eq!(attempt.phase, InstallPhase::Uncertain); + assert!(!screen.can_install("a")); + }); +} + +#[gpui::test] +fn outgoing_repair_invalidates_status_and_keeps_unresolved_install_identity( + cx: &mut TestAppContext, +) { + use crate::remote::update::InstallAccepted; + let screen = screen(cx); + screen.update(cx, |screen, cx| { + for phase in [ + InstallPhase::Sending, + InstallPhase::Accepted, + InstallPhase::Uncertain, + ] { + for host_grant in [Some([9; 16]), None] { + seed_update(screen, "a", [1; 16]); + let request = InstallRequest { + operation_id: uuid::Uuid::new_v4().to_string(), + target: target_release(), + }; + let update = screen.computer_updates.get_mut("a").unwrap(); + update.query = PeerInfoQuery::Check; + update.attempt = Some(InstallAttempt { + request: request.clone(), + phase, + label: "Original request".into(), + }); + let mut status = screen.status.clone().unwrap(); + status.peers.get_mut("a").unwrap().host_grant = host_grant; + screen.update_status(status, cx); + assert_eq!(screen.peer_grant("a"), Some([1; 16])); + let update = &screen.computer_updates["a"]; + let expected = if phase == InstallPhase::Sending { + InstallPhase::Uncertain + } else { + phase + }; + assert_eq!(update.phase, PeerInfoPhase::Unchecked); + assert_eq!(update.query, PeerInfoQuery::Read); + assert_ne!(update.generation, 1); + assert!(update.update_label.starts_with("Last reported update:")); + assert_eq!(update.attempt.as_ref().unwrap().request, request); + assert_eq!(update.attempt.as_ref().unwrap().phase, expected); + if phase == InstallPhase::Sending { + assert_eq!(update.attempt.as_ref().unwrap().label, UNCONFIRMED_INSTALL); + } + assert!(!screen.can_install("a")); + let generation = update.generation; + screen.apply_peer_info_result( + "a", + [1; 16], + 1, + PeerInfoQuery::Check, + Ok(Some(ready_peer())), + ); + assert_eq!(screen.computer_updates["a"].phase, PeerInfoPhase::Unchecked); + screen.apply_install_reply( + "a", + [1; 16], + &request, + Ok(InstallReply::Accepted { + receipt: InstallAccepted { + operation_id: request.operation_id.clone(), + target: request.target.clone(), + }, + }), + ); + assert_eq!( + screen.computer_updates["a"].attempt.as_ref().unwrap().phase, + expected + ); + screen.apply_peer_info_result( + "a", + [1; 16], + generation, + PeerInfoQuery::Check, + Ok(Some(ready_peer())), + ); + assert_eq!( + screen.computer_updates["a"] + .attempt + .as_ref() + .unwrap() + .request, + request + ); + assert_eq!( + screen.computer_updates["a"].attempt.as_ref().unwrap().phase, + expected + ); + assert!(!screen.can_install("a")); + } + } + }); +} + +#[gpui::test] +fn lost_outgoing_authority_preserves_unknown_request_through_repair(cx: &mut TestAppContext) { + let screen = screen(cx); + screen.update(cx, |screen, cx| { + for absent in [false, true] { + seed_update(screen, "a", [1; 16]); + let request = InstallRequest { + operation_id: uuid::Uuid::new_v4().to_string(), + target: target_release(), + }; + screen.computer_updates.get_mut("a").unwrap().attempt = Some(InstallAttempt { + request: request.clone(), + phase: InstallPhase::Sending, + label: String::new(), + }); + let original = screen.status.clone().unwrap(); + let mut revoked = original.clone(); + if absent { + revoked.peers.remove("a"); + } else { + revoked.peers.get_mut("a").unwrap().can_connect = false; + } + screen.update_status(revoked, cx); + assert_eq!( + screen.computer_updates["a"].attempt.as_ref().unwrap().phase, + InstallPhase::Uncertain + ); + assert!(!screen.can_install("a")); + screen.update_status(original, cx); + let update = &screen.computer_updates["a"]; + assert_eq!(update.attempt.as_ref().unwrap().request, request); + assert_eq!(update.attempt.as_ref().unwrap().label, UNCONFIRMED_INSTALL); + let generation = update.generation; + screen.apply_peer_info_result( + "a", + [1; 16], + generation, + PeerInfoQuery::Check, + Ok(Some(ready_peer())), + ); + assert_eq!( + screen.computer_updates["a"].attempt.as_ref().unwrap().phase, + InstallPhase::Uncertain + ); + assert!(!screen.can_install("a")); + } + }); +} + +#[gpui::test] +fn outgoing_binding_change_discards_observations_without_unresolved_requests( + cx: &mut TestAppContext, +) { + let screen = screen(cx); + screen.update(cx, |screen, cx| { + seed_update(screen, "a", [1; 16]); + let mut status = screen.status.clone().unwrap(); + status.peers.get_mut("a").unwrap().host_grant = Some([9; 16]); + screen.update_status(status, cx); + assert!(!screen.computer_updates.contains_key("a")); + assert!(!screen.can_install("a")); + }); +} diff --git a/apps/maple-agent/app/src/update.rs b/apps/maple-agent/app/src/update.rs index 1bbbab892..3299a14c4 100644 --- a/apps/maple-agent/app/src/update.rs +++ b/apps/maple-agent/app/src/update.rs @@ -423,11 +423,51 @@ impl Controller { let Ok(_operation) = self.0.operation.try_lock() else { return; }; - self.check_inner().await; + self.check_inner(true).await; } - async fn check_inner(&self) { + /// A remote check must actually acquire the updater and read its signed + /// feed. A busy operation or failed recheck cannot turn cached Ready into + /// a successful fresh response. Downloads resume separately after the reply. + pub(crate) async fn refresh_remote( + &self, + ) -> Result + { + self.refresh_remote_with(|| self.check_inner(false)).await + } + + async fn refresh_remote_with( + &self, + check: C, + ) -> Result + where + C: FnOnce() -> F, + F: std::future::Future, + { + use crate::remote_updates::RefreshFailure; + if self.0.cancel.is_cancelled() || self.restart_path().is_some() { + return Err(RefreshFailure::Unavailable); + } if self.snapshot().state == State::Disabled { - return; + return Err(RefreshFailure::Disabled); + } + let Ok(_operation) = self.0.operation.try_lock() else { + return Err(RefreshFailure::Busy); + }; + // The broker bounds the remote caller's wait. Keep this operation + // owned until it finishes: a caller timeout must not abandon a blocking + // package validation and let a second updater operation overlap it. + tokio::select! { + _ = self.0.cancel.cancelled() => Err(RefreshFailure::Unavailable), + checked = check() => match checked { + true => Ok(self.remote_status()), + false => Err(RefreshFailure::CheckFailed), + }, + } + } + + async fn check_inner(&self, automatic_download: bool) -> bool { + if self.snapshot().state == State::Disabled { + return false; } let previous = self.snapshot(); self.begin_check(); @@ -435,7 +475,7 @@ impl Controller { Ok(service) => service, Err(error) => { self.fail(error); - return; + return false; } }; match service.updater.check().await { @@ -454,12 +494,29 @@ impl Controller { snapshot.primary = Some(UpdateAction::Check); snapshot.primary_label = "Check for Updates"; self.publish(snapshot); + true + } + Ok(CheckOutcome::Available(release)) => { + self.accept_release_with_policy(service, *release, automatic_download) + .await; + true + } + Err(_) => { + self.recheck_failed(previous); + false } - Ok(CheckOutcome::Available(release)) => self.accept_release(service, *release).await, - Err(_) => self.recheck_failed(previous), } } async fn accept_release(&self, service: &Service, release: VerifiedRelease) { + self.accept_release_with_policy(service, release, true) + .await; + } + async fn accept_release_with_policy( + &self, + service: &Service, + release: VerifiedRelease, + automatic_download: bool, + ) { let (prepared, previous_staging, previous_lock) = { let mut releases = self.0.releases.lock().await; let prepared = releases @@ -484,7 +541,10 @@ impl Controller { Ok(()) => self.release_snapshot(service, &release, State::Ready).await, Err(error) => self.installation_failed(service, &release, error).await, } - } else if self.0.automatic.load(Ordering::Relaxed) && !service.manual_only() { + } else if automatic_download + && self.0.automatic.load(Ordering::Relaxed) + && !service.manual_only() + { self.download_inner(service, release).await; } else { self.release_snapshot(service, &release, State::Available) @@ -1148,6 +1208,53 @@ fn manual_macos_path(path: &Path) -> bool { mod tests { use super::*; + #[tokio::test] + async fn remote_refresh_invokes_a_new_check_even_for_cached_current_or_ready() { + use std::sync::atomic::AtomicUsize; + for cached in [State::Idle, State::Current, State::Ready, State::Error] { + let controller = Controller::new(false); + controller.set_test_state(cached); + let calls = AtomicUsize::new(0); + let result = controller + .refresh_remote_with(|| async { + calls.fetch_add(1, Ordering::SeqCst); + controller.publish(Snapshot::new(State::Available, "New signed offer")); + true + }) + .await + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(result.state, crate::remote::update::UpdateState::Available); + } + } + + #[tokio::test] + async fn remote_refresh_failure_never_reports_cached_ready_as_fresh() { + let controller = Controller::new(false); + controller.set_test_state(State::Ready); + let result = controller.refresh_remote_with(|| async { false }).await; + assert_eq!( + result, + Err(crate::remote_updates::RefreshFailure::CheckFailed) + ); + assert_eq!(controller.snapshot().state, State::Ready); + let operation = controller.0.operation.lock().await; + assert_eq!( + controller + .refresh_remote_with(|| async { panic!("busy check must not run") }) + .await, + Err(crate::remote_updates::RefreshFailure::Busy) + ); + drop(operation); + controller.set_test_state(State::Disabled); + assert_eq!( + controller + .refresh_remote_with(|| async { panic!("disabled check must not run") }) + .await, + Err(crate::remote_updates::RefreshFailure::Disabled) + ); + } + // Model the controller after preparation has passed, without replacing an // application or bypassing the installer's separate validation tests. async fn ready_remote_fixture(root: &Path) -> (Controller, Snapshot) { diff --git a/apps/maple-agent/crates/maple-agent/src/store/records.rs b/apps/maple-agent/crates/maple-agent/src/store/records.rs index 370d19ff7..821ca0790 100644 --- a/apps/maple-agent/crates/maple-agent/src/store/records.rs +++ b/apps/maple-agent/crates/maple-agent/src/store/records.rs @@ -205,9 +205,14 @@ pub struct RemoteNamespace { #[serde(deny_unknown_fields)] pub struct RemotePeer { pub name: String, + /// Incoming controller authority; never used as an outgoing host grant. pub grant: [u8; 16], pub controls_this_computer: bool, pub can_connect: bool, + /// Confirmed outgoing authority, independent of the incoming direction. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub host_grant: Option<[u8; 16]>, + /// Durable CPace/lost-ack intent, not confirmed outgoing authority. #[serde(default)] pub pending_confirmation: Option<[u8; 16]>, } @@ -1196,8 +1201,13 @@ impl Store { checked_computer_name(&peer.name)?; self.edit_remote(account, namespace, Durability::Durable, |identity| { let mut peer = peer; + // Incoming pairing cannot introduce or replace outgoing authority. + peer.can_connect = false; + peer.host_grant = None; + peer.pending_confirmation = None; if let Some(previous) = identity.peers.get(peer_key) { - peer.can_connect |= previous.can_connect; + peer.can_connect = previous.can_connect; + peer.host_grant = previous.host_grant; peer.pending_confirmation = previous.pending_confirmation; } identity.peers.insert(peer_key.to_owned(), peer); @@ -1224,6 +1234,7 @@ impl Store { grant: [0; 16], controls_this_computer: false, can_connect: false, + host_grant: None, pending_confirmation: None, }); peer.name = name; @@ -1241,6 +1252,9 @@ impl Store { window: [u8; 16], ) -> StoreResult<()> { checked_peer_key(peer_key)?; + if window == [0; 16] { + return Err(StoreError::InvalidInput); + } self.edit_remote(account, namespace, Durability::Durable, |identity| { let peer = identity .peers @@ -1249,12 +1263,56 @@ impl Store { if peer.pending_confirmation != Some(window) { return Err(StoreError::Conflict); } + peer.host_grant = Some(window); peer.pending_confirmation = None; peer.can_connect = true; Ok(()) }) } + /// Persist an authenticated host confirmation for an existing legacy peer. + /// The caller verifies the exact endpoint and acknowledgement before this + /// compare-and-set; attempt history supplies candidates, never authority. + pub fn recover_remote_host_grant( + &self, + account: AccountId, + namespace: &RemoteNamespace, + peer_key: &str, + expected_peer: &RemotePeer, + expected_attempts: &[[u8; 16]], + window: [u8; 16], + ) -> StoreResult<()> { + checked_peer_key(peer_key)?; + if window == [0; 16] { + return Err(StoreError::InvalidInput); + } + self.edit_remote(account, namespace, Durability::Durable, |identity| { + let attempts = identity.attempted_windows.get(peer_key).map(Vec::as_slice); + if attempts != Some(expected_attempts) || !expected_attempts.contains(&window) { + return Err(StoreError::Conflict); + } + let peer = identity + .peers + .get_mut(peer_key) + .ok_or(StoreError::Conflict)?; + // Discovery may rename a peer while the host confirms its binding. + // Only authority fields participate in this compare-and-set. + if peer.grant != expected_peer.grant + || peer.controls_this_computer != expected_peer.controls_this_computer + || peer.can_connect != expected_peer.can_connect + || peer.host_grant != expected_peer.host_grant + || peer.pending_confirmation != expected_peer.pending_confirmation + || !peer.can_connect + || peer.host_grant.is_some() + || peer.pending_confirmation.is_some() + { + return Err(StoreError::Conflict); + } + peer.host_grant = Some(window); + Ok(()) + }) + } + /// Commit before sending a code-derived proof. Attempts are never evicted; /// replayed discovery cannot turn a used window back into a fresh one. pub fn reserve_remote_pairing_attempt( @@ -1423,6 +1481,9 @@ fn validate_remote_identity(identity: &RemoteIdentity) -> StoreResult<()> { for (key, peer) in &identity.peers { checked_peer_key(key)?; checked_computer_name(&peer.name)?; + if peer.host_grant == Some([0; 16]) || (peer.host_grant.is_some() && !peer.can_connect) { + return Err(StoreError::InvalidInput); + } } for (key, windows) in &identity.attempted_windows { checked_peer_key(key)?; diff --git a/apps/maple-agent/crates/maple-agent/src/store/records/tests.rs b/apps/maple-agent/crates/maple-agent/src/store/records/tests.rs index 3cb1af48a..8a3c6574f 100644 --- a/apps/maple-agent/crates/maple-agent/src/store/records/tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/store/records/tests.rs @@ -1282,12 +1282,14 @@ fn pairing_confirmation_attempts_and_discovery_preserve_authority() { grant: [4; 16], controls_this_computer: true, can_connect: false, + host_grant: None, pending_confirmation: None, }, ) .unwrap(); let before = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); assert!(before.peers[&key].can_connect && before.peers[&key].controls_this_computer); + assert_eq!(before.peers[&key].host_grant, Some([2; 16])); store .refresh_remote_peer_names( alice, @@ -1301,11 +1303,21 @@ fn pairing_confirmation_attempts_and_discovery_preserve_authority() { let after = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); assert_eq!(after.peers.len(), 1); assert_eq!(after.peers[&key].grant, before.peers[&key].grant); + assert_eq!(after.peers[&key].host_grant, before.peers[&key].host_grant); assert_eq!(after.attempted_windows, before.attempted_windows); assert!(!after.enabled); store.close().unwrap(); drop(store); let reopened = Store::open(StorePaths::new(directory.path())).unwrap(); + assert_eq!( + reopened + .saved_remote_identity(alice, &ns) + .unwrap() + .unwrap() + .peers[&key] + .host_grant, + Some([2; 16]) + ); assert_eq!( reopened.reserve_remote_pairing_attempt(alice, &ns, &key, [1; 16]), Err(StoreError::Conflict) @@ -1316,6 +1328,423 @@ fn pairing_confirmation_attempts_and_discovery_preserve_authority() { assert_eq!(saved.attempted_windows[&key], vec![[1; 16]]); } +fn incoming_remote_peer(grant: [u8; 16]) -> RemotePeer { + RemotePeer { + name: "Synthetic peer".into(), + grant, + controls_this_computer: true, + can_connect: false, + host_grant: None, + pending_confirmation: None, + } +} + +#[test] +fn remote_host_confirmation_is_durable_and_pending_proof_is_not_authority() { + let (directory, store, alice, _bob, _task) = fixture(); + let ns = namespace("local"); + let key = "11".repeat(32); + let window = [1; 16]; + store.remote_identity(alice, &ns).unwrap(); + store + .reserve_remote_pairing_attempt(alice, &ns, &key, window) + .unwrap(); + store + .remember_remote_host_proof(alice, &ns, &key, window, "Synthetic peer") + .unwrap(); + store.close().unwrap(); + drop(store); + + let store = Store::open(StorePaths::new(directory.path())).unwrap(); + let pending = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); + assert_eq!(pending.peers[&key].pending_confirmation, Some(window)); + assert_eq!(pending.peers[&key].host_grant, None); + assert!(!pending.peers[&key].can_connect); + assert_eq!( + store.confirm_remote_host(alice, &ns, &key, [2; 16]), + Err(StoreError::Conflict) + ); + assert_eq!( + store.saved_remote_identity(alice, &ns).unwrap(), + Some(pending) + ); + store.confirm_remote_host(alice, &ns, &key, window).unwrap(); + store.close().unwrap(); + drop(store); + + let reopened = Store::open(StorePaths::new(directory.path())).unwrap(); + let saved = reopened.saved_remote_identity(alice, &ns).unwrap().unwrap(); + let peer = &saved.peers[&key]; + assert_eq!(peer.host_grant, Some(window)); + assert_eq!(peer.pending_confirmation, None); + assert!(peer.can_connect); + assert_eq!(peer.grant, [0; 16]); + assert!(!peer.controls_this_computer); + assert_eq!(saved.attempted_windows[&key], vec![window]); +} + +#[test] +fn mutual_pairing_keeps_both_directions_across_sequential_and_interleaved_acks() { + for interleaved in [false, true] { + let (a_directory, a, a_account, _bob, _task) = fixture(); + let (b_directory, b, b_account, _bob, _task) = fixture(); + let ns = namespace("local"); + let a_key = "11".repeat(32); + let b_key = "22".repeat(32); + let a_window = [1; 16]; + let b_window = [2; 16]; + a.remote_identity(a_account, &ns).unwrap(); + b.remote_identity(b_account, &ns).unwrap(); + a.reserve_remote_pairing_attempt(a_account, &ns, &b_key, b_window) + .unwrap(); + a.remember_remote_host_proof(a_account, &ns, &b_key, b_window, "B") + .unwrap(); + if interleaved { + b.reserve_remote_pairing_attempt(b_account, &ns, &a_key, a_window) + .unwrap(); + b.remember_remote_host_proof(b_account, &ns, &a_key, a_window, "A") + .unwrap(); + a.grant_remote_peer(a_account, &ns, &b_key, incoming_remote_peer(a_window)) + .unwrap(); + } + b.grant_remote_peer(b_account, &ns, &a_key, incoming_remote_peer(b_window)) + .unwrap(); + a.confirm_remote_host(a_account, &ns, &b_key, b_window) + .unwrap(); + if !interleaved { + b.reserve_remote_pairing_attempt(b_account, &ns, &a_key, a_window) + .unwrap(); + b.remember_remote_host_proof(b_account, &ns, &a_key, a_window, "A") + .unwrap(); + a.grant_remote_peer(a_account, &ns, &b_key, incoming_remote_peer(a_window)) + .unwrap(); + } + b.confirm_remote_host(b_account, &ns, &a_key, a_window) + .unwrap(); + a.close().unwrap(); + b.close().unwrap(); + drop(a); + drop(b); + + let a = Store::open(StorePaths::new(a_directory.path())).unwrap(); + let b = Store::open(StorePaths::new(b_directory.path())).unwrap(); + let a_saved = a.saved_remote_identity(a_account, &ns).unwrap().unwrap(); + let b_saved = b.saved_remote_identity(b_account, &ns).unwrap().unwrap(); + let a_peer = &a_saved.peers[&b_key]; + let b_peer = &b_saved.peers[&a_key]; + assert_eq!(a_peer.grant, a_window); + assert_eq!(b_peer.grant, b_window); + assert_eq!(a_peer.host_grant, Some(b_peer.grant)); + assert_eq!(b_peer.host_grant, Some(a_peer.grant)); + assert!(a_peer.can_connect && a_peer.controls_this_computer); + assert!(b_peer.can_connect && b_peer.controls_this_computer); + assert_eq!(a_peer.pending_confirmation, None); + assert_eq!(b_peer.pending_confirmation, None); + } +} + +#[test] +fn incoming_pairing_cannot_create_outgoing_authority_or_pending_proof() { + let (_directory, store, alice, _bob, _task) = fixture(); + let ns = namespace("local"); + let key = "11".repeat(32); + store.remote_identity(alice, &ns).unwrap(); + let mut supplied = incoming_remote_peer([1; 16]); + supplied.can_connect = true; + supplied.host_grant = Some([2; 16]); + supplied.pending_confirmation = Some([3; 16]); + store.grant_remote_peer(alice, &ns, &key, supplied).unwrap(); + let saved = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); + assert_eq!(saved.peers[&key], incoming_remote_peer([1; 16])); + assert_eq!( + store.confirm_remote_host(alice, &ns, &key, [3; 16]), + Err(StoreError::Conflict) + ); + store + .reserve_remote_pairing_attempt(alice, &ns, &key, [2; 16]) + .unwrap(); + let saved = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); + assert_eq!( + store.recover_remote_host_grant( + alice, + &ns, + &key, + &saved.peers[&key], + &saved.attempted_windows[&key], + [2; 16], + ), + Err(StoreError::Conflict) + ); + assert_eq!( + store.saved_remote_identity(alice, &ns).unwrap(), + Some(saved) + ); +} + +fn legacy_remote_host( + store: &Store, + account: AccountId, + ns: &RemoteNamespace, + key: &str, + incoming_grant: Option<[u8; 16]>, +) -> RemoteIdentity { + store.remote_identity(account, ns).unwrap(); + store + .reserve_remote_pairing_attempt(account, ns, key, [1; 16]) + .unwrap(); + let legacy: RemotePeer = serde_json::from_value(json!({ + "name": "Legacy peer", + "grant": incoming_grant.unwrap_or([0; 16]), + "controls_this_computer": incoming_grant.is_some(), + "can_connect": true, + "pending_confirmation": null, + })) + .unwrap(); + assert_eq!(legacy.host_grant, None); + assert!( + serde_json::to_value(&legacy) + .unwrap() + .get("host_grant") + .is_none() + ); + store + .edit_remote(account, ns, Durability::Durable, |identity| { + identity.peers.insert(key.to_owned(), legacy); + Ok(()) + }) + .unwrap(); + store.saved_remote_identity(account, ns).unwrap().unwrap() +} + +#[test] +fn legacy_host_grant_recovery_is_durable_and_changes_only_confirmed_outgoing_binding() { + for incoming_grant in [None, Some([4; 16])] { + let (directory, store, alice, _bob, _task) = fixture(); + let ns = namespace("local"); + let key = "11".repeat(32); + let before = legacy_remote_host(&store, alice, &ns, &key, incoming_grant); + assert!(!before.enabled); + assert_eq!( + store.recover_remote_host_grant( + alice, + &ns, + &key, + &before.peers[&key], + &before.attempted_windows[&key], + [9; 16], + ), + Err(StoreError::Conflict) + ); + assert_eq!( + store.saved_remote_identity(alice, &ns).unwrap(), + Some(before.clone()) + ); + store + .recover_remote_host_grant( + alice, + &ns, + &key, + &before.peers[&key], + &before.attempted_windows[&key], + [1; 16], + ) + .unwrap(); + let mut expected = before.clone(); + expected.peers.get_mut(&key).unwrap().host_grant = Some([1; 16]); + assert_eq!( + store.saved_remote_identity(alice, &ns).unwrap(), + Some(expected.clone()) + ); + assert_eq!( + store.recover_remote_host_grant( + alice, + &ns, + &key, + &before.peers[&key], + &before.attempted_windows[&key], + [1; 16], + ), + Err(StoreError::Conflict) + ); + assert_eq!( + store.recover_remote_host_grant( + alice, + &ns, + &key, + &expected.peers[&key], + &expected.attempted_windows[&key], + [1; 16], + ), + Err(StoreError::Conflict) + ); + store.close().unwrap(); + drop(store); + let reopened = Store::open(StorePaths::new(directory.path())).unwrap(); + assert_eq!( + reopened.saved_remote_identity(alice, &ns).unwrap(), + Some(expected) + ); + } +} + +#[test] +fn host_grant_recovery_preserves_a_new_discovery_name() { + let (_directory, store, alice, _bob, _task) = fixture(); + let ns = namespace("local"); + let key = "11".repeat(32); + let before = legacy_remote_host(&store, alice, &ns, &key, Some([4; 16])); + store + .refresh_remote_peer_names( + alice, + &ns, + &BTreeMap::from([(key.clone(), "Changed peer".into())]), + ) + .unwrap(); + let mut expected = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); + expected.peers.get_mut(&key).unwrap().host_grant = Some([1; 16]); + store + .recover_remote_host_grant( + alice, + &ns, + &key, + &before.peers[&key], + &before.attempted_windows[&key], + [1; 16], + ) + .unwrap(); + let saved = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); + assert_eq!(saved, expected); + assert_eq!(saved.peers[&key].name, "Changed peer"); +} + +#[test] +fn host_grant_recovery_rejects_changed_authority_attempts_pending_proof_and_revocation() { + for changed in ["authority", "attempts", "pending", "revoked"] { + let (_directory, store, alice, _bob, _task) = fixture(); + let ns = namespace("local"); + let key = "11".repeat(32); + let before = legacy_remote_host(&store, alice, &ns, &key, Some([4; 16])); + match changed { + "authority" => store + .grant_remote_peer(alice, &ns, &key, incoming_remote_peer([5; 16])) + .unwrap(), + "attempts" => store + .reserve_remote_pairing_attempt(alice, &ns, &key, [2; 16]) + .unwrap(), + "pending" => store + .remember_remote_host_proof(alice, &ns, &key, [2; 16], "Legacy peer") + .unwrap(), + "revoked" => store.revoke_remote_peer(alice, &ns, &key).unwrap(), + _ => unreachable!(), + } + let changed_state = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); + assert_eq!( + store.recover_remote_host_grant( + alice, + &ns, + &key, + &before.peers[&key], + &before.attempted_windows[&key], + [1; 16], + ), + Err(StoreError::Conflict) + ); + if changed == "pending" { + assert_eq!( + store.recover_remote_host_grant( + alice, + &ns, + &key, + &changed_state.peers[&key], + &changed_state.attempted_windows[&key], + [1; 16], + ), + Err(StoreError::Conflict) + ); + } + assert_eq!( + store.saved_remote_identity(alice, &ns).unwrap(), + Some(changed_state) + ); + } +} + +#[test] +fn zero_host_acknowledgements_never_become_confirmed_authority() { + let (directory, store, alice, _bob, _task) = fixture(); + let ns = namespace("local"); + let key = "11".repeat(32); + store.remote_identity(alice, &ns).unwrap(); + store + .reserve_remote_pairing_attempt(alice, &ns, &key, [0; 16]) + .unwrap(); + store + .remember_remote_host_proof(alice, &ns, &key, [0; 16], "Synthetic peer") + .unwrap(); + let pending = store.saved_remote_identity(alice, &ns).unwrap().unwrap(); + assert_eq!(pending.peers[&key].host_grant, None); + assert_eq!( + store.confirm_remote_host(alice, &ns, &key, [0; 16]), + Err(StoreError::InvalidInput) + ); + assert_eq!( + store.saved_remote_identity(alice, &ns).unwrap(), + Some(pending) + ); + let legacy = legacy_remote_host(&store, alice, &ns, &key, None); + assert!(legacy.attempted_windows[&key].contains(&[0; 16])); + assert_eq!( + store.recover_remote_host_grant( + alice, + &ns, + &key, + &legacy.peers[&key], + &legacy.attempted_windows[&key], + [0; 16], + ), + Err(StoreError::InvalidInput) + ); + store.close().unwrap(); + drop(store); + let reopened = Store::open(StorePaths::new(directory.path())).unwrap(); + assert_eq!( + reopened.saved_remote_identity(alice, &ns).unwrap(), + Some(legacy) + ); +} + +#[test] +fn persisted_zero_or_unpermitted_confirmed_host_grants_are_invalid() { + for (grant, can_connect) in [([0; 16], true), ([1; 16], false)] { + let (_directory, store, alice, _bob, _task) = fixture(); + let ns = namespace("local"); + let key = "11".repeat(32); + store.remote_identity(alice, &ns).unwrap(); + let mut peer = incoming_remote_peer([2; 16]); + peer.host_grant = Some(grant); + peer.can_connect = can_connect; + let record = serde_json::to_string(&RemoteRecord { + version: 1, + peers: BTreeMap::from([(key, peer)]), + attempted_windows: BTreeMap::new(), + }) + .unwrap(); + store + .write(Durability::Durable, |tx| { + tx.execute( + "UPDATE remote_identities SET record=?2 WHERE account_id=?1", + params![alice.0, record], + )?; + Ok(()) + }) + .unwrap(); + assert_eq!( + store.saved_remote_identity(alice, &ns), + Err(StoreError::InvalidData) + ); + } +} + #[test] fn remote_record_does_not_duplicate_the_secret_and_rejects_malformed_json() { let (_directory, store, alice, _bob, _task) = fixture(); diff --git a/apps/maple-agent/docs/remote-access.md b/apps/maple-agent/docs/remote-access.md index 3cf09aa25..745e7c9c2 100644 --- a/apps/maple-agent/docs/remote-access.md +++ b/apps/maple-agent/docs/remote-access.md @@ -111,12 +111,19 @@ Quit or `maple-agent host stop` completes local shutdown. In **Settings → Remote computers**, each paired execution computer can report its running application version and update status. **Refresh update status** -queries that computer; it does not check or update the controller's installation. -It reads the remote updater's existing state; checks and automatic downloads -continue to follow that computer's own schedule and saved preference. +requests a new check by that computer's running desktop. Opening the view or +reading status in the background only reads its last reported updater state. +The explicit check requires a compatible peer with remote-refresh support; +older peers need a local check. Automatic downloads still follow the execution +computer's saved preference. Neither action installs an update or checks the +controller's installation. A failed check retains the last reported values as +history and cannot make a cached offer eligible for installation. When available, the row shows the offered version and whether its verified download is ready. Numeric package build numbers may be absent from running executable metadata; the UI does not infer them from the protocol version. +Status reads remain available while a confirmed pairing from an older version +is recovered. If confirmation cannot be recovered, ordinary access is preserved +but remote installation requires confirming or pairing that computer again. Installation requires the remote computer's running desktop updater and its exact prepared, signed candidate. A daemon without its desktop reports that @@ -149,6 +156,11 @@ after a restart. The restarted host records success only when it observes the requested version and source running, checking the package build when known. That recorded outcome survives later updates. A changed or withdrawn offer needs a fresh confirmation. +An explicit refusal before acceptance is shown as a declined request, rather +than an unconfirmed installation. Check the remote's status again before +reviewing another request. A transport failure that could have followed sending +the request remains uncertain and retains the original operation ID, even when +no receipt is visible yet. A refusal before shutdown leaves the same verified download ready to try again; it does not require quitting a healthy execution host. Uncertain shutdowns still diff --git a/apps/maple-agent/docs/updates.md b/apps/maple-agent/docs/updates.md index 7fc9f1bb5..7753097d2 100644 --- a/apps/maple-agent/docs/updates.md +++ b/apps/maple-agent/docs/updates.md @@ -35,6 +35,11 @@ menu, and the tray provide update access. Paired controllers can inspect a computer's version and prepared update through the optional [experimental remote update controls](remote-access.md#experimental-remote-updates). +Opening that view reads the remote's reported state. **Refresh update status** +requests a new update check on a compatible remote desktop; automatic downloads +still follow that computer's saved preference. It never installs an update or +checks the controller's updater. Failed checks retain clearly labelled last +reported information. Older peers without this capability require a local check. An acknowledged request is consumed by the execution computer's running desktop and uses this same installer. Password-protected hosts are excluded even while unlocked; an independent host-side check runs at the final shutdown boundary. @@ -48,6 +53,48 @@ retain the original install operation rather than resubmitting it; verify the running identity and its receipt after both computers reconnect on wire 7. Updating the controller first requires updating the execution computer directly. A remote mismatch does not disable the controller's local update controls. +Definite install refusals are shown as declined requests and require a new check +before another confirmation. A potentially sent request with an unknown outcome +keeps its original operation identity; refreshing never resends installation. + +### Remote upgrade qualification + +Changes to pairing, remote update controls or the installer require a composed +upgrade test in addition to package verification. Package consumers establish +provenance, signatures and package startup; they do not establish a paired +remote upgrade. + +The normal Agent test suite includes +`cpace_confirm_restart_and_mutual_pairing_feed_the_real_install_grant_policy`. +It exchanges real CPace and ConfirmGrant frames between two loopback endpoints, +reopens both durable stores, and checks the production install-grant policy. +It covers lost pairing acknowledgement, independent pairing directions and +revocation. Do not replace its pairing-produced records with pre-authorized +fixtures. Enrollment, account sessions and signed installation are separate +boundaries. + +Before claiming native remote-upgrade coverage, record these observations from +isolated installations and a disposable account: + +1. Pair through the normal application path, restart both applications, and + request the execution computer's exact verified update from its controller. + Observe admission, the durable receipt, shutdown, replacement, relaunch and + the new running version/source. The controller must remain unchanged. +2. Change the signed feed after observing Current. Explicit refresh must produce + a new HTTP check on the execution computer and report that result. A cached + heartbeat, failed check, disconnected desktop or unsupported old peer cannot + be reported as a successful fresh check. +3. Exercise a definite pre-admission refusal and a lost acknowledgement after + sending. The latter must retain its operation ID without replaying the + installation. Include re-pairing or disconnect while its reply is pending. +4. For a wire-major transition, exercise the supported remote-first order with + the previous published application. Record the expected incompatibility if + the controller is updated first and verify the direct local recovery path. + +Preserve failures and identify the exact source, binary hashes, profiles, +operating systems and signing identities used. A Local fixture signed by a test +key can establish that fixture's lifecycle; it does not establish production +Developer ID, notarization, production account routing or public publication. The installer holds the per-application lock while staging a verified bundle beside the installed application and while that update is ready. It checks the