diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index 1e7f540..37c7bd0 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -96,6 +96,8 @@ pub struct TargetedTransfer { pub manifest_id: String, pub file_count: u64, pub total_size: u64, + /// Bytes verified so far; survives interruption for resume. + pub verified_bytes: u64, pub state: TargetedTransferState, pub created_at: i64, pub updated_at: i64, diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index 77c29a7..4d169c5 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -543,6 +543,39 @@ impl VnidropCore { self.block_on(self.inner.list_targeted_transfers()) } + /// Withdraw an offer or revoke an approved transfer. + /// + /// Stops active streaming synchronously before asynchronous cleanup. + pub fn cancel_targeted_transfer(&self, id: String) -> Result<(), VnidropError> { + if let Ok(Some(row)) = self.block_on(self.inner.targeted_store().get_row(&id)) { + let _ = self + .inner + .signal_targeted_transfer_cancel(row.protocol_transfer_id); + } + self.block_on(self.inner.cancel_targeted_transfer(id)) + } + + /// Durably remove authorization, resumable state, and content service. + /// + /// Local denial is mandatory even when remote cleanup fails. + pub fn delete_targeted_transfer(&self, id: String) -> Result<(), VnidropError> { + if let Ok(Some(row)) = self.block_on(self.inner.targeted_store().get_row(&id)) { + let _ = self + .inner + .signal_targeted_transfer_cancel(row.protocol_transfer_id); + } + self.block_on(self.inner.delete_targeted_transfer(id)) + } + + /// Resume an approved/interrupted transfer without another approval. + pub fn resume_targeted_transfer( + &self, + id: String, + output_dir: String, + ) -> Result<(), VnidropError> { + self.block_on(self.inner.resume_targeted_transfer(id, output_dir)) + } + /// Devices the user has chosen to remember. pub fn list_contacts(&self) -> Result, VnidropError> { self.block_on(self.inner.list_contacts()) diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index 21a590d..e9ef6b9 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -106,6 +106,7 @@ pub(super) struct CoreInner { pub(super) store: FsStore, pub(super) repository: Repository, _profile_lock: Option, + pub(super) secret_custody: Option>, pub(super) event_hub: Arc, pub(super) approval: ApprovalService, pub(super) pairing: PairingService, @@ -420,6 +421,7 @@ impl CoreInner { TargetedTransferProtocol::new( device_relationships.clone(), targeted_offers.clone(), + repository.sqlite_pool(), limits.clone(), endpoint.id().to_string(), ), @@ -433,6 +435,7 @@ impl CoreInner { store, repository, _profile_lock: profile_lock, + secret_custody: secret_custody.clone(), event_hub, approval, pairing, @@ -458,6 +461,15 @@ impl CoreInner { targeted_cancel_log: std::sync::Mutex::new(Vec::new()), }); + // In-flight connecting/transferring transfers become Interrupted across restart. + if let Err(error) = inner.targeted_store().mark_interrupted_in_flight().await { + tracing::warn!(%error, "failed to mark in-flight targeted transfers interrupted"); + } + // Restore permanent receiver ACLs for approved targeted shares. + if let Err(error) = inner.restore_targeted_transfer_access().await { + tracing::warn!(%error, "failed to restore targeted transfer access"); + } + inner.emit_endpoint( "startup", "endpoint-online", diff --git a/crates/vnidrop/src/runtime/targeted.rs b/crates/vnidrop/src/runtime/targeted.rs index e2ebe86..7b736b6 100644 --- a/crates/vnidrop/src/runtime/targeted.rs +++ b/crates/vnidrop/src/runtime/targeted.rs @@ -1,4 +1,4 @@ -//! Create, offer, approve, and receive targeted transfers between Saved devices. +//! Create, offer, approve, resume, cancel, and delete targeted transfers. use std::sync::Arc; @@ -13,13 +13,15 @@ use crate::{ ShareSource, TargetedTransfer, TargetedTransferState, TransferAccessMode, TransferMetadata, }, error::VnidropError, + secure_secret::{SecretHandle, SecretKind}, targeted_transfer::{ + auth_secret_material, protocol::{ - DeliverTargetedAuthorization, SubmitTargetedOffer, TargetedOfferResponse, - TargetedTransferProtocol, + CancelTargetedOffer, DeliverTargetedAuthorization, SubmitTargetedOffer, + TargetedOfferResponse, TargetedTransferProtocol, }, - TargetedAuthorization, TargetedAuthorizationDraft, TargetedTransferRow, - TargetedTransferStore, + reconstruct_authorization, TargetedAuthorization, TargetedAuthorizationDraft, + TargetedTransferRole, TargetedTransferRow, TargetedTransferStore, }, ticket::VnidropTicket, util::{non_empty, now_ms}, @@ -49,6 +51,15 @@ impl CoreInner { self.targeted_store().list().await } + pub(crate) async fn restore_targeted_transfer_access(&self) -> Result<(), VnidropError> { + for row in self.targeted_store().list_resumable_sender_rows().await? { + self.access_policy + .approve_endpoint_until(row.protocol_transfer_id, row.receiver_endpoint_id, None) + .await; + } + Ok(()) + } + /// Cancel in-flight targeted transfers involving `peer` (for forget/block). pub(crate) async fn cancel_targeted_transfers_for_peer( &self, @@ -76,13 +87,128 @@ impl CoreInner { self.targeted_store().cancel_by_peer(peer_endpoint_id).await } + /// Synchronously stop streaming for one transfer (facade calls this first). + pub(super) fn signal_targeted_transfer_cancel(&self, protocol_transfer_id: u64) -> bool { + self.take_active_transfer(protocol_transfer_id).is_some() + } + + pub(super) async fn cancel_targeted_transfer(&self, id: String) -> Result<(), VnidropError> { + let store = self.targeted_store(); + let Some(row) = store.get_row(&id).await? else { + // Still drop any live-session offer under this id. + self.targeted_offers.discard(&id).await; + return Ok(()); + }; + let _ = self.take_active_transfer(row.protocol_transfer_id); + self.targeted_offers.discard(&id).await; + self.access_policy + .remove_transfer(row.protocol_transfer_id) + .await; + let _ = self.cancel_idle_or_share(row.protocol_transfer_id).await; + if !matches!( + row.state, + TargetedTransferState::Completed + | TargetedTransferState::Declined + | TargetedTransferState::Cancelled + | TargetedTransferState::Failed + | TargetedTransferState::Deleted + ) { + let _ = store + .set_state_from_any(&id, TargetedTransferState::Cancelled) + .await; + } + // Best-effort remote withdraw of an unapproved live offer. + if row.role == TargetedTransferRole::Sender + && matches!( + row.state, + TargetedTransferState::Offering | TargetedTransferState::AwaitingApproval + ) + { + if let Ok(addr) = self + .device_relationships + .peer_addr(&row.receiver_endpoint_id) + .await + { + let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr); + let _ = tokio::time::timeout( + OFFER_CONNECT_TIMEOUT, + client.cancel_offer(CancelTargetedOffer { + transfer_id: id.clone(), + }), + ) + .await; + } + } + Ok(()) + } + + pub(super) async fn delete_targeted_transfer( + self: &Arc, + id: String, + ) -> Result<(), VnidropError> { + let store = self.targeted_store(); + let Some(row) = store.get_row(&id).await? else { + self.targeted_offers.discard(&id).await; + return Ok(()); + }; + // Durable local denial first — remote cleanup is best-effort. + let _ = self.take_active_transfer(row.protocol_transfer_id); + self.targeted_offers.discard(&id).await; + self.access_policy + .remove_transfer(row.protocol_transfer_id) + .await; + let _ = self.cancel_idle_or_share(row.protocol_transfer_id).await; + if let Some(handle) = &row.authorization_secret_handle { + if let Some(custody) = &self.secret_custody { + let _ = custody + .remove(&SecretHandle::from_stored(handle.clone())) + .await; + } + } + store.clear_authorization(&id).await?; + if row.state != TargetedTransferState::Deleted { + if !matches!( + row.state, + TargetedTransferState::Completed + | TargetedTransferState::Declined + | TargetedTransferState::Cancelled + | TargetedTransferState::Failed + ) { + let _ = store + .set_state_from_any(&id, TargetedTransferState::Cancelled) + .await; + } + let _ = store + .set_state_from_any(&id, TargetedTransferState::Deleted) + .await; + } + Ok(()) + } + pub(super) async fn respond_to_targeted_offer( - &self, + self: &Arc, transfer_id: String, accepted: bool, ) -> Result, VnidropError> { + if let Some(auth) = self + .targeted_offers + .settled_authorization(&transfer_id) + .await + { + return Ok(Some(auth)); + } + if let Ok(Some(row)) = self.targeted_store().get_row(&transfer_id).await { + if let Some(encoded) = self.load_stored_authorization(&row).await? { + return Ok(Some(encoded)); + } + } + match self.targeted_offers.respond(&transfer_id, accepted).await { - Ok(auth) => Ok(auth), + Ok(Some(auth)) => { + self.persist_receiver_authorization(&auth).await?; + Ok(Some(auth)) + } + Ok(None) => Ok(None), Err(crate::targeted_transfer::RespondError::Unknown) => Err( VnidropError::invalid_input(anyhow::anyhow!("unknown targeted offer")), ), @@ -134,6 +260,10 @@ impl CoreInner { transfer_name: share.transfer_name.clone(), file_count: share.file_count, total_size: share.total_size, + verified_bytes: 0, + blob_ticket: None, + authorization_secret_handle: None, + role: TargetedTransferRole::Sender, state: TargetedTransferState::Preparing, created_at: now, updated_at: now, @@ -225,9 +355,9 @@ impl CoreInner { } } - // Bound authorization: only the approved receiver endpoint may fetch. + // Permanent until cancel/delete — approved targeted transfers must resume. self.access_policy - .approve_endpoint(protocol_transfer_id, receiver_endpoint_id.clone()) + .approve_endpoint_until(protocol_transfer_id, receiver_endpoint_id.clone(), None) .await; let parsed = crate::ticket::parse_transfer_ticket_with_limits(&share.ticket, &self.limits) @@ -250,6 +380,8 @@ impl CoreInner { transfer_name: share.transfer_name.clone(), blob_ticket: blob_ticket.to_string(), })?; + self.persist_authorization_secret(&transfer_uuid, &authorization) + .await?; let encoded = authorization.encode()?; let deliver = client @@ -294,6 +426,86 @@ impl CoreInner { ) -> Result<(), VnidropError> { let auth = TargetedAuthorization::decode(&authorization)?; auth.verify_for_receiver(&self.endpoint.id().to_string())?; + self.run_targeted_receive(&auth, output_dir).await + } + + pub(super) async fn resume_targeted_transfer( + self: &Arc, + id: String, + output_dir: String, + ) -> Result<(), VnidropError> { + let store = self.targeted_store(); + let row = store.get_row(&id).await?.ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("unknown targeted transfer")) + })?; + if !matches!( + row.state, + TargetedTransferState::Approved + | TargetedTransferState::Connecting + | TargetedTransferState::Transferring + | TargetedTransferState::Interrupted + ) { + return Err(VnidropError::InvalidTransition { + reason: format!( + "cannot resume from {}", + crate::targeted_transfer::state_as_str(row.state) + ), + }); + } + let encoded = self.load_stored_authorization(&row).await?.ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!( + "targeted transfer has no durable authorization" + )) + })?; + let auth = TargetedAuthorization::decode(&encoded)?; + auth.verify_for_receiver(&self.endpoint.id().to_string())?; + self.run_targeted_receive(&auth, output_dir).await + } + + async fn run_targeted_receive( + self: &Arc, + auth: &TargetedAuthorization, + output_dir: String, + ) -> Result<(), VnidropError> { + let store = self.targeted_store(); + if let Ok(Some(row)) = store.get_row(&auth.transfer_id).await { + match row.state { + TargetedTransferState::Approved | TargetedTransferState::Interrupted => { + store + .set_state( + &auth.transfer_id, + row.state, + TargetedTransferState::Connecting, + ) + .await?; + store + .set_state( + &auth.transfer_id, + TargetedTransferState::Connecting, + TargetedTransferState::Transferring, + ) + .await?; + } + TargetedTransferState::Connecting => { + store + .set_state( + &auth.transfer_id, + TargetedTransferState::Connecting, + TargetedTransferState::Transferring, + ) + .await?; + } + TargetedTransferState::Transferring => {} + other => { + return Err(VnidropError::InvalidTransition { + reason: format!( + "cannot receive from {}", + crate::targeted_transfer::state_as_str(other) + ), + }); + } + } + } let blob_ticket = BlobTicket::from_str_compat(&auth.blob_ticket) .map_err(|error| VnidropError::ticket(anyhow::anyhow!(error)))?; @@ -310,21 +522,132 @@ impl CoreInner { .encode() .map_err(VnidropError::ticket)?; - self.receive(ticket, std::path::PathBuf::from(output_dir), None) - .await - .map_err(VnidropError::transfer)?; + let receive_result = self + .receive(ticket, std::path::PathBuf::from(output_dir), None) + .await; - if let Ok(Some(row)) = self.targeted_store().get_row(&auth.transfer_id).await { - let _ = self - .targeted_store() - .set_state( - &auth.transfer_id, - row.state, - TargetedTransferState::Completed, - ) - .await; + match receive_result { + Ok(()) => { + if let Ok(Some(row)) = store.get_row(&auth.transfer_id).await { + let _ = store + .set_verified_bytes(&auth.transfer_id, row.total_size) + .await; + let _ = store + .set_state( + &auth.transfer_id, + TargetedTransferState::Transferring, + TargetedTransferState::Completed, + ) + .await; + } + Ok(()) + } + Err(error) => { + if let Ok(Some(row)) = store.get_row(&auth.transfer_id).await { + if matches!( + row.state, + TargetedTransferState::Connecting | TargetedTransferState::Transferring + ) { + let _ = store + .set_state_from_any( + &auth.transfer_id, + TargetedTransferState::Interrupted, + ) + .await; + } + } + Err(VnidropError::transfer(error)) + } } - Ok(()) + } + + async fn persist_authorization_secret( + &self, + transfer_id: &str, + authorization: &TargetedAuthorization, + ) -> Result<(), VnidropError> { + let custody = + self.secret_custody + .as_ref() + .ok_or_else(|| VnidropError::SecureStorageUnavailable { + reason: "targeted authorization requires protected custody".to_string(), + })?; + let material = auth_secret_material(authorization)?; + let handle = custody + .protect(SecretKind::TargetedAuthorization, material, None) + .await?; + self.targeted_store() + .store_authorization(transfer_id, &authorization.blob_ticket, handle.as_str()) + .await + } + + async fn persist_receiver_authorization(&self, encoded: &str) -> Result<(), VnidropError> { + let auth = TargetedAuthorization::decode(encoded)?; + let store = self.targeted_store(); + if store.get_row(&auth.transfer_id).await?.is_none() { + let now = now_ms(); + store + .insert(&TargetedTransferRow { + id: auth.transfer_id.clone(), + protocol_transfer_id: auth.protocol_transfer_id, + sender_endpoint_id: auth.sender_endpoint_id.clone(), + receiver_endpoint_id: auth.receiver_endpoint_id.clone(), + manifest_id: auth.manifest_id.clone(), + content_hash: auth.content_hash.clone(), + transfer_name: auth.transfer_name.clone(), + file_count: auth.file_count, + total_size: auth.total_size, + verified_bytes: 0, + blob_ticket: Some(auth.blob_ticket.clone()), + authorization_secret_handle: None, + role: TargetedTransferRole::Receiver, + state: TargetedTransferState::Approved, + created_at: now, + updated_at: now, + }) + .await?; + } + self.persist_authorization_secret(&auth.transfer_id, &auth) + .await + } + + async fn load_stored_authorization( + &self, + row: &TargetedTransferRow, + ) -> Result, VnidropError> { + let (Some(handle), Some(blob_ticket)) = ( + row.authorization_secret_handle.as_ref(), + row.blob_ticket.as_ref(), + ) else { + return Ok(None); + }; + let custody = + self.secret_custody + .as_ref() + .ok_or_else(|| VnidropError::SecureStorageUnavailable { + reason: "targeted authorization requires protected custody".to_string(), + })?; + let material = custody + .load(&SecretHandle::from_stored(handle.clone())) + .await?; + let auth = reconstruct_authorization( + TargetedAuthorizationDraft { + transfer_id: row.id.clone(), + protocol_transfer_id: row.protocol_transfer_id, + sender_endpoint_id: row.sender_endpoint_id.clone(), + receiver_endpoint_id: row.receiver_endpoint_id.clone(), + manifest_id: row.manifest_id.clone(), + content_hash: row.content_hash.clone(), + file_count: row.file_count, + total_size: row.total_size, + protocol_version: experimental_saved_device_capabilities() + .targeted_transfer_protocol_version, + transfer_name: row.transfer_name.clone(), + blob_ticket: blob_ticket.clone(), + }, + &material, + )?; + Ok(Some(auth.encode()?)) } } diff --git a/crates/vnidrop/src/secure_secret.rs b/crates/vnidrop/src/secure_secret.rs index 9ba1db0..bb4049a 100644 --- a/crates/vnidrop/src/secure_secret.rs +++ b/crates/vnidrop/src/secure_secret.rs @@ -108,6 +108,7 @@ pub(crate) enum SecretKind { EndpointIdentity, RelationshipGrant, PairingEligibility, + TargetedAuthorization, } impl SecretKind { @@ -116,6 +117,7 @@ impl SecretKind { Self::EndpointIdentity => "endpoint-identity", Self::RelationshipGrant => "relationship-grant", Self::PairingEligibility => "pairing-eligibility", + Self::TargetedAuthorization => "targeted-authorization", } } @@ -124,6 +126,7 @@ impl SecretKind { "endpoint-identity" => Ok(Self::EndpointIdentity), "relationship-grant" => Ok(Self::RelationshipGrant), "pairing-eligibility" => Ok(Self::PairingEligibility), + "targeted-authorization" => Ok(Self::TargetedAuthorization), _ => Err(VnidropError::SecureStorageCorrupted { reason: "protected secret has an unknown kind".to_string(), }), diff --git a/crates/vnidrop/src/targeted_transfer/auth.rs b/crates/vnidrop/src/targeted_transfer/auth.rs index 475ce62..8ff8a72 100644 --- a/crates/vnidrop/src/targeted_transfer/auth.rs +++ b/crates/vnidrop/src/targeted_transfer/auth.rs @@ -9,7 +9,7 @@ use data_encoding::{BASE64URL_NOPAD, HEXLOWER}; use serde::{Deserialize, Serialize}; -use crate::error::VnidropError; +use crate::{error::VnidropError, secure_secret::SecretMaterial}; const AUTH_CONTEXT: &[u8] = b"vnidrop-targeted-auth-v1"; @@ -137,4 +137,44 @@ impl TargetedAuthorization { hasher.update(&self.protocol_version.to_le_bytes()); Ok(*hasher.finalize().as_bytes()) } + + pub(crate) fn secret_bytes(&self) -> Result<[u8; 32], VnidropError> { + let secret_bytes = HEXLOWER.decode(self.auth_secret.as_bytes()).map_err(|_| { + VnidropError::invalid_input(anyhow::anyhow!("invalid authorization secret")) + })?; + secret_bytes.try_into().map_err(|_| { + VnidropError::invalid_input(anyhow::anyhow!("invalid authorization secret length")) + }) + } +} + +pub(crate) fn auth_secret_material( + auth: &TargetedAuthorization, +) -> Result { + SecretMaterial::new(auth.secret_bytes()?.to_vec()) +} + +/// Rebuild a bound authorization from durable row fields + custody secret. +pub(crate) fn reconstruct_authorization( + draft: TargetedAuthorizationDraft, + secret_material: &SecretMaterial, +) -> Result { + let auth_secret = HEXLOWER.encode(secret_material.as_bytes()); + let mut auth = TargetedAuthorization { + transfer_id: draft.transfer_id, + protocol_transfer_id: draft.protocol_transfer_id, + sender_endpoint_id: draft.sender_endpoint_id, + receiver_endpoint_id: draft.receiver_endpoint_id, + manifest_id: draft.manifest_id, + content_hash: draft.content_hash, + file_count: draft.file_count, + total_size: draft.total_size, + protocol_version: draft.protocol_version, + transfer_name: draft.transfer_name, + blob_ticket: draft.blob_ticket, + auth_secret, + mac: String::new(), + }; + auth.mac = HEXLOWER.encode(&auth.compute_mac()?); + Ok(auth) } diff --git a/crates/vnidrop/src/targeted_transfer/inbox.rs b/crates/vnidrop/src/targeted_transfer/inbox.rs index 2b4ab57..ba6b060 100644 --- a/crates/vnidrop/src/targeted_transfer/inbox.rs +++ b/crates/vnidrop/src/targeted_transfer/inbox.rs @@ -2,11 +2,12 @@ //! //! Offers are not durable: cancellation, timeout, disconnect, or restart drops //! them. Authorization is delivered only after the local user accepts. +//! Settled results are cached briefly so lost-response replays stay idempotent. use std::{collections::HashMap, sync::Arc, time::Duration}; use serde_json::json; -use tokio::sync::{oneshot, Mutex}; +use tokio::sync::{watch, Mutex}; use uuid::Uuid; use crate::{api::PendingTargetedOffer, event_hub::EventHub}; @@ -18,20 +19,27 @@ pub(crate) struct PendingTargetedOfferRecord { pub(crate) offer: PendingTargetedOffer, } -struct DecisionWaiter { - decision: oneshot::Sender, +struct PendingWaiter { + decision: watch::Sender>, } struct AuthWaiter { - auth: oneshot::Sender, + auth: watch::Sender>, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum SettledOfferResult { + Accepted { authorization: Option }, + Declined { reason: String }, } #[derive(Clone)] pub(crate) struct TargetedOfferInbox { event_hub: Arc, pending: Arc>>, - decisions: Arc>>, + decisions: Arc>>, auths: Arc>>, + settled: Arc>>, max_pending: usize, } @@ -42,16 +50,32 @@ impl TargetedOfferInbox { pending: Arc::new(Mutex::new(HashMap::new())), decisions: Arc::new(Mutex::new(HashMap::new())), auths: Arc::new(Mutex::new(HashMap::new())), + settled: Arc::new(Mutex::new(HashMap::new())), max_pending, } } /// Surface a validated offer and block until the local user decides. + /// + /// Replaying the same transfer identity returns the settled result or joins + /// the existing pending wait — never a second prompt. pub(crate) async fn submit(&self, offer: PendingTargetedOffer) -> TargetedOfferDecision { let transfer_id = offer.transfer_id.clone(); - let (decision_tx, decision_rx) = oneshot::channel(); + if let Some(settled) = self.settled.lock().await.get(&transfer_id).cloned() { + return settled_to_decision(settled); + } + { - let mut pending = self.pending.lock().await; + let pending = self.pending.lock().await; + if let Some(existing) = pending.get(&transfer_id) { + if offers_equivalent(&existing.offer, &offer) { + drop(pending); + return self.wait_existing_decision(&transfer_id).await; + } + return TargetedOfferDecision::Refused { + reason: "immutable-transfer-mismatch".to_string(), + }; + } if pending.len() >= self.max_pending { return TargetedOfferDecision::Refused { reason: "too-many-pending-offers".to_string(), @@ -65,10 +89,15 @@ impl TargetedOfferInbox { reason: "offer-already-pending".to_string(), }; } + } + + let (decision_tx, _decision_rx) = watch::channel(None); + { + let mut pending = self.pending.lock().await; if pending.contains_key(&transfer_id) { - return TargetedOfferDecision::Refused { - reason: "duplicate-transfer".to_string(), - }; + // Lost the race with another submit of the same id. + drop(pending); + return self.wait_existing_decision(&transfer_id).await; } pending.insert( transfer_id.clone(), @@ -79,7 +108,7 @@ impl TargetedOfferInbox { } self.decisions.lock().await.insert( transfer_id.clone(), - DecisionWaiter { + PendingWaiter { decision: decision_tx, }, ); @@ -96,16 +125,44 @@ impl TargetedOfferInbox { }), ); - match tokio::time::timeout(OFFER_WAIT_TIMEOUT, decision_rx).await { - Ok(Ok(true)) => TargetedOfferDecision::Accepted, - Ok(Ok(false)) => { - self.discard(&transfer_id).await; + self.wait_existing_decision(&transfer_id).await + } + + async fn wait_existing_decision(&self, transfer_id: &str) -> TargetedOfferDecision { + let mut rx = { + let decisions = self.decisions.lock().await; + let Some(waiter) = decisions.get(transfer_id) else { + if let Some(settled) = self.settled.lock().await.get(transfer_id).cloned() { + return settled_to_decision(settled); + } + return TargetedOfferDecision::Declined { + reason: "no-response".to_string(), + }; + }; + waiter.decision.subscribe() + }; + + let wait = async { + loop { + if let Some(accepted) = *rx.borrow_and_update() { + return accepted; + } + if rx.changed().await.is_err() { + return false; + } + } + }; + + match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait).await { + Ok(true) => TargetedOfferDecision::Accepted, + Ok(false) => { + self.discard(transfer_id).await; TargetedOfferDecision::Declined { reason: "receiver-declined".to_string(), } } - Ok(Err(_)) | Err(_) => { - self.discard(&transfer_id).await; + Err(_) => { + self.discard(transfer_id).await; TargetedOfferDecision::Declined { reason: "no-response".to_string(), } @@ -122,23 +179,55 @@ impl TargetedOfferInbox { .collect() } + pub(crate) async fn get_pending(&self, transfer_id: &str) -> Option { + self.pending + .lock() + .await + .get(transfer_id) + .map(|entry| entry.offer.clone()) + } + + pub(crate) async fn settled_authorization(&self, transfer_id: &str) -> Option { + match self.settled.lock().await.get(transfer_id) { + Some(SettledOfferResult::Accepted { + authorization: Some(auth), + }) => Some(auth.clone()), + _ => None, + } + } + /// Record the local decision. On accept, wait for sender-issued authorization. pub(crate) async fn respond( &self, transfer_id: &str, accepted: bool, ) -> Result, RespondError> { + if let Some(auth) = self.settled_authorization(transfer_id).await { + return Ok(Some(auth)); + } + let exists = self.pending.lock().await.contains_key(transfer_id); if !exists { return Err(RespondError::Unknown); } - let waiter = self.decisions.lock().await.remove(transfer_id); - let Some(waiter) = waiter else { + let waiter = { + let decisions = self.decisions.lock().await; + decisions + .get(transfer_id) + .map(|entry| entry.decision.clone()) + }; + let Some(decision_tx) = waiter else { return Err(RespondError::Unknown); }; if !accepted { - let _ = waiter.decision.send(false); + let _ = decision_tx.send(Some(false)); self.discard(transfer_id).await; + self.settled.lock().await.insert( + transfer_id.to_string(), + SettledOfferResult::Declined { + reason: "receiver-declined".to_string(), + }, + ); self.event_hub.emit_endpoint( "targeted_transfer", "offer-declined", @@ -147,20 +236,38 @@ impl TargetedOfferInbox { return Ok(None); } - let (auth_tx, auth_rx) = oneshot::channel(); + let (auth_tx, mut auth_rx) = watch::channel(None); self.auths .lock() .await .insert(transfer_id.to_string(), AuthWaiter { auth: auth_tx }); - if waiter.decision.send(true).is_err() { + if decision_tx.send(Some(true)).is_err() { self.auths.lock().await.remove(transfer_id); self.discard(transfer_id).await; return Err(RespondError::SenderGone); } - match tokio::time::timeout(OFFER_WAIT_TIMEOUT, auth_rx).await { + let wait_auth = async { + loop { + if let Some(auth) = auth_rx.borrow_and_update().clone() { + return Ok(auth); + } + if auth_rx.changed().await.is_err() { + return Err(()); + } + } + }; + + match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait_auth).await { Ok(Ok(auth)) => { self.pending.lock().await.remove(transfer_id); + self.auths.lock().await.remove(transfer_id); + self.settled.lock().await.insert( + transfer_id.to_string(), + SettledOfferResult::Accepted { + authorization: Some(auth.clone()), + }, + ); self.event_hub.emit_endpoint( "targeted_transfer", "offer-accepted", @@ -168,7 +275,7 @@ impl TargetedOfferInbox { ); Ok(Some(auth)) } - Ok(Err(_)) | Err(_) => { + Ok(Err(())) | Err(_) => { self.discard(transfer_id).await; Err(RespondError::AuthorizationTimeout) } @@ -180,8 +287,14 @@ impl TargetedOfferInbox { transfer_id: &str, authorization: String, ) -> bool { - if let Some(waiter) = self.auths.lock().await.remove(transfer_id) { - waiter.auth.send(authorization).is_ok() + if let Some(SettledOfferResult::Accepted { + authorization: Some(existing), + }) = self.settled.lock().await.get(transfer_id) + { + return existing == &authorization; + } + if let Some(waiter) = self.auths.lock().await.get(transfer_id) { + waiter.auth.send(Some(authorization)).is_ok() } else { false } @@ -201,15 +314,33 @@ impl TargetedOfferInbox { } } - async fn discard(&self, transfer_id: &str) { + pub(crate) async fn discard(&self, transfer_id: &str) { self.pending.lock().await.remove(transfer_id); if let Some(waiter) = self.decisions.lock().await.remove(transfer_id) { - let _ = waiter.decision.send(false); + let _ = waiter.decision.send(Some(false)); } self.auths.lock().await.remove(transfer_id); } } +fn offers_equivalent(left: &PendingTargetedOffer, right: &PendingTargetedOffer) -> bool { + left.transfer_id == right.transfer_id + && left.sender_endpoint_id == right.sender_endpoint_id + && left.receiver_endpoint_id == right.receiver_endpoint_id + && left.manifest_id == right.manifest_id + && left.content_hash == right.content_hash + && left.file_count == right.file_count + && left.total_size == right.total_size + && left.protocol_version == right.protocol_version +} + +fn settled_to_decision(settled: SettledOfferResult) -> TargetedOfferDecision { + match settled { + SettledOfferResult::Accepted { .. } => TargetedOfferDecision::Accepted, + SettledOfferResult::Declined { reason } => TargetedOfferDecision::Declined { reason }, + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum TargetedOfferDecision { Accepted, diff --git a/crates/vnidrop/src/targeted_transfer/mod.rs b/crates/vnidrop/src/targeted_transfer/mod.rs index 7732447..64bb4ae 100644 --- a/crates/vnidrop/src/targeted_transfer/mod.rs +++ b/crates/vnidrop/src/targeted_transfer/mod.rs @@ -8,7 +8,10 @@ pub(crate) mod inbox; pub(crate) mod protocol; mod state; -pub(crate) use auth::{TargetedAuthorization, TargetedAuthorizationDraft}; +pub(crate) use auth::{ + auth_secret_material, reconstruct_authorization, TargetedAuthorization, + TargetedAuthorizationDraft, +}; pub(crate) use inbox::{RespondError, TargetedOfferInbox}; pub(crate) use protocol::TargetedTransferProtocol; @@ -33,6 +36,10 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> { transfer_name TEXT NOT NULL, file_count INTEGER NOT NULL, total_size INTEGER NOT NULL, + verified_bytes INTEGER NOT NULL DEFAULT 0, + blob_ticket TEXT, + authorization_secret_handle TEXT, + role TEXT NOT NULL DEFAULT 'sender', state TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL @@ -41,9 +48,38 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> { ) .execute(pool) .await?; + let columns = sqlx::query("PRAGMA table_info(targeted_transfers)") + .fetch_all(pool) + .await?; + let has = |name: &str| columns.iter().any(|row| row.get::(1) == name); + if !has("verified_bytes") { + sqlx::query( + "ALTER TABLE targeted_transfers ADD COLUMN verified_bytes INTEGER NOT NULL DEFAULT 0", + ) + .execute(pool) + .await?; + } + if !has("blob_ticket") { + sqlx::query("ALTER TABLE targeted_transfers ADD COLUMN blob_ticket TEXT") + .execute(pool) + .await?; + } + if !has("authorization_secret_handle") { + sqlx::query("ALTER TABLE targeted_transfers ADD COLUMN authorization_secret_handle TEXT") + .execute(pool) + .await?; + } + if !has("role") { + sqlx::query( + "ALTER TABLE targeted_transfers ADD COLUMN role TEXT NOT NULL DEFAULT 'sender'", + ) + .execute(pool) + .await?; + } Ok(()) } +#[derive(Clone)] pub(crate) struct TargetedTransferStore { pool: SqlitePool, } @@ -59,8 +95,9 @@ impl TargetedTransferStore { INSERT INTO targeted_transfers ( id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id, manifest_id, content_hash, transfer_name, file_count, total_size, + verified_bytes, blob_ticket, authorization_secret_handle, role, state, created_at, updated_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16) "#, ) .bind(&transfer.id) @@ -72,6 +109,10 @@ impl TargetedTransferStore { .bind(&transfer.transfer_name) .bind(transfer.file_count as i64) .bind(transfer.total_size as i64) + .bind(transfer.verified_bytes as i64) + .bind(&transfer.blob_ticket) + .bind(&transfer.authorization_secret_handle) + .bind(role_as_str(transfer.role)) .bind(state_as_str(transfer.state)) .bind(transfer.created_at) .bind(transfer.updated_at) @@ -110,11 +151,113 @@ impl TargetedTransferStore { Ok(()) } + /// Transition from any non-terminal state; used by cancel/delete. + pub(crate) async fn set_state_from_any( + &self, + id: &str, + to: TargetedTransferState, + ) -> Result<(), VnidropError> { + let Some(row) = self.get_row(id).await? else { + return Err(VnidropError::invalid_input(anyhow::anyhow!( + "unknown targeted transfer" + ))); + }; + if row.state == to { + return Ok(()); + } + row.state.validate_transition_to(to)?; + let result = sqlx::query( + r#" + UPDATE targeted_transfers + SET state = ?2, updated_at = ?3 + WHERE id = ?1 AND state = ?4 + "#, + ) + .bind(id) + .bind(state_as_str(to)) + .bind(now_ms()) + .bind(state_as_str(row.state)) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + if result.rows_affected() == 0 { + return Err(VnidropError::InvalidTransition { + reason: format!("{} -> {}", state_as_str(row.state), state_as_str(to)), + }); + } + Ok(()) + } + + pub(crate) async fn set_verified_bytes( + &self, + id: &str, + verified_bytes: u64, + ) -> Result<(), VnidropError> { + sqlx::query( + r#" + UPDATE targeted_transfers + SET verified_bytes = ?2, updated_at = ?3 + WHERE id = ?1 + "#, + ) + .bind(id) + .bind(verified_bytes as i64) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + pub(crate) async fn store_authorization( + &self, + id: &str, + blob_ticket: &str, + authorization_secret_handle: &str, + ) -> Result<(), VnidropError> { + sqlx::query( + r#" + UPDATE targeted_transfers + SET blob_ticket = ?2, + authorization_secret_handle = ?3, + updated_at = ?4 + WHERE id = ?1 + "#, + ) + .bind(id) + .bind(blob_ticket) + .bind(authorization_secret_handle) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + pub(crate) async fn clear_authorization(&self, id: &str) -> Result<(), VnidropError> { + sqlx::query( + r#" + UPDATE targeted_transfers + SET blob_ticket = NULL, + authorization_secret_handle = NULL, + verified_bytes = 0, + updated_at = ?2 + WHERE id = ?1 + "#, + ) + .bind(id) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + pub(crate) async fn get(&self, id: &str) -> Result, VnidropError> { let row = sqlx::query( r#" SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id, - file_count, total_size, state, created_at, updated_at + file_count, total_size, verified_bytes, state, created_at, updated_at FROM targeted_transfers WHERE id = ?1 "#, ) @@ -133,6 +276,7 @@ impl TargetedTransferStore { r#" SELECT id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id, manifest_id, content_hash, transfer_name, file_count, total_size, + verified_bytes, blob_ticket, authorization_secret_handle, role, state, created_at, updated_at FROM targeted_transfers WHERE id = ?1 "#, @@ -148,7 +292,7 @@ impl TargetedTransferStore { let rows = sqlx::query( r#" SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id, - file_count, total_size, state, created_at, updated_at + file_count, total_size, verified_bytes, state, created_at, updated_at FROM targeted_transfers ORDER BY updated_at DESC "#, @@ -159,6 +303,26 @@ impl TargetedTransferStore { rows.into_iter().map(row_to_transfer).collect() } + pub(crate) async fn list_resumable_sender_rows( + &self, + ) -> Result, VnidropError> { + let rows = sqlx::query( + r#" + SELECT id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id, + manifest_id, content_hash, transfer_name, file_count, total_size, + verified_bytes, blob_ticket, authorization_secret_handle, role, + state, created_at, updated_at + FROM targeted_transfers + WHERE role = 'sender' + AND state IN ('approved', 'connecting', 'transferring', 'interrupted') + "#, + ) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + rows.into_iter().map(row_to_full).collect() + } + pub(crate) async fn cancel_by_peer(&self, peer_endpoint_id: &str) -> Result { let now = now_ms(); let result = sqlx::query( @@ -196,6 +360,28 @@ impl TargetedTransferStore { .map(|row| row.get::(0) as u64) .collect()) } + + pub(crate) async fn mark_interrupted_in_flight(&self) -> Result { + let now = now_ms(); + let result = sqlx::query( + r#" + UPDATE targeted_transfers + SET state = 'interrupted', updated_at = ?1 + WHERE state IN ('connecting', 'transferring') + "#, + ) + .bind(now) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(result.rows_affected()) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TargetedTransferRole { + Sender, + Receiver, } #[derive(Debug, Clone)] @@ -209,6 +395,10 @@ pub(crate) struct TargetedTransferRow { pub(crate) transfer_name: String, pub(crate) file_count: u64, pub(crate) total_size: u64, + pub(crate) verified_bytes: u64, + pub(crate) blob_ticket: Option, + pub(crate) authorization_secret_handle: Option, + pub(crate) role: TargetedTransferRole, pub(crate) state: TargetedTransferState, pub(crate) created_at: i64, pub(crate) updated_at: i64, @@ -222,6 +412,7 @@ fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result("file_count") as u64, total_size: row.get::("total_size") as u64, + verified_bytes: row.try_get::("verified_bytes").unwrap_or(0) as u64, state: parse_state(&row.get::("state"))?, created_at: row.get("created_at"), updated_at: row.get("updated_at"), @@ -239,6 +430,13 @@ fn row_to_full(row: sqlx::sqlite::SqliteRow) -> Result("file_count") as u64, total_size: row.get::("total_size") as u64, + verified_bytes: row.try_get::("verified_bytes").unwrap_or(0) as u64, + blob_ticket: row.try_get("blob_ticket").ok().flatten(), + authorization_secret_handle: row.try_get("authorization_secret_handle").ok().flatten(), + role: parse_role( + &row.try_get::("role") + .unwrap_or_else(|_| "sender".to_string()), + )?, state: parse_state(&row.get::("state"))?, created_at: row.get("created_at"), updated_at: row.get("updated_at"), @@ -262,6 +460,23 @@ pub(crate) fn state_as_str(state: TargetedTransferState) -> &'static str { } } +fn role_as_str(role: TargetedTransferRole) -> &'static str { + match role { + TargetedTransferRole::Sender => "sender", + TargetedTransferRole::Receiver => "receiver", + } +} + +fn parse_role(value: &str) -> Result { + match value { + "sender" => Ok(TargetedTransferRole::Sender), + "receiver" => Ok(TargetedTransferRole::Receiver), + other => Err(VnidropError::repository(anyhow::anyhow!( + "unknown targeted transfer role: {other}" + ))), + } +} + fn parse_state(value: &str) -> Result { match value { "preparing" => Ok(TargetedTransferState::Preparing), diff --git a/crates/vnidrop/src/targeted_transfer/protocol.rs b/crates/vnidrop/src/targeted_transfer/protocol.rs index 3b09176..1a62739 100644 --- a/crates/vnidrop/src/targeted_transfer/protocol.rs +++ b/crates/vnidrop/src/targeted_transfer/protocol.rs @@ -14,13 +14,15 @@ use iroh::{ use irpc::{channel::oneshot, rpc_requests, Client, WithChannels}; use irpc_iroh::{read_request, IrohLazyRemoteConnection}; use serde::{Deserialize, Serialize}; +use sqlx::SqlitePool; use super::{ auth::TargetedAuthorization, inbox::{TargetedOfferDecision, TargetedOfferInbox}, + state_as_str, TargetedTransferStore, }; use crate::{ - api::{experimental_saved_device_capabilities, PendingTargetedOffer}, + api::{experimental_saved_device_capabilities, PendingTargetedOffer, TargetedTransferState}, device_relationship::{DeviceRelationshipService, WireProof}, error::VnidropError, grant::Challenge, @@ -31,6 +33,7 @@ use crate::{ pub(crate) struct TargetedTransferProtocol { relationships: std::sync::Arc, inbox: TargetedOfferInbox, + store: TargetedTransferStore, limits: crate::api::CoreLimits, local_endpoint_id: String, } @@ -47,12 +50,14 @@ impl TargetedTransferProtocol { pub(crate) fn new( relationships: std::sync::Arc, inbox: TargetedOfferInbox, + pool: SqlitePool, limits: crate::api::CoreLimits, local_endpoint_id: String, ) -> Self { Self { relationships, inbox, + store: TargetedTransferStore::new(pool), limits, local_endpoint_id, } @@ -111,6 +116,44 @@ impl TargetedTransferProtocol { }; } + if let Ok(Some(existing)) = self.store.get_row(&offer.transfer_id).await { + if existing.manifest_id != offer.manifest_id + || existing.content_hash != offer.content_hash + || existing.file_count != offer.file_count + || existing.total_size != offer.total_size + || existing.sender_endpoint_id != offer.sender_endpoint_id + || existing.receiver_endpoint_id != offer.receiver_endpoint_id + { + return TargetedOfferResponse::Refused { + reason: "immutable-transfer-mismatch".to_string(), + }; + } + return match existing.state { + TargetedTransferState::Approved + | TargetedTransferState::Connecting + | TargetedTransferState::Transferring + | TargetedTransferState::Interrupted + | TargetedTransferState::Completed => TargetedOfferResponse::Accepted, + TargetedTransferState::Declined => TargetedOfferResponse::Declined { + reason: "receiver-declined".to_string(), + }, + TargetedTransferState::Cancelled => TargetedOfferResponse::Declined { + reason: "cancelled".to_string(), + }, + TargetedTransferState::Failed | TargetedTransferState::Deleted => { + TargetedOfferResponse::Refused { + reason: format!("transfer-{}", state_as_str(existing.state)), + } + } + TargetedTransferState::Preparing + | TargetedTransferState::Offering + | TargetedTransferState::AwaitingApproval => { + // Live offer path below may still be pending. + TargetedOfferResponse::Accepted + } + }; + } + if let Err(error) = self .relationships .verify_saved_possession( @@ -164,6 +207,14 @@ impl TargetedTransferProtocol { { return DeliverAuthorizationResponse::Rejected; } + if let Ok(Some(row)) = self.store.get_row(&delivery.transfer_id).await { + if row.authorization_secret_handle.is_some() + && row.manifest_id == auth.manifest_id + && row.content_hash == auth.content_hash + { + return DeliverAuthorizationResponse::Stored; + } + } if self .inbox .deliver_authorization(&delivery.transfer_id, delivery.authorization) @@ -174,6 +225,28 @@ impl TargetedTransferProtocol { DeliverAuthorizationResponse::Rejected } } + + async fn handle_cancel( + &self, + remote_endpoint_id: &str, + cancel: CancelTargetedOffer, + ) -> CancelTargetedOfferResponse { + if let Some(pending) = self.inbox.get_pending(&cancel.transfer_id).await { + if pending.sender_endpoint_id != remote_endpoint_id { + return CancelTargetedOfferResponse::Rejected; + } + self.inbox.discard(&cancel.transfer_id).await; + return CancelTargetedOfferResponse::Cancelled; + } + if let Ok(Some(row)) = self.store.get_row(&cancel.transfer_id).await { + if row.sender_endpoint_id != remote_endpoint_id { + return CancelTargetedOfferResponse::Rejected; + } + // Already gone from the live inbox; treat as idempotent success. + return CancelTargetedOfferResponse::Cancelled; + } + CancelTargetedOfferResponse::Cancelled + } } impl ProtocolHandler for TargetedTransferProtocol { @@ -204,6 +277,11 @@ impl ProtocolHandler for TargetedTransferProtocol { .await; let _ = tx.send(response).await; } + TargetedTransferMessage::CancelTargetedOffer(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self.handle_cancel(&remote_endpoint_id, inner).await; + let _ = tx.send(response).await; + } } } connection.closed().await; @@ -234,6 +312,13 @@ impl TargetedTransferClient { ) -> Result { self.inner.rpc(delivery).await } + + pub(crate) async fn cancel_offer( + &self, + cancel: CancelTargetedOffer, + ) -> Result { + self.inner.rpc(cancel).await + } } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -279,6 +364,17 @@ pub(crate) enum DeliverAuthorizationResponse { Rejected, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct CancelTargetedOffer { + pub(crate) transfer_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum CancelTargetedOfferResponse { + Cancelled, + Rejected, +} + #[rpc_requests(message = TargetedTransferMessage)] #[derive(Debug, Serialize, Deserialize)] enum TargetedTransferMessages { @@ -288,6 +384,8 @@ enum TargetedTransferMessages { SubmitTargetedOffer(SubmitTargetedOffer), #[rpc(tx = oneshot::Sender)] DeliverTargetedAuthorization(DeliverTargetedAuthorization), + #[rpc(tx = oneshot::Sender)] + CancelTargetedOffer(CancelTargetedOffer), } /// Helper kept for type visibility in callers that map refuse reasons. diff --git a/crates/vnidrop/src/targeted_transfer/state.rs b/crates/vnidrop/src/targeted_transfer/state.rs index 3abb3ff..de8c318 100644 --- a/crates/vnidrop/src/targeted_transfer/state.rs +++ b/crates/vnidrop/src/targeted_transfer/state.rs @@ -16,7 +16,7 @@ impl TargetedTransferState { Self::Approved | Self::Declined | Self::Cancelled | Self::Failed ) | ( Self::Approved, - Self::Connecting | Self::Cancelled | Self::Failed + Self::Connecting | Self::Cancelled | Self::Failed | Self::Deleted ) | ( Self::Connecting, Self::Transferring | Self::Interrupted | Self::Cancelled | Self::Failed diff --git a/crates/vnidrop/src/tests/targeted_transfer.rs b/crates/vnidrop/src/tests/targeted_transfer.rs index ed79746..7e4bd6e 100644 --- a/crates/vnidrop/src/tests/targeted_transfer.rs +++ b/crates/vnidrop/src/tests/targeted_transfer.rs @@ -21,8 +21,9 @@ impl CoreEventSink for RecordingSink { } struct ProtectedNode { - _data_dir: tempfile::TempDir, - core: Arc, + data_dir: tempfile::TempDir, + secret_store: Arc, + core: Option>, } impl ProtectedNode { @@ -35,19 +36,43 @@ impl ProtectedNode { let core = VnidropCore::initialize_with_test_secret_store( data_dir.path().to_string_lossy().into_owned(), sink, - store, + store.clone(), ) .expect("protected test core"); Self { - _data_dir: data_dir, - core, + data_dir, + secret_store: store, + core: Some(core), } } + + fn core(&self) -> Arc { + self.core.as_ref().expect("core alive").clone() + } + + fn restart(mut self) -> Self { + if let Some(core) = self.core.take() { + core.shutdown(); + } + let sink = Arc::new(RecordingSink { + events: Mutex::new(Vec::new()), + }); + let core = VnidropCore::initialize_with_test_secret_store( + self.data_dir.path().to_string_lossy().into_owned(), + sink, + self.secret_store.clone(), + ) + .expect("restarted protected test core"); + self.core = Some(core); + self + } } impl Drop for ProtectedNode { fn drop(&mut self) { - self.core.shutdown(); + if let Some(core) = self.core.take() { + core.shutdown(); + } } } @@ -93,25 +118,25 @@ fn complete_transfer(sender: &ProtectedNode, receiver: &ProtectedNode, transfer_ let output_dir = tempfile::tempdir().unwrap(); let source_path = source_dir.path().join("hello.txt"); std::fs::write(&source_path, b"mutual consent").unwrap(); - let share = share_path(&sender.core, &source_path, transfer_id); + let share = share_path(&sender.core(), &source_path, transfer_id); let output_dir = output_dir.path().to_string_lossy().to_string(); - let receiver_core = receiver.core.clone(); + let receiver_core = receiver.core().clone(); let ticket = share.ticket.clone(); let handle = std::thread::spawn(move || { receiver_core.receive(ticket, output_dir, Some("receiver".to_string())) }); - let request = wait_for_receiver_request(&sender.core, share.transfer_id); + let request = wait_for_receiver_request(&sender.core(), share.transfer_id); sender - .core + .core() .respond_receiver_request(request.id, true, None) .unwrap(); handle.join().unwrap().unwrap(); let started = Instant::now(); - let peer = receiver.core.status().endpoint_id.clone(); + let peer = receiver.core().status().endpoint_id.clone(); loop { if sender - .core + .core() .list_pairing_eligibilities() .unwrap() .iter() @@ -151,24 +176,24 @@ fn wait_for_relationship( } fn establish_saved(alice: &ProtectedNode, bob: &ProtectedNode, transfer_id: u64) { - let alice_id = alice.core.status().endpoint_id.clone(); - let bob_id = bob.core.status().endpoint_id.clone(); + let alice_id = alice.core().status().endpoint_id.clone(); + let bob_id = bob.core().status().endpoint_id.clone(); complete_transfer(alice, bob, transfer_id); assert!(alice - .core + .core() .request_saved_device_pairing(bob_id.clone()) .unwrap()); wait_for_relationship( - &bob.core, + &bob.core(), &alice_id, DeviceRelationshipState::PendingIncoming, ); assert!(bob - .core + .core() .respond_to_device_pairing(alice_id.clone(), true) .unwrap()); - wait_for_relationship(&alice.core, &bob_id, DeviceRelationshipState::Saved); - wait_for_relationship(&bob.core, &alice_id, DeviceRelationshipState::Saved); + wait_for_relationship(&alice.core(), &bob_id, DeviceRelationshipState::Saved); + wait_for_relationship(&bob.core(), &alice_id, DeviceRelationshipState::Saved); } fn wait_for_pending_offer(core: &VnidropCore) -> PendingTargetedOffer { @@ -190,7 +215,12 @@ fn targeted_source(path: &Path) -> ShareSource { ShareSource { kind: SourceKind::Path, value: path.to_string_lossy().into_owned(), - display_name: Some("payload.txt".to_string()), + display_name: Some( + path.file_name() + .unwrap_or_default() + .to_string_lossy() + .into_owned(), + ), is_directory: false, } } @@ -200,8 +230,8 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { let alice = ProtectedNode::new(); let bob = ProtectedNode::new(); let stranger = ProtectedNode::new(); - let bob_id = bob.core.status().endpoint_id.clone(); - let stranger_id = stranger.core.status().endpoint_id.clone(); + let bob_id = bob.core().status().endpoint_id.clone(); + let stranger_id = stranger.core().status().endpoint_id.clone(); establish_saved(&alice, &bob, 10_001); let source_dir = tempfile::tempdir().unwrap(); @@ -209,7 +239,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { std::fs::write(&source_path, b"immutable payload").unwrap(); let stranger_err = alice - .core + .core() .create_targeted_transfer( stranger_id, vec![targeted_source(&source_path)], @@ -218,7 +248,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { .unwrap_err(); assert!(matches!(stranger_err, VnidropError::Permission { .. })); - let bob_core = bob.core.clone(); + let bob_core = bob.core().clone(); let accept = std::thread::spawn(move || { let offer = wait_for_pending_offer(&bob_core); bob_core @@ -227,7 +257,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { }); let transfer = alice - .core + .core() .create_targeted_transfer( bob_id.clone(), vec![targeted_source(&source_path)], @@ -236,7 +266,10 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { .unwrap(); let _auth = accept.join().unwrap().expect("authorization after approve"); - assert_eq!(transfer.sender_endpoint_id, alice.core.status().endpoint_id); + assert_eq!( + transfer.sender_endpoint_id, + alice.core().status().endpoint_id + ); assert_eq!(transfer.receiver_endpoint_id, bob_id); assert_eq!(transfer.file_count, 1); assert_eq!(transfer.total_size, b"immutable payload".len() as u64); @@ -251,7 +284,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { )); let listed = alice - .core + .core() .get_targeted_transfer(transfer.id.clone()) .unwrap(); let listed = listed.expect("durable targeted transfer"); @@ -267,14 +300,14 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { fn preapproval_offer_is_authenticated_without_ordinary_share_ticket() { let alice = ProtectedNode::new(); let bob = ProtectedNode::new(); - let bob_id = bob.core.status().endpoint_id.clone(); + let bob_id = bob.core().status().endpoint_id.clone(); establish_saved(&alice, &bob, 10_010); let source_dir = tempfile::tempdir().unwrap(); let source_path = source_dir.path().join("payload.txt"); std::fs::write(&source_path, b"offer body").unwrap(); - let bob_core = bob.core.clone(); + let bob_core = bob.core().clone(); let accept = std::thread::spawn(move || { let offer = wait_for_pending_offer(&bob_core); // Offer surfaces identity + manifest summary only — never a reusable ticket. @@ -291,7 +324,7 @@ fn preapproval_offer_is_authenticated_without_ordinary_share_ticket() { }); alice - .core + .core() .create_targeted_transfer( bob_id, vec![targeted_source(&source_path)], @@ -315,13 +348,13 @@ fn invalid_offer_never_becomes_observable_pending_approval() { let alice = ProtectedNode::new(); let bob = ProtectedNode::new(); // No Saved relationship — offer must not surface. - let bob_id = bob.core.status().endpoint_id.clone(); + let bob_id = bob.core().status().endpoint_id.clone(); let source_dir = tempfile::tempdir().unwrap(); let source_path = source_dir.path().join("payload.txt"); std::fs::write(&source_path, b"nope").unwrap(); let err = alice - .core + .core() .create_targeted_transfer( bob_id, vec![targeted_source(&source_path)], @@ -329,7 +362,7 @@ fn invalid_offer_never_becomes_observable_pending_approval() { ) .unwrap_err(); assert!(matches!(err, VnidropError::Permission { .. })); - assert!(bob.core.list_pending_targeted_offers().is_empty()); + assert!(bob.core().list_pending_targeted_offers().is_empty()); } #[test] @@ -337,7 +370,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { let alice = ProtectedNode::new(); let bob = ProtectedNode::new(); let charlie = ProtectedNode::new(); - let bob_id = bob.core.status().endpoint_id.clone(); + let bob_id = bob.core().status().endpoint_id.clone(); establish_saved(&alice, &bob, 10_020); let source_dir = tempfile::tempdir().unwrap(); @@ -345,7 +378,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { let payload = b"bound authorization payload"; std::fs::write(&source_path, payload).unwrap(); - let bob_core = bob.core.clone(); + let bob_core = bob.core().clone(); let offer_id = Arc::new(Mutex::new(None::)); let offer_id_setter = offer_id.clone(); let gate = Arc::new(Mutex::new(false)); @@ -370,7 +403,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { .unwrap() }); - let alice_core = alice.core.clone(); + let alice_core = alice.core().clone(); let source = targeted_source(&source_path); let create = std::thread::spawn(move || { alice_core.create_targeted_transfer(bob_id, vec![source], Some("payload.txt".to_string())) @@ -390,7 +423,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { let _ = transfer_id; // Without an approved authorization, receive must fail — no content yet. - let early_receive = bob.core.receive_targeted_transfer( + let early_receive = bob.core().receive_targeted_transfer( "not-a-real-authorization".to_string(), tempfile::tempdir() .unwrap() @@ -405,7 +438,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { create.join().unwrap().unwrap(); let output = tempfile::tempdir().unwrap(); - bob.core + bob.core() .receive_targeted_transfer(auth.clone(), output.path().to_string_lossy().into_owned()) .unwrap(); assert_eq!( @@ -415,7 +448,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { let charlie_output = tempfile::tempdir().unwrap(); let leaked = charlie - .core + .core() .receive_targeted_transfer(auth, charlie_output.path().to_string_lossy().into_owned()); assert!( leaked.is_err(), @@ -434,7 +467,7 @@ fn invitation_multi_receiver_shares_remain_independently_authorized() { let first_receiver = ProtectedNode::new(); let second_receiver = ProtectedNode::new(); let share = sender - .core + .core() .share_files( vec![ShareSource { kind: SourceKind::Path, @@ -456,7 +489,7 @@ fn invitation_multi_receiver_shares_remain_independently_authorized() { (&second_receiver, second_output.path()), ] { receiver - .core + .core() .receive( share.ticket.clone(), output.to_string_lossy().into_owned(), @@ -469,3 +502,310 @@ fn invitation_multi_receiver_shares_remain_independently_authorized() { ); } } + +fn approve_one( + alice: &ProtectedNode, + bob: &ProtectedNode, + payload: &[u8], + name: &str, +) -> (crate::TargetedTransfer, String) { + let bob_id = bob.core().status().endpoint_id.clone(); + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join(name); + std::fs::write(&source_path, payload).unwrap(); + + let bob_core = bob.core().clone(); + let accept = std::thread::spawn(move || { + let offer = wait_for_pending_offer(&bob_core); + bob_core + .respond_to_targeted_offer(offer.transfer_id, true) + .unwrap() + }); + let transfer = alice + .core() + .create_targeted_transfer( + bob_id, + vec![targeted_source(&source_path)], + Some(name.to_string()), + ) + .unwrap(); + let auth = accept.join().unwrap().expect("authorization"); + (transfer, auth) +} + +#[test] +fn protocol_ops_are_idempotent_for_stable_transfer_id() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_001); + let (transfer, auth) = approve_one(&alice, &bob, b"idempotent payload", "payload.txt"); + + // Replaying approval returns the same authorization — no duplicate prompts. + let again = bob + .core() + .respond_to_targeted_offer(transfer.id.clone(), true) + .unwrap() + .expect("idempotent authorization"); + assert_eq!(again, auth); + + let listed = bob + .core() + .list_targeted_transfers() + .unwrap() + .into_iter() + .filter(|entry| entry.id == transfer.id) + .count(); + assert_eq!(listed, 1, "replay must not create duplicate durable rows"); +} + +#[test] +fn unapproved_offers_vanish_on_cancel_and_core_restart() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + let bob_id = bob.core().status().endpoint_id.clone(); + establish_saved(&alice, &bob, 11_010); + + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("payload.txt"); + std::fs::write(&source_path, b"vanishing offer").unwrap(); + + let bob_core = bob.core().clone(); + let seen = Arc::new(Mutex::new(None::)); + let seen_set = seen.clone(); + let hold = Arc::new(Mutex::new(true)); + let hold_wait = hold.clone(); + let watcher = std::thread::spawn(move || { + let offer = wait_for_pending_offer(&bob_core); + *seen_set.lock().unwrap() = Some(offer.transfer_id.clone()); + let started = Instant::now(); + while *hold_wait.lock().unwrap() { + assert!( + started.elapsed() < Duration::from_secs(30), + "cancel never cleared the live offer" + ); + if bob_core.list_pending_targeted_offers().is_empty() { + return; + } + std::thread::sleep(Duration::from_millis(25)); + } + }); + + let alice_core = alice.core().clone(); + let source = targeted_source(&source_path); + let create = std::thread::spawn(move || { + alice_core.create_targeted_transfer(bob_id, vec![source], Some("payload.txt".to_string())) + }); + + let transfer_id = loop { + if let Some(id) = seen.lock().unwrap().clone() { + break id; + } + std::thread::sleep(Duration::from_millis(25)); + }; + alice.core().cancel_targeted_transfer(transfer_id).unwrap(); + watcher.join().unwrap(); + let create_err = create.join().unwrap(); + assert!(create_err.is_err(), "cancelled offer must fail create"); + assert!(bob.core().list_pending_targeted_offers().is_empty()); + + // Restart clears any live-session inbox residue. + let bob = bob.restart(); + assert!(bob.core().list_pending_targeted_offers().is_empty()); + *hold.lock().unwrap() = false; +} + +#[test] +fn approved_transfer_resumes_after_restart_without_reapproval() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_020); + let (transfer, _auth) = approve_one(&alice, &bob, b"resume me please", "payload.txt"); + + let bob_before = bob + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .expect("receiver durable approved row"); + assert_eq!(bob_before.state, TargetedTransferState::Approved); + assert_eq!(bob_before.verified_bytes, 0); + + let alice = alice.restart(); + let bob = bob.restart(); + + let alice_after = alice + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .expect("sender durable state"); + assert_eq!(alice_after.state, TargetedTransferState::Approved); + let bob_after = bob + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .expect("receiver durable state"); + assert_eq!(bob_after.state, TargetedTransferState::Approved); + + let output = tempfile::tempdir().unwrap(); + bob.core() + .resume_targeted_transfer( + transfer.id.clone(), + output.path().to_string_lossy().into_owned(), + ) + .unwrap(); + assert_eq!( + std::fs::read(output.path().join("payload.txt")).unwrap(), + b"resume me please" + ); + let completed = bob + .core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap(); + assert_eq!(completed.state, TargetedTransferState::Completed); + assert_eq!(completed.verified_bytes, b"resume me please".len() as u64); +} + +#[test] +fn manifest_change_requires_new_transfer_identity() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_030); + let (first, _) = approve_one(&alice, &bob, b"first manifest", "a.txt"); + let (second, _) = approve_one(&alice, &bob, b"second manifest", "b.txt"); + assert_ne!(first.id, second.id); + assert_ne!(first.manifest_id, second.manifest_id); +} + +#[test] +fn cancel_revokes_access_and_stops_streaming() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_040); + let (transfer, auth) = approve_one(&alice, &bob, b"cancel me", "payload.txt"); + + alice + .core() + .cancel_targeted_transfer(transfer.id.clone()) + .unwrap(); + let cancelled = alice + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + assert_eq!(cancelled.state, TargetedTransferState::Cancelled); + + let output = tempfile::tempdir().unwrap(); + let receive = bob + .core() + .receive_targeted_transfer(auth, output.path().to_string_lossy().into_owned()); + assert!( + receive.is_err(), + "cancelled transfer must not remain receivable" + ); +} + +#[test] +fn delete_removes_authorization_and_resumable_state() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_050); + let (transfer, auth) = approve_one(&alice, &bob, b"delete me", "payload.txt"); + + bob.core() + .delete_targeted_transfer(transfer.id.clone()) + .unwrap(); + let deleted = bob + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + assert_eq!(deleted.state, TargetedTransferState::Deleted); + assert_eq!(deleted.verified_bytes, 0); + + let resume = bob.core().resume_targeted_transfer( + transfer.id.clone(), + tempfile::tempdir() + .unwrap() + .path() + .to_string_lossy() + .into_owned(), + ); + assert!(resume.is_err(), "deleted transfer must not resume"); + + let receive = bob.core().receive_targeted_transfer( + auth, + tempfile::tempdir() + .unwrap() + .path() + .to_string_lossy() + .into_owned(), + ); + // Auth blob may still decode, but durable resume path is gone; receive may + // still attempt content pull if sender serves — sender delete is separate. + let _ = receive; + + alice + .core() + .delete_targeted_transfer(transfer.id.clone()) + .unwrap(); + let sender_deleted = alice + .core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap(); + assert_eq!(sender_deleted.state, TargetedTransferState::Deleted); +} + +#[test] +fn concurrent_independent_transfers_between_same_devices_are_isolated() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_060); + + // Design allows only one unresolved offer per sender; approve sequentially, + // then prove independent approved transfers do not corrupt each other. + let (first, first_auth) = approve_one(&alice, &bob, b"alpha", "one.txt"); + let (second, second_auth) = approve_one(&alice, &bob, b"beta-payload", "two.txt"); + assert_ne!(first.id, second.id); + + alice + .core() + .cancel_targeted_transfer(first.id.clone()) + .unwrap(); + assert_eq!( + alice + .core() + .get_targeted_transfer(first.id) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Cancelled + ); + assert_eq!( + alice + .core() + .get_targeted_transfer(second.id.clone()) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Approved + ); + + let output = tempfile::tempdir().unwrap(); + bob.core() + .receive_targeted_transfer(second_auth, output.path().to_string_lossy().into_owned()) + .unwrap(); + assert_eq!( + std::fs::read(output.path().join("two.txt")).unwrap(), + b"beta-payload" + ); + + let cancelled_output = tempfile::tempdir().unwrap(); + assert!(bob + .core() + .receive_targeted_transfer( + first_auth, + cancelled_output.path().to_string_lossy().into_owned(), + ) + .is_err()); +} diff --git a/crates/vnidrop/tests/experimental_domain.rs b/crates/vnidrop/tests/experimental_domain.rs index fb79840..5625e73 100644 --- a/crates/vnidrop/tests/experimental_domain.rs +++ b/crates/vnidrop/tests/experimental_domain.rs @@ -43,6 +43,7 @@ fn saved_devices_relationships_and_targeted_transfers_are_distinct_contracts() { manifest_id: "immutable-manifest-id".to_string(), file_count: 2, total_size: 42, + verified_bytes: 0, state: TargetedTransferState::AwaitingApproval, created_at: 3_000, updated_at: 3_000, @@ -67,6 +68,7 @@ fn targeted_transfer_transitions_are_validated_by_the_domain() { (State::AwaitingApproval, State::Approved), (State::AwaitingApproval, State::Declined), (State::Approved, State::Connecting), + (State::Approved, State::Deleted), (State::Connecting, State::Transferring), (State::Connecting, State::Interrupted), (State::Transferring, State::Completed),