diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index e684446..1e7f540 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -101,6 +101,21 @@ pub struct TargetedTransfer { pub updated_at: i64, } +/// Pre-approval offer summary. Deliberately omits any reusable share ticket. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, uniffi::Record)] +pub struct PendingTargetedOffer { + pub transfer_id: String, + pub sender_endpoint_id: String, + pub receiver_endpoint_id: String, + pub manifest_id: String, + pub content_hash: String, + pub transfer_name: String, + pub file_count: u64, + pub total_size: u64, + pub protocol_version: u16, + pub received_at: i64, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, uniffi::Enum)] pub enum CoreRelayMode { Automatic, diff --git a/crates/vnidrop/src/device_relationship/mod.rs b/crates/vnidrop/src/device_relationship/mod.rs index 409b397..df1a1c2 100644 --- a/crates/vnidrop/src/device_relationship/mod.rs +++ b/crates/vnidrop/src/device_relationship/mod.rs @@ -847,6 +847,75 @@ impl DeviceRelationshipService { Ok(()) } + /// Prove possession of the held grant for a Saved peer (targeted offers). + pub(crate) async fn prove_saved_possession( + &self, + peer_endpoint_id: &str, + challenge: &Challenge, + ) -> Result<(WireProof, u64, u16), VnidropError> { + let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| { + VnidropError::permission(anyhow::anyhow!("no saved relationship with peer")) + })?; + if row.state != DeviceRelationshipState::Saved { + return Err(VnidropError::permission(anyhow::anyhow!( + "peer is not a saved device" + ))); + } + let proof = self + .prove_held_possession( + peer_endpoint_id, + challenge, + row.generation, + row.minimum_protocol_version, + ) + .await?; + Ok((proof, row.generation, row.minimum_protocol_version)) + } + + /// Verify a Saved peer's held-grant proof against our issued grant. + pub(crate) async fn verify_saved_possession( + &self, + peer_endpoint_id: &str, + challenge: &Challenge, + proof: &WireProof, + generation: u64, + protocol_version: u16, + ) -> Result<(), VnidropError> { + let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| { + VnidropError::permission(anyhow::anyhow!("no saved relationship with peer")) + })?; + if row.state != DeviceRelationshipState::Saved { + return Err(VnidropError::permission(anyhow::anyhow!( + "peer is not a saved device" + ))); + } + if row.generation != generation { + return Err(VnidropError::permission(anyhow::anyhow!( + "relationship generation mismatch" + ))); + } + self.verify_issued_possession( + peer_endpoint_id, + challenge, + proof, + generation, + protocol_version, + ) + .await + } + + pub(crate) async fn require_saved(&self, peer_endpoint_id: &str) -> Result<(), VnidropError> { + let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| { + VnidropError::permission(anyhow::anyhow!("no saved relationship with peer")) + })?; + if row.state != DeviceRelationshipState::Saved { + return Err(VnidropError::permission(anyhow::anyhow!( + "peer is not a saved device" + ))); + } + Ok(()) + } + async fn prove_held_possession( &self, peer_endpoint_id: &str, @@ -1080,7 +1149,10 @@ impl DeviceRelationshipService { Ok(()) } - async fn peer_addr(&self, peer_endpoint_id: &str) -> Result { + pub(crate) async fn peer_addr( + &self, + peer_endpoint_id: &str, + ) -> Result { let parsed: EndpointId = peer_endpoint_id .parse() .context("unusable peer endpoint id") @@ -1277,10 +1349,10 @@ struct WireGrant { } #[derive(Debug, Clone, Serialize, Deserialize)] -struct WireProof { - grant_id: String, - mac: String, - challenge: String, +pub(crate) struct WireProof { + pub(crate) grant_id: String, + pub(crate) mac: String, + pub(crate) challenge: String, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/crates/vnidrop/src/lib.rs b/crates/vnidrop/src/lib.rs index 9c68b6c..7851553 100644 --- a/crates/vnidrop/src/lib.rs +++ b/crates/vnidrop/src/lib.rs @@ -32,10 +32,10 @@ pub use api::{ CoreEventSink, CoreLimits, CoreNetworkConfig, CoreRelayMode, CoreStorageUsage, DeviceRelationship, DeviceRelationshipState, ExperimentalSavedDeviceCapabilities, GrantLifetimeSetting, HeldOfferSummary, IncomingOffer, PairingEligibilitySummary, - PendingPairing, PublishedOutput, ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedArtifact, - ReceivedLocatorKind, ReceiverRequest, RuntimeStatus, SavedDevice, ShareMetadataInput, - ShareResult, ShareSource, SourceKind, StoredTransfer, TargetedTransfer, TargetedTransferState, - TicketInspection, TransferAccessMode, TransferMetadata, + PendingPairing, PendingTargetedOffer, PublishedOutput, ReceiveOutputSink, ReceiveOutputSinkV2, + ReceivedArtifact, ReceivedLocatorKind, ReceiverRequest, RuntimeStatus, SavedDevice, + ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer, TargetedTransfer, + TargetedTransferState, TicketInspection, TransferAccessMode, TransferMetadata, }; pub use error::VnidropError; pub use runtime::VnidropCore; diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index 7b58375..2db9c0f 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -324,6 +324,7 @@ impl Repository { crate::contacts::ensure_schema(&self.pool).await?; crate::secure_secret::ensure_schema(&self.pool).await?; crate::device_relationship::DeviceRelationshipService::ensure_schema(&self.pool).await?; + crate::targeted_transfer::ensure_schema(&self.pool).await?; sqlx::query( r#" CREATE TABLE IF NOT EXISTS pairing_eligibilities ( diff --git a/crates/vnidrop/src/runtime/contacts.rs b/crates/vnidrop/src/runtime/contacts.rs index b1ce605..35b7fd6 100644 --- a/crates/vnidrop/src/runtime/contacts.rs +++ b/crates/vnidrop/src/runtime/contacts.rs @@ -106,7 +106,7 @@ impl CoreInner { // Targeted transfers for this relationship only (ticket 10 fills in). // Invitation-domain shares are deliberately not cancelled here. self.cancel_targeted_transfers_for_peer(&peer_endpoint_id) - .await; + .await?; self.emit_endpoint( "pairing", "saved-device-forgotten", @@ -139,7 +139,7 @@ impl CoreInner { .revoke_for_block(&peer_endpoint_id) .await?; self.cancel_targeted_transfers_for_peer(&peer_endpoint_id) - .await; + .await?; self.offers.discard_from(&peer_endpoint_id).await; self.emit_endpoint( "pairing", @@ -182,21 +182,6 @@ impl CoreInner { .await } - /// Cancels active or resumable targeted transfers for a peer relationship. - /// - /// Stub until ticket 10 lands targeted-transfer cancellation. Forget/block - /// call this for immediate local effect without touching invitation shares. - pub(super) async fn cancel_targeted_transfers_for_peer(&self, peer_endpoint_id: &str) { - #[cfg(test)] - { - self.targeted_cancel_log - .lock() - .expect("targeted cancel log") - .push(peer_endpoint_id.to_string()); - } - let _ = peer_endpoint_id; - } - #[cfg(test)] pub(super) fn targeted_cancel_log_for_test(&self) -> Vec { self.targeted_cancel_log diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index 8064a96..77c29a7 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -484,6 +484,65 @@ impl VnidropCore { self.block_on(self.inner.rotate_relationship_grant(peer_endpoint_id)) } + /// Create an immutable one-receiver transfer and submit its pre-approval offer. + /// + /// Blocks until the saved receiver approves or declines. On approval the + /// receiver obtains bound authorization via [`Self::respond_to_targeted_offer`]. + pub fn create_targeted_transfer( + &self, + receiver_endpoint_id: String, + sources: Vec, + transfer_name: Option, + ) -> Result { + self.block_on(self.inner.create_targeted_transfer( + receiver_endpoint_id, + sources, + transfer_name, + )) + } + + /// Offline-only pending offers awaiting explicit local approval. + pub fn list_pending_targeted_offers(&self) -> Vec { + self.block_on(self.inner.list_pending_targeted_offers()) + } + + /// Approve or decline a pending targeted offer. + /// + /// On approval, returns the recipient-bound authorization capability used + /// with [`Self::receive_targeted_transfer`]. Declining returns `None`. + pub fn respond_to_targeted_offer( + &self, + transfer_id: String, + accepted: bool, + ) -> Result, VnidropError> { + self.block_on(self.inner.respond_to_targeted_offer(transfer_id, accepted)) + } + + /// Pull an approved targeted transfer through existing output-sink machinery. + pub fn receive_targeted_transfer( + &self, + authorization: String, + output_dir: String, + ) -> Result<(), VnidropError> { + self.block_on( + self.inner + .receive_targeted_transfer(authorization, output_dir), + ) + } + + pub fn get_targeted_transfer( + &self, + id: String, + ) -> Result, VnidropError> { + self.block_on(self.inner.get_targeted_transfer(id)) + } + + pub fn list_targeted_transfers( + &self, + ) -> Result, VnidropError> { + self.block_on(self.inner.list_targeted_transfers()) + } + /// 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 d5b5e86..21a590d 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -7,6 +7,7 @@ //! - [`lifecycle`] — cancel/delete/shutdown/status/access //! - [`provider`] — blob provider events and per-connection send progress //! - [`contacts`] — device history: pairing, forgetting, blocking +//! - [`targeted`] — saved-device targeted transfers mod contacts; mod delivery; @@ -16,6 +17,7 @@ mod provider; mod receive; mod share; mod storage; +mod targeted; #[cfg(test)] pub(crate) use self::contacts::should_poll; @@ -67,6 +69,7 @@ use crate::{ repository::Repository, secret::load_or_create_secret, secure_secret::{start_endpoint_identity, ProfileLock, SecureSecretStore}, + targeted_transfer::{TargetedOfferInbox, TargetedTransferProtocol}, ticket::ticket_matches_relay_profile, transfer_state::{TransferDirection, TransferStatus}, }; @@ -109,6 +112,7 @@ pub(super) struct CoreInner { pub(super) pairing_eligibility: PairingEligibilityService, pub(super) device_relationships: Arc, pub(super) offers: OfferInbox, + pub(super) targeted_offers: TargetedOfferInbox, /// Endpoint → last poll time, for the rate limit above. pub(super) last_polled: TokioMutex>, pub(super) limits: CoreLimits, @@ -390,6 +394,8 @@ impl CoreInner { tracing::warn!(%error, "failed to sweep dead grants"); } let offers = OfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize); + let targeted_offers = + TargetedOfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize); let device_relationships = Arc::new(DeviceRelationshipService::new( repository.sqlite_pool(), secret_custody.clone(), @@ -409,6 +415,15 @@ impl CoreInner { RelationshipProtocol::ALPN, RelationshipProtocol::new(device_relationships.clone()), ) + .accept( + TargetedTransferProtocol::ALPN, + TargetedTransferProtocol::new( + device_relationships.clone(), + targeted_offers.clone(), + limits.clone(), + endpoint.id().to_string(), + ), + ) .spawn(); let inner = Arc::new(Self { @@ -424,6 +439,7 @@ impl CoreInner { pairing_eligibility, device_relationships, offers, + targeted_offers, last_polled: TokioMutex::new(HashMap::new()), relay_mode, custom_relay_urls: relay_urls, diff --git a/crates/vnidrop/src/runtime/targeted.rs b/crates/vnidrop/src/runtime/targeted.rs new file mode 100644 index 0000000..007ae6b --- /dev/null +++ b/crates/vnidrop/src/runtime/targeted.rs @@ -0,0 +1,342 @@ +//! Create, offer, approve, and receive targeted transfers between Saved devices. + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use iroh_blobs::{ticket::BlobTicket, BlobFormat}; +use uuid::Uuid; + +use super::CoreInner; +use crate::{ + api::{ + experimental_saved_device_capabilities, PendingTargetedOffer, ShareMetadataInput, + ShareSource, TargetedTransfer, TargetedTransferState, TransferAccessMode, TransferMetadata, + }, + error::VnidropError, + targeted_transfer::{ + protocol::{ + DeliverTargetedAuthorization, SubmitTargetedOffer, TargetedOfferResponse, + TargetedTransferProtocol, + }, + TargetedAuthorization, TargetedAuthorizationDraft, TargetedTransferRow, + TargetedTransferStore, + }, + ticket::VnidropTicket, + util::{non_empty, now_ms}, +}; + +const OFFER_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +impl CoreInner { + pub(super) fn targeted_store(&self) -> TargetedTransferStore { + TargetedTransferStore::new(self.repository.sqlite_pool()) + } + + pub(super) async fn list_pending_targeted_offers(&self) -> Vec { + self.targeted_offers.list().await + } + + pub(super) async fn get_targeted_transfer( + &self, + id: String, + ) -> Result, VnidropError> { + self.targeted_store().get(&id).await + } + + pub(super) async fn list_targeted_transfers( + &self, + ) -> Result, VnidropError> { + self.targeted_store().list().await + } + + /// Cancel in-flight targeted transfers involving `peer` (for forget/block). + pub(crate) async fn cancel_targeted_transfers_for_peer( + &self, + peer_endpoint_id: &str, + ) -> Result { + #[cfg(test)] + { + self.targeted_cancel_log + .lock() + .expect("targeted cancel log") + .push(peer_endpoint_id.to_string()); + } + self.targeted_offers.discard_from(peer_endpoint_id).await; + self.targeted_store().cancel_by_peer(peer_endpoint_id).await + } + + pub(super) async fn respond_to_targeted_offer( + &self, + transfer_id: String, + accepted: bool, + ) -> Result, VnidropError> { + match self.targeted_offers.respond(&transfer_id, accepted).await { + Ok(auth) => Ok(auth), + Err(crate::targeted_transfer::RespondError::Unknown) => Err( + VnidropError::invalid_input(anyhow::anyhow!("unknown targeted offer")), + ), + Err(crate::targeted_transfer::RespondError::SenderGone) => Err(VnidropError::network( + anyhow::anyhow!("sender disconnected before approval completed"), + )), + Err(crate::targeted_transfer::RespondError::AuthorizationTimeout) => Err( + VnidropError::network(anyhow::anyhow!("authorization was not delivered in time")), + ), + } + } + + pub(super) async fn create_targeted_transfer( + self: &Arc, + receiver_endpoint_id: String, + sources: Vec, + transfer_name: Option, + ) -> Result { + self.device_relationships + .require_saved(&receiver_endpoint_id) + .await?; + + let transfer_uuid = Uuid::new_v4().to_string(); + let protocol_transfer_id = allocate_protocol_transfer_id(&transfer_uuid); + let sender_endpoint_id = self.endpoint.id().to_string(); + let now = now_ms(); + + let share = self + .share_files( + sources, + ShareMetadataInput { + transfer_id: protocol_transfer_id, + transfer_name: transfer_name.clone(), + sender_name: None, + access_mode: TransferAccessMode::ApprovalRequired, + }, + ) + .await + .map_err(VnidropError::transfer)?; + + let store = self.targeted_store(); + let row = TargetedTransferRow { + id: transfer_uuid.clone(), + protocol_transfer_id, + sender_endpoint_id: sender_endpoint_id.clone(), + receiver_endpoint_id: receiver_endpoint_id.clone(), + manifest_id: share.hash.clone(), + content_hash: share.hash.clone(), + transfer_name: share.transfer_name.clone(), + file_count: share.file_count, + total_size: share.total_size, + state: TargetedTransferState::Preparing, + created_at: now, + updated_at: now, + }; + store.insert(&row).await?; + store + .set_state( + &transfer_uuid, + TargetedTransferState::Preparing, + TargetedTransferState::Offering, + ) + .await?; + + let addr = self + .device_relationships + .peer_addr(&receiver_endpoint_id) + .await?; + let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr); + let challenge = tokio::time::timeout(OFFER_CONNECT_TIMEOUT, client.request_challenge()) + .await + .map_err(|_| VnidropError::network(anyhow::anyhow!("device did not answer in time")))? + .context("device is not reachable") + .map_err(VnidropError::network)?; + + let (proof, generation, relationship_protocol_version) = self + .device_relationships + .prove_saved_possession(&receiver_endpoint_id, &challenge) + .await?; + + let protocol_version = + experimental_saved_device_capabilities().targeted_transfer_protocol_version; + store + .set_state( + &transfer_uuid, + TargetedTransferState::Offering, + TargetedTransferState::AwaitingApproval, + ) + .await?; + + let response = tokio::time::timeout( + OFFER_CONNECT_TIMEOUT + std::time::Duration::from_secs(120), + client.submit_offer(SubmitTargetedOffer { + proof, + generation, + relationship_protocol_version, + protocol_version, + transfer_id: transfer_uuid.clone(), + sender_endpoint_id: sender_endpoint_id.clone(), + receiver_endpoint_id: receiver_endpoint_id.clone(), + manifest_id: share.hash.clone(), + content_hash: share.hash.clone(), + transfer_name: share.transfer_name.clone(), + file_count: share.file_count, + total_size: share.total_size, + }), + ) + .await + .map_err(|_| VnidropError::network(anyhow::anyhow!("offer timed out")))? + .context("failed to submit targeted offer") + .map_err(VnidropError::network)?; + + match response { + TargetedOfferResponse::Accepted => {} + TargetedOfferResponse::Declined { reason } => { + let _ = store + .set_state( + &transfer_uuid, + TargetedTransferState::AwaitingApproval, + TargetedTransferState::Declined, + ) + .await; + let _ = self.cancel_idle_or_share(protocol_transfer_id).await; + return Err(VnidropError::permission(anyhow::anyhow!( + "targeted offer declined: {reason}" + ))); + } + TargetedOfferResponse::Refused { reason } => { + let _ = store + .set_state( + &transfer_uuid, + TargetedTransferState::AwaitingApproval, + TargetedTransferState::Failed, + ) + .await; + let _ = self.cancel_idle_or_share(protocol_transfer_id).await; + return Err(VnidropError::permission(anyhow::anyhow!( + "targeted offer refused: {reason}" + ))); + } + } + + // Bound authorization: only the approved receiver endpoint may fetch. + self.access_policy + .approve_endpoint(protocol_transfer_id, receiver_endpoint_id.clone()) + .await; + + let parsed = crate::ticket::parse_transfer_ticket_with_limits(&share.ticket, &self.limits) + .map_err(VnidropError::ticket)?; + let blob_ticket = BlobTicket::new( + parsed.blob_ticket.addr().clone(), + parsed.blob_ticket.hash(), + BlobFormat::HashSeq, + ); + let authorization = TargetedAuthorization::issue(TargetedAuthorizationDraft { + transfer_id: transfer_uuid.clone(), + protocol_transfer_id, + sender_endpoint_id, + receiver_endpoint_id, + manifest_id: share.hash.clone(), + content_hash: share.hash.clone(), + file_count: share.file_count, + total_size: share.total_size, + protocol_version, + transfer_name: share.transfer_name.clone(), + blob_ticket: blob_ticket.to_string(), + })?; + let encoded = authorization.encode()?; + + let deliver = client + .deliver_authorization(DeliverTargetedAuthorization { + transfer_id: transfer_uuid.clone(), + authorization: encoded, + }) + .await + .context("failed to deliver targeted authorization") + .map_err(VnidropError::network)?; + if deliver != crate::targeted_transfer::protocol::DeliverAuthorizationResponse::Stored { + let _ = store + .set_state( + &transfer_uuid, + TargetedTransferState::AwaitingApproval, + TargetedTransferState::Failed, + ) + .await; + return Err(VnidropError::network(anyhow::anyhow!( + "receiver rejected authorization delivery" + ))); + } + + store + .set_state( + &transfer_uuid, + TargetedTransferState::AwaitingApproval, + TargetedTransferState::Approved, + ) + .await?; + + store + .get(&transfer_uuid) + .await? + .ok_or_else(|| VnidropError::internal(anyhow::anyhow!("targeted transfer missing"))) + } + + pub(super) async fn receive_targeted_transfer( + self: &Arc, + authorization: String, + output_dir: String, + ) -> Result<(), VnidropError> { + let auth = TargetedAuthorization::decode(&authorization)?; + auth.verify_for_receiver(&self.endpoint.id().to_string())?; + + let blob_ticket = BlobTicket::from_str_compat(&auth.blob_ticket) + .map_err(|error| VnidropError::ticket(anyhow::anyhow!(error)))?; + let metadata = TransferMetadata::new( + auth.protocol_transfer_id, + non_empty(auth.transfer_name.clone()).unwrap_or_else(|| "transfer".to_string()), + None, + blob_ticket.hash(), + auth.file_count, + auth.total_size, + ); + let ticket = + VnidropTicket::new_with_relay_urls(blob_ticket, metadata, &self.custom_relay_urls) + .encode() + .map_err(VnidropError::ticket)?; + + self.receive(ticket, std::path::PathBuf::from(output_dir), None) + .await + .map_err(VnidropError::transfer)?; + + 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; + } + Ok(()) + } +} + +fn allocate_protocol_transfer_id(transfer_uuid: &str) -> u64 { + let hash = blake3::hash(transfer_uuid.as_bytes()); + let mut bytes = [0u8; 8]; + bytes.copy_from_slice(&hash.as_bytes()[..8]); + // SQLite transfer ids are signed; keep within i64::MAX. + let value = u64::from_le_bytes(bytes) & (i64::MAX as u64); + if value == 0 { + 1 + } else { + value + } +} + +trait BlobTicketParse { + fn from_str_compat(value: &str) -> Result; +} + +impl BlobTicketParse for BlobTicket { + fn from_str_compat(value: &str) -> Result { + use std::str::FromStr; + BlobTicket::from_str(value).map_err(|error| error.to_string()) + } +} diff --git a/crates/vnidrop/src/targeted_transfer/auth.rs b/crates/vnidrop/src/targeted_transfer/auth.rs new file mode 100644 index 0000000..475ce62 --- /dev/null +++ b/crates/vnidrop/src/targeted_transfer/auth.rs @@ -0,0 +1,140 @@ +//! Bound post-approval authorization for a single targeted transfer. +//! +//! The pre-approval offer never carries this material. After explicit approval, +//! the sender issues a capability whose MAC binds the exact recipient, sender, +//! transfer, manifest, hashes, sizes, and protocol generation. Tampering with +//! the receiver identity invalidates the MAC; presenting an intact capability +//! from another endpoint still fails provider ACL and local identity checks. + +use data_encoding::{BASE64URL_NOPAD, HEXLOWER}; +use serde::{Deserialize, Serialize}; + +use crate::error::VnidropError; + +const AUTH_CONTEXT: &[u8] = b"vnidrop-targeted-auth-v1"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct TargetedAuthorization { + pub(crate) transfer_id: String, + pub(crate) protocol_transfer_id: u64, + pub(crate) sender_endpoint_id: String, + pub(crate) receiver_endpoint_id: String, + pub(crate) manifest_id: String, + pub(crate) content_hash: String, + pub(crate) file_count: u64, + pub(crate) total_size: u64, + pub(crate) protocol_version: u16, + pub(crate) transfer_name: String, + /// BlobTicket string used only after approval to pull through existing sinks. + pub(crate) blob_ticket: String, + pub(crate) auth_secret: String, + pub(crate) mac: String, +} + +#[derive(Debug, Clone)] +pub(crate) struct TargetedAuthorizationDraft { + pub(crate) transfer_id: String, + pub(crate) protocol_transfer_id: u64, + pub(crate) sender_endpoint_id: String, + pub(crate) receiver_endpoint_id: String, + pub(crate) manifest_id: String, + pub(crate) content_hash: String, + pub(crate) file_count: u64, + pub(crate) total_size: u64, + pub(crate) protocol_version: u16, + pub(crate) transfer_name: String, + pub(crate) blob_ticket: String, +} + +impl TargetedAuthorization { + pub(crate) fn issue(draft: TargetedAuthorizationDraft) -> Result { + let auth_secret = crate::grant::GrantSecret::generate().encode(); + let mut auth = Self { + 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) + } + + pub(crate) fn encode(&self) -> Result { + let bytes = serde_json::to_vec(self).map_err(VnidropError::internal)?; + Ok(format!("vndta1:{}", BASE64URL_NOPAD.encode(&bytes))) + } + + pub(crate) fn decode(value: &str) -> Result { + let encoded = value.strip_prefix("vndta1:").ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("not a targeted authorization")) + })?; + let bytes = BASE64URL_NOPAD.decode(encoded.as_bytes()).map_err(|_| { + VnidropError::invalid_input(anyhow::anyhow!("invalid targeted authorization encoding")) + })?; + let auth: Self = serde_json::from_slice(&bytes).map_err(|_| { + VnidropError::invalid_input(anyhow::anyhow!("invalid targeted authorization payload")) + })?; + auth.verify_integrity()?; + Ok(auth) + } + + pub(crate) fn verify_for_receiver(&self, local_endpoint_id: &str) -> Result<(), VnidropError> { + self.verify_integrity()?; + if self.receiver_endpoint_id != local_endpoint_id { + return Err(VnidropError::permission(anyhow::anyhow!( + "targeted authorization is bound to a different receiver" + ))); + } + Ok(()) + } + + fn verify_integrity(&self) -> Result<(), VnidropError> { + let expected = self.compute_mac()?; + let presented = HEXLOWER.decode(self.mac.as_bytes()).map_err(|_| { + VnidropError::invalid_input(anyhow::anyhow!("invalid authorization mac")) + })?; + if presented.as_slice() != expected.as_slice() { + return Err(VnidropError::permission(anyhow::anyhow!( + "targeted authorization mac mismatch" + ))); + } + Ok(()) + } + + fn compute_mac(&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")) + })?; + let key: [u8; 32] = secret_bytes.try_into().map_err(|_| { + VnidropError::invalid_input(anyhow::anyhow!("invalid authorization secret length")) + })?; + let mut hasher = blake3::Hasher::new_keyed(&key); + hasher.update(AUTH_CONTEXT); + for field in [ + self.transfer_id.as_bytes(), + self.sender_endpoint_id.as_bytes(), + self.receiver_endpoint_id.as_bytes(), + self.manifest_id.as_bytes(), + self.content_hash.as_bytes(), + self.transfer_name.as_bytes(), + self.blob_ticket.as_bytes(), + ] { + hasher.update(&(field.len() as u64).to_le_bytes()); + hasher.update(field); + } + hasher.update(&self.protocol_transfer_id.to_le_bytes()); + hasher.update(&self.file_count.to_le_bytes()); + hasher.update(&self.total_size.to_le_bytes()); + hasher.update(&self.protocol_version.to_le_bytes()); + Ok(*hasher.finalize().as_bytes()) + } +} diff --git a/crates/vnidrop/src/targeted_transfer/inbox.rs b/crates/vnidrop/src/targeted_transfer/inbox.rs new file mode 100644 index 0000000..dbed28a --- /dev/null +++ b/crates/vnidrop/src/targeted_transfer/inbox.rs @@ -0,0 +1,234 @@ +//! Live-session inbox for unapproved targeted-transfer offers. +//! +//! Offers are not durable: cancellation, timeout, disconnect, or restart drops +//! them. Authorization is delivered only after the local user accepts. + +use std::{collections::HashMap, sync::Arc, time::Duration}; + +use serde_json::json; +use tokio::sync::{oneshot, Mutex}; +use uuid::Uuid; + +use crate::{api::PendingTargetedOffer, event_hub::EventHub}; + +const OFFER_WAIT_TIMEOUT: Duration = Duration::from_secs(120); + +#[derive(Debug, Clone)] +pub(crate) struct PendingTargetedOfferRecord { + pub(crate) offer: PendingTargetedOffer, +} + +struct DecisionWaiter { + decision: oneshot::Sender, +} + +struct AuthWaiter { + auth: oneshot::Sender, +} + +#[derive(Clone)] +pub(crate) struct TargetedOfferInbox { + event_hub: Arc, + pending: Arc>>, + decisions: Arc>>, + auths: Arc>>, + max_pending: usize, +} + +impl TargetedOfferInbox { + pub(crate) fn new(event_hub: Arc, max_pending: usize) -> Self { + Self { + event_hub, + pending: Arc::new(Mutex::new(HashMap::new())), + decisions: Arc::new(Mutex::new(HashMap::new())), + auths: Arc::new(Mutex::new(HashMap::new())), + max_pending, + } + } + + /// Surface a validated offer and block until the local user decides. + pub(crate) async fn submit(&self, offer: PendingTargetedOffer) -> TargetedOfferDecision { + let transfer_id = offer.transfer_id.clone(); + let (decision_tx, decision_rx) = oneshot::channel(); + { + let mut pending = self.pending.lock().await; + if pending.len() >= self.max_pending { + return TargetedOfferDecision::Refused { + reason: "too-many-pending-offers".to_string(), + }; + } + if pending + .values() + .any(|entry| entry.offer.sender_endpoint_id == offer.sender_endpoint_id) + { + return TargetedOfferDecision::Refused { + reason: "offer-already-pending".to_string(), + }; + } + if pending.contains_key(&transfer_id) { + return TargetedOfferDecision::Refused { + reason: "duplicate-transfer".to_string(), + }; + } + pending.insert( + transfer_id.clone(), + PendingTargetedOfferRecord { + offer: offer.clone(), + }, + ); + } + self.decisions.lock().await.insert( + transfer_id.clone(), + DecisionWaiter { + decision: decision_tx, + }, + ); + + self.event_hub.emit_endpoint( + "targeted_transfer", + "offer-received", + json!({ + "transfer_id": offer.transfer_id, + "sender_endpoint_id": offer.sender_endpoint_id, + "file_count": offer.file_count, + "total_size": offer.total_size, + "manifest_id": offer.manifest_id, + }), + ); + + match tokio::time::timeout(OFFER_WAIT_TIMEOUT, decision_rx).await { + Ok(Ok(true)) => TargetedOfferDecision::Accepted, + Ok(Ok(false)) => { + self.discard(&transfer_id).await; + TargetedOfferDecision::Declined { + reason: "receiver-declined".to_string(), + } + } + Ok(Err(_)) | Err(_) => { + self.discard(&transfer_id).await; + TargetedOfferDecision::Declined { + reason: "no-response".to_string(), + } + } + } + } + + pub(crate) async fn list(&self) -> Vec { + self.pending + .lock() + .await + .values() + .map(|entry| entry.offer.clone()) + .collect() + } + + /// Record the local decision. On accept, wait for sender-issued authorization. + pub(crate) async fn respond( + &self, + transfer_id: &str, + accepted: bool, + ) -> Result, RespondError> { + 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 { + return Err(RespondError::Unknown); + }; + if !accepted { + let _ = waiter.decision.send(false); + self.discard(transfer_id).await; + self.event_hub.emit_endpoint( + "targeted_transfer", + "offer-declined", + json!({ "transfer_id": transfer_id }), + ); + return Ok(None); + } + + let (auth_tx, auth_rx) = oneshot::channel(); + self.auths + .lock() + .await + .insert(transfer_id.to_string(), AuthWaiter { auth: auth_tx }); + if waiter.decision.send(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 { + Ok(Ok(auth)) => { + self.pending.lock().await.remove(transfer_id); + self.event_hub.emit_endpoint( + "targeted_transfer", + "offer-accepted", + json!({ "transfer_id": transfer_id }), + ); + Ok(Some(auth)) + } + Ok(Err(_)) | Err(_) => { + self.discard(transfer_id).await; + Err(RespondError::AuthorizationTimeout) + } + } + } + + pub(crate) async fn deliver_authorization( + &self, + transfer_id: &str, + authorization: String, + ) -> bool { + if let Some(waiter) = self.auths.lock().await.remove(transfer_id) { + waiter.auth.send(authorization).is_ok() + } else { + false + } + } + + #[allow( + dead_code, + reason = "called via cancel_targeted_transfers_for_peer for ticket 09" + )] + pub(crate) async fn discard_from(&self, endpoint_id: &str) { + let ids: Vec = { + let pending = self.pending.lock().await; + pending + .values() + .filter(|entry| entry.offer.sender_endpoint_id == endpoint_id) + .map(|entry| entry.offer.transfer_id.clone()) + .collect() + }; + for id in ids { + self.discard(&id).await; + } + } + + 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); + } + self.auths.lock().await.remove(transfer_id); + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum TargetedOfferDecision { + Accepted, + Declined { reason: String }, + Refused { reason: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum RespondError { + Unknown, + SenderGone, + AuthorizationTimeout, +} + +#[allow(dead_code)] +pub(crate) fn new_offer_id() -> String { + Uuid::new_v4().to_string() +} diff --git a/crates/vnidrop/src/targeted_transfer/mod.rs b/crates/vnidrop/src/targeted_transfer/mod.rs new file mode 100644 index 0000000..59883e8 --- /dev/null +++ b/crates/vnidrop/src/targeted_transfer/mod.rs @@ -0,0 +1,267 @@ +//! Immutable one-sender, one-receiver transfers between Saved devices. +//! +//! Separate from ordinary multi-receiver shares: own protocol, authorization, +//! and public APIs. Blob import/streaming/output sinks are reused. + +mod auth; +pub(crate) mod inbox; +pub(crate) mod protocol; +mod state; + +pub(crate) use auth::{TargetedAuthorization, TargetedAuthorizationDraft}; +pub(crate) use inbox::{RespondError, TargetedOfferInbox}; +pub(crate) use protocol::TargetedTransferProtocol; + +use sqlx::{Row, SqlitePool}; + +use crate::{ + api::{TargetedTransfer, TargetedTransferState}, + error::VnidropError, + util::now_ms, +}; + +pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> { + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS targeted_transfers ( + id TEXT PRIMARY KEY, + protocol_transfer_id INTEGER NOT NULL UNIQUE, + sender_endpoint_id TEXT NOT NULL, + receiver_endpoint_id TEXT NOT NULL, + manifest_id TEXT NOT NULL, + content_hash TEXT NOT NULL, + transfer_name TEXT NOT NULL, + file_count INTEGER NOT NULL, + total_size INTEGER NOT NULL, + state TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + "#, + ) + .execute(pool) + .await?; + Ok(()) +} + +pub(crate) struct TargetedTransferStore { + pool: SqlitePool, +} + +impl TargetedTransferStore { + pub(crate) fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + pub(crate) async fn insert(&self, transfer: &TargetedTransferRow) -> Result<(), VnidropError> { + sqlx::query( + r#" + INSERT INTO targeted_transfers ( + id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id, + manifest_id, content_hash, transfer_name, file_count, total_size, + state, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) + "#, + ) + .bind(&transfer.id) + .bind(transfer.protocol_transfer_id as i64) + .bind(&transfer.sender_endpoint_id) + .bind(&transfer.receiver_endpoint_id) + .bind(&transfer.manifest_id) + .bind(&transfer.content_hash) + .bind(&transfer.transfer_name) + .bind(transfer.file_count as i64) + .bind(transfer.total_size as i64) + .bind(state_as_str(transfer.state)) + .bind(transfer.created_at) + .bind(transfer.updated_at) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + pub(crate) async fn set_state( + &self, + id: &str, + from: TargetedTransferState, + to: TargetedTransferState, + ) -> Result<(), VnidropError> { + from.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(from)) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + if result.rows_affected() == 0 { + return Err(VnidropError::InvalidTransition { + reason: format!("{} -> {}", state_as_str(from), state_as_str(to)), + }); + } + 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 + FROM targeted_transfers WHERE id = ?1 + "#, + ) + .bind(id) + .fetch_optional(&self.pool) + .await + .map_err(VnidropError::repository)?; + row.map(row_to_transfer).transpose() + } + + pub(crate) async fn get_row( + &self, + id: &str, + ) -> Result, VnidropError> { + let row = sqlx::query( + r#" + SELECT id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id, + manifest_id, content_hash, transfer_name, file_count, total_size, + state, created_at, updated_at + FROM targeted_transfers WHERE id = ?1 + "#, + ) + .bind(id) + .fetch_optional(&self.pool) + .await + .map_err(VnidropError::repository)?; + row.map(row_to_full).transpose() + } + + pub(crate) async fn list(&self) -> Result, VnidropError> { + let rows = sqlx::query( + r#" + SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id, + file_count, total_size, state, created_at, updated_at + FROM targeted_transfers + ORDER BY updated_at DESC + "#, + ) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + rows.into_iter().map(row_to_transfer).collect() + } + + #[allow( + dead_code, + reason = "called via cancel_targeted_transfers_for_peer for ticket 09" + )] + pub(crate) async fn cancel_by_peer(&self, peer_endpoint_id: &str) -> Result { + let now = now_ms(); + let result = sqlx::query( + r#" + UPDATE targeted_transfers + SET state = 'cancelled', updated_at = ?2 + WHERE (sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1) + AND state NOT IN ('completed', 'declined', 'cancelled', 'failed', 'deleted') + "#, + ) + .bind(peer_endpoint_id) + .bind(now) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(result.rows_affected()) + } +} + +#[derive(Debug, Clone)] +pub(crate) struct TargetedTransferRow { + pub(crate) id: String, + pub(crate) protocol_transfer_id: u64, + pub(crate) sender_endpoint_id: String, + pub(crate) receiver_endpoint_id: String, + pub(crate) manifest_id: String, + pub(crate) content_hash: String, + pub(crate) transfer_name: String, + pub(crate) file_count: u64, + pub(crate) total_size: u64, + pub(crate) state: TargetedTransferState, + pub(crate) created_at: i64, + pub(crate) updated_at: i64, +} + +fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result { + Ok(TargetedTransfer { + id: row.get("id"), + sender_endpoint_id: row.get("sender_endpoint_id"), + receiver_endpoint_id: row.get("receiver_endpoint_id"), + manifest_id: row.get("manifest_id"), + file_count: row.get::("file_count") as u64, + total_size: row.get::("total_size") as u64, + state: parse_state(&row.get::("state"))?, + created_at: row.get("created_at"), + updated_at: row.get("updated_at"), + }) +} + +fn row_to_full(row: sqlx::sqlite::SqliteRow) -> Result { + Ok(TargetedTransferRow { + id: row.get("id"), + protocol_transfer_id: row.get::("protocol_transfer_id") as u64, + sender_endpoint_id: row.get("sender_endpoint_id"), + receiver_endpoint_id: row.get("receiver_endpoint_id"), + manifest_id: row.get("manifest_id"), + content_hash: row.get("content_hash"), + transfer_name: row.get("transfer_name"), + file_count: row.get::("file_count") as u64, + total_size: row.get::("total_size") as u64, + state: parse_state(&row.get::("state"))?, + created_at: row.get("created_at"), + updated_at: row.get("updated_at"), + }) +} + +pub(crate) fn state_as_str(state: TargetedTransferState) -> &'static str { + match state { + TargetedTransferState::Preparing => "preparing", + TargetedTransferState::Offering => "offering", + TargetedTransferState::AwaitingApproval => "awaiting_approval", + TargetedTransferState::Approved => "approved", + TargetedTransferState::Connecting => "connecting", + TargetedTransferState::Transferring => "transferring", + TargetedTransferState::Interrupted => "interrupted", + TargetedTransferState::Completed => "completed", + TargetedTransferState::Declined => "declined", + TargetedTransferState::Cancelled => "cancelled", + TargetedTransferState::Failed => "failed", + TargetedTransferState::Deleted => "deleted", + } +} + +fn parse_state(value: &str) -> Result { + match value { + "preparing" => Ok(TargetedTransferState::Preparing), + "offering" => Ok(TargetedTransferState::Offering), + "awaiting_approval" => Ok(TargetedTransferState::AwaitingApproval), + "approved" => Ok(TargetedTransferState::Approved), + "connecting" => Ok(TargetedTransferState::Connecting), + "transferring" => Ok(TargetedTransferState::Transferring), + "interrupted" => Ok(TargetedTransferState::Interrupted), + "completed" => Ok(TargetedTransferState::Completed), + "declined" => Ok(TargetedTransferState::Declined), + "cancelled" => Ok(TargetedTransferState::Cancelled), + "failed" => Ok(TargetedTransferState::Failed), + "deleted" => Ok(TargetedTransferState::Deleted), + other => Err(VnidropError::repository(anyhow::anyhow!( + "unknown targeted transfer state: {other}" + ))), + } +} diff --git a/crates/vnidrop/src/targeted_transfer/protocol.rs b/crates/vnidrop/src/targeted_transfer/protocol.rs new file mode 100644 index 0000000..3b09176 --- /dev/null +++ b/crates/vnidrop/src/targeted_transfer/protocol.rs @@ -0,0 +1,297 @@ +//! Targeted-transfer control-plane protocol (design §10). +//! +//! Separate ALPN from ordinary offers: pre-approval messages carry a manifest +//! summary and relationship proof only — never a reusable share ticket. + +use std::fmt; + +use anyhow::Result; +use iroh::{ + endpoint::Connection, + protocol::{AcceptError, ProtocolHandler}, + Endpoint, EndpointAddr, +}; +use irpc::{channel::oneshot, rpc_requests, Client, WithChannels}; +use irpc_iroh::{read_request, IrohLazyRemoteConnection}; +use serde::{Deserialize, Serialize}; + +use super::{ + auth::TargetedAuthorization, + inbox::{TargetedOfferDecision, TargetedOfferInbox}, +}; +use crate::{ + api::{experimental_saved_device_capabilities, PendingTargetedOffer}, + device_relationship::{DeviceRelationshipService, WireProof}, + error::VnidropError, + grant::Challenge, + util::now_ms, +}; + +#[derive(Clone)] +pub(crate) struct TargetedTransferProtocol { + relationships: std::sync::Arc, + inbox: TargetedOfferInbox, + limits: crate::api::CoreLimits, + local_endpoint_id: String, +} + +impl fmt::Debug for TargetedTransferProtocol { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("TargetedTransferProtocol") + } +} + +impl TargetedTransferProtocol { + pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/1"; + + pub(crate) fn new( + relationships: std::sync::Arc, + inbox: TargetedOfferInbox, + limits: crate::api::CoreLimits, + local_endpoint_id: String, + ) -> Self { + Self { + relationships, + inbox, + limits, + local_endpoint_id, + } + } + + pub(crate) fn client(endpoint: Endpoint, addr: EndpointAddr) -> TargetedTransferClient { + TargetedTransferClient { + inner: Client::boxed(IrohLazyRemoteConnection::new( + endpoint, + addr, + Self::ALPN.to_vec(), + )), + } + } + + async fn handle_offer( + &self, + remote_endpoint_id: &str, + challenge: &Challenge, + offer: SubmitTargetedOffer, + ) -> TargetedOfferResponse { + let expected = experimental_saved_device_capabilities().targeted_transfer_protocol_version; + if offer.protocol_version != expected { + return TargetedOfferResponse::Refused { + reason: "protocol-incompatible".to_string(), + }; + } + if offer.receiver_endpoint_id != self.local_endpoint_id { + return TargetedOfferResponse::Refused { + reason: "receiver-mismatch".to_string(), + }; + } + if offer.sender_endpoint_id != remote_endpoint_id { + return TargetedOfferResponse::Refused { + reason: "sender-mismatch".to_string(), + }; + } + if offer.file_count == 0 + || offer.file_count > self.limits.max_collection_files + || offer.total_size == 0 + || offer.total_size > self.limits.max_total_bytes + || offer.transfer_id.is_empty() + || offer.manifest_id.is_empty() + || offer.content_hash.is_empty() + { + return TargetedOfferResponse::Refused { + reason: "manifest-limits".to_string(), + }; + } + if let Err(error) = self + .limits + .validate_metadata_text("transfer name", Some(offer.transfer_name.as_str())) + { + return TargetedOfferResponse::Refused { + reason: error.to_string(), + }; + } + + if let Err(error) = self + .relationships + .verify_saved_possession( + remote_endpoint_id, + challenge, + &offer.proof, + offer.generation, + offer.relationship_protocol_version, + ) + .await + { + tracing::debug!(%error, "targeted offer relationship proof rejected"); + return TargetedOfferResponse::Refused { + reason: "unauthenticated".to_string(), + }; + } + + let pending = PendingTargetedOffer { + transfer_id: offer.transfer_id, + sender_endpoint_id: remote_endpoint_id.to_string(), + receiver_endpoint_id: self.local_endpoint_id.clone(), + manifest_id: offer.manifest_id, + content_hash: offer.content_hash, + transfer_name: offer.transfer_name, + file_count: offer.file_count, + total_size: offer.total_size, + protocol_version: offer.protocol_version, + received_at: now_ms(), + }; + + match self.inbox.submit(pending).await { + TargetedOfferDecision::Accepted => TargetedOfferResponse::Accepted, + TargetedOfferDecision::Declined { reason } => { + TargetedOfferResponse::Declined { reason } + } + TargetedOfferDecision::Refused { reason } => TargetedOfferResponse::Refused { reason }, + } + } + + async fn handle_deliver_authorization( + &self, + remote_endpoint_id: &str, + delivery: DeliverTargetedAuthorization, + ) -> DeliverAuthorizationResponse { + let Ok(auth) = TargetedAuthorization::decode(&delivery.authorization) else { + return DeliverAuthorizationResponse::Rejected; + }; + if auth.sender_endpoint_id != remote_endpoint_id + || auth.receiver_endpoint_id != self.local_endpoint_id + || auth.transfer_id != delivery.transfer_id + { + return DeliverAuthorizationResponse::Rejected; + } + if self + .inbox + .deliver_authorization(&delivery.transfer_id, delivery.authorization) + .await + { + DeliverAuthorizationResponse::Stored + } else { + DeliverAuthorizationResponse::Rejected + } + } +} + +impl ProtocolHandler for TargetedTransferProtocol { + async fn accept(&self, connection: Connection) -> Result<(), AcceptError> { + let remote_endpoint_id = connection.remote_id().to_string(); + let challenge = Challenge::generate(); + while let Some(message) = read_request::(&connection).await? { + match message { + TargetedTransferMessage::RequestChallenge(message) => { + let WithChannels { tx, .. } = message; + let _ = tx + .send(ChallengeResponse { + challenge: challenge.clone(), + }) + .await; + } + TargetedTransferMessage::SubmitTargetedOffer(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self + .handle_offer(&remote_endpoint_id, &challenge, inner) + .await; + let _ = tx.send(response).await; + } + TargetedTransferMessage::DeliverTargetedAuthorization(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self + .handle_deliver_authorization(&remote_endpoint_id, inner) + .await; + let _ = tx.send(response).await; + } + } + } + connection.closed().await; + Ok(()) + } +} + +#[derive(Debug, Clone)] +pub(crate) struct TargetedTransferClient { + inner: Client, +} + +impl TargetedTransferClient { + pub(crate) async fn request_challenge(&self) -> Result { + Ok(self.inner.rpc(RequestChallenge).await?.challenge) + } + + pub(crate) async fn submit_offer( + &self, + offer: SubmitTargetedOffer, + ) -> Result { + self.inner.rpc(offer).await + } + + pub(crate) async fn deliver_authorization( + &self, + delivery: DeliverTargetedAuthorization, + ) -> Result { + self.inner.rpc(delivery).await + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct RequestChallenge; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct ChallengeResponse { + challenge: Challenge, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct SubmitTargetedOffer { + pub(crate) proof: WireProof, + pub(crate) generation: u64, + pub(crate) relationship_protocol_version: u16, + pub(crate) protocol_version: u16, + pub(crate) transfer_id: String, + pub(crate) sender_endpoint_id: String, + pub(crate) receiver_endpoint_id: String, + pub(crate) manifest_id: String, + pub(crate) content_hash: String, + pub(crate) transfer_name: String, + pub(crate) file_count: u64, + pub(crate) total_size: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum TargetedOfferResponse { + Accepted, + Declined { reason: String }, + Refused { reason: String }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct DeliverTargetedAuthorization { + pub(crate) transfer_id: String, + pub(crate) authorization: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum DeliverAuthorizationResponse { + Stored, + Rejected, +} + +#[rpc_requests(message = TargetedTransferMessage)] +#[derive(Debug, Serialize, Deserialize)] +enum TargetedTransferMessages { + #[rpc(tx = oneshot::Sender)] + RequestChallenge(RequestChallenge), + #[rpc(tx = oneshot::Sender)] + SubmitTargetedOffer(SubmitTargetedOffer), + #[rpc(tx = oneshot::Sender)] + DeliverTargetedAuthorization(DeliverTargetedAuthorization), +} + +/// Helper kept for type visibility in callers that map refuse reasons. +#[allow(dead_code)] +pub(crate) fn map_offer_error(reason: &str) -> VnidropError { + VnidropError::permission(anyhow::anyhow!("targeted offer refused: {reason}")) +} diff --git a/crates/vnidrop/src/targeted_transfer.rs b/crates/vnidrop/src/targeted_transfer/state.rs similarity index 100% rename from crates/vnidrop/src/targeted_transfer.rs rename to crates/vnidrop/src/targeted_transfer/state.rs diff --git a/crates/vnidrop/src/tests.rs b/crates/vnidrop/src/tests.rs index 4ef6972..764810f 100644 --- a/crates/vnidrop/src/tests.rs +++ b/crates/vnidrop/src/tests.rs @@ -38,6 +38,8 @@ mod secure_secret_tests; #[cfg(target_os = "windows")] #[path = "tests/secure_secret_windows.rs"] mod secure_secret_windows_tests; +#[path = "tests/targeted_transfer.rs"] +mod targeted_transfer_tests; #[path = "tests/ticket.rs"] mod ticket_tests; #[path = "tests/transfer_state.rs"] diff --git a/crates/vnidrop/src/tests/targeted_transfer.rs b/crates/vnidrop/src/tests/targeted_transfer.rs new file mode 100644 index 0000000..ed79746 --- /dev/null +++ b/crates/vnidrop/src/tests/targeted_transfer.rs @@ -0,0 +1,471 @@ +use std::{ + path::Path, + sync::{Arc, Mutex}, + time::{Duration, Instant}, +}; + +use crate::{ + secure_secret::FaultInjectingSecretStore, CoreEvent, CoreEventSink, DeviceRelationshipState, + PendingTargetedOffer, ShareMetadataInput, ShareSource, SourceKind, TargetedTransferState, + TransferAccessMode, VnidropCore, VnidropError, +}; + +struct RecordingSink { + events: Mutex>, +} + +impl CoreEventSink for RecordingSink { + fn on_event(&self, event: CoreEvent) { + self.events.lock().unwrap().push(event); + } +} + +struct ProtectedNode { + _data_dir: tempfile::TempDir, + core: Arc, +} + +impl ProtectedNode { + fn new() -> Self { + let data_dir = tempfile::tempdir().unwrap(); + let sink = Arc::new(RecordingSink { + events: Mutex::new(Vec::new()), + }); + let store = Arc::new(FaultInjectingSecretStore::default()); + let core = VnidropCore::initialize_with_test_secret_store( + data_dir.path().to_string_lossy().into_owned(), + sink, + store, + ) + .expect("protected test core"); + Self { + _data_dir: data_dir, + core, + } + } +} + +impl Drop for ProtectedNode { + fn drop(&mut self) { + self.core.shutdown(); + } +} + +fn share_path(core: &VnidropCore, source: &Path, transfer_id: u64) -> crate::ShareResult { + core.share_files( + vec![ShareSource { + kind: SourceKind::Path, + value: source.to_string_lossy().into_owned(), + display_name: Some("hello.txt".to_string()), + is_directory: false, + }], + ShareMetadataInput { + transfer_id, + transfer_name: Some("hello.txt".to_string()), + sender_name: Some("sender".to_string()), + access_mode: TransferAccessMode::ApprovalRequired, + }, + ) + .unwrap() +} + +fn wait_for_receiver_request(sender: &VnidropCore, transfer_id: u64) -> crate::ReceiverRequest { + let started = Instant::now(); + loop { + if let Some(request) = sender + .list_receiver_requests(transfer_id) + .unwrap() + .into_iter() + .find(|request| request.status == "requested") + { + return request; + } + assert!( + started.elapsed() < Duration::from_secs(15), + "timed out waiting for receiver request" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +fn complete_transfer(sender: &ProtectedNode, receiver: &ProtectedNode, transfer_id: u64) { + let source_dir = tempfile::tempdir().unwrap(); + 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 output_dir = output_dir.path().to_string_lossy().to_string(); + 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); + sender + .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(); + loop { + if sender + .core + .list_pairing_eligibilities() + .unwrap() + .iter() + .any(|entry| entry.peer_endpoint_id == peer) + { + break; + } + assert!( + started.elapsed() < Duration::from_secs(10), + "eligibility never appeared" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +fn wait_for_relationship( + core: &VnidropCore, + peer: &str, + state: DeviceRelationshipState, +) -> crate::DeviceRelationship { + let started = Instant::now(); + loop { + if let Some(relationship) = core + .list_device_relationships() + .unwrap() + .into_iter() + .find(|entry| entry.remote_endpoint_id == peer && entry.state == state) + { + return relationship; + } + assert!( + started.elapsed() < Duration::from_secs(15), + "relationship {peer} never reached {state:?}" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +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(); + complete_transfer(alice, bob, transfer_id); + assert!(alice + .core + .request_saved_device_pairing(bob_id.clone()) + .unwrap()); + wait_for_relationship( + &bob.core, + &alice_id, + DeviceRelationshipState::PendingIncoming, + ); + assert!(bob + .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); +} + +fn wait_for_pending_offer(core: &VnidropCore) -> PendingTargetedOffer { + let started = Instant::now(); + loop { + let pending = core.list_pending_targeted_offers(); + if let Some(offer) = pending.into_iter().next() { + return offer; + } + assert!( + started.elapsed() < Duration::from_secs(20), + "timed out waiting for pending targeted offer" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +fn targeted_source(path: &Path) -> ShareSource { + ShareSource { + kind: SourceKind::Path, + value: path.to_string_lossy().into_owned(), + display_name: Some("payload.txt".to_string()), + is_directory: false, + } +} + +#[test] +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(); + establish_saved(&alice, &bob, 10_001); + + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("payload.txt"); + std::fs::write(&source_path, b"immutable payload").unwrap(); + + let stranger_err = alice + .core + .create_targeted_transfer( + stranger_id, + vec![targeted_source(&source_path)], + Some("payload.txt".to_string()), + ) + .unwrap_err(); + assert!(matches!(stranger_err, VnidropError::Permission { .. })); + + 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.clone(), + vec![targeted_source(&source_path)], + Some("payload.txt".to_string()), + ) + .unwrap(); + let _auth = accept.join().unwrap().expect("authorization after approve"); + + 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); + assert!(!transfer.id.is_empty()); + assert!(!transfer.manifest_id.is_empty()); + assert!(matches!( + transfer.state, + TargetedTransferState::Approved + | TargetedTransferState::Connecting + | TargetedTransferState::Transferring + | TargetedTransferState::Completed + )); + + let listed = alice + .core + .get_targeted_transfer(transfer.id.clone()) + .unwrap(); + let listed = listed.expect("durable targeted transfer"); + assert_eq!(listed.id, transfer.id); + assert_eq!(listed.sender_endpoint_id, transfer.sender_endpoint_id); + assert_eq!(listed.receiver_endpoint_id, transfer.receiver_endpoint_id); + assert_eq!(listed.manifest_id, transfer.manifest_id); + assert_eq!(listed.file_count, transfer.file_count); + assert_eq!(listed.total_size, transfer.total_size); +} + +#[test] +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(); + 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 accept = std::thread::spawn(move || { + let offer = wait_for_pending_offer(&bob_core); + // Offer surfaces identity + manifest summary only — never a reusable ticket. + assert!(!offer.transfer_id.is_empty()); + assert_eq!(offer.sender_endpoint_id, alice_id_from(&bob_core)); + assert_eq!(offer.file_count, 1); + assert_eq!(offer.total_size, b"offer body".len() as u64); + assert!(!offer.manifest_id.is_empty()); + assert!(!offer.content_hash.is_empty()); + assert!(offer.protocol_version >= 1); + bob_core + .respond_to_targeted_offer(offer.transfer_id, true) + .unwrap() + }); + + alice + .core + .create_targeted_transfer( + bob_id, + vec![targeted_source(&source_path)], + Some("payload.txt".to_string()), + ) + .unwrap(); + accept.join().unwrap().unwrap(); +} + +fn alice_id_from(bob: &VnidropCore) -> String { + bob.list_saved_devices() + .unwrap() + .into_iter() + .next() + .expect("saved alice") + .endpoint_id +} + +#[test] +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 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 + .create_targeted_transfer( + bob_id, + vec![targeted_source(&source_path)], + Some("payload.txt".to_string()), + ) + .unwrap_err(); + assert!(matches!(err, VnidropError::Permission { .. })); + assert!(bob.core.list_pending_targeted_offers().is_empty()); +} + +#[test] +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(); + establish_saved(&alice, &bob, 10_020); + + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("payload.txt"); + let payload = b"bound authorization payload"; + std::fs::write(&source_path, payload).unwrap(); + + 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)); + let gate_wait = gate.clone(); + let accept = std::thread::spawn(move || { + let offer = wait_for_pending_offer(&bob_core); + *offer_id_setter.lock().unwrap() = Some(offer.transfer_id.clone()); + // Hold approval until the main thread asserts that content is unavailable. + let started = Instant::now(); + loop { + if *gate_wait.lock().unwrap() { + break; + } + assert!( + started.elapsed() < Duration::from_secs(20), + "approval gate never opened" + ); + std::thread::sleep(Duration::from_millis(25)); + } + bob_core + .respond_to_targeted_offer(offer.transfer_id, true) + .unwrap() + }); + + 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 started = Instant::now(); + let transfer_id = loop { + if let Some(id) = offer_id.lock().unwrap().clone() { + break id; + } + assert!( + started.elapsed() < Duration::from_secs(20), + "offer id never observed" + ); + std::thread::sleep(Duration::from_millis(25)); + }; + + let _ = transfer_id; + // Without an approved authorization, receive must fail — no content yet. + let early_receive = bob.core.receive_targeted_transfer( + "not-a-real-authorization".to_string(), + tempfile::tempdir() + .unwrap() + .path() + .to_string_lossy() + .into_owned(), + ); + assert!(early_receive.is_err()); + + *gate.lock().unwrap() = true; + let auth = accept.join().unwrap().expect("receiver authorization"); + create.join().unwrap().unwrap(); + + let output = tempfile::tempdir().unwrap(); + bob.core + .receive_targeted_transfer(auth.clone(), output.path().to_string_lossy().into_owned()) + .unwrap(); + assert_eq!( + std::fs::read(output.path().join("payload.txt")).unwrap(), + payload + ); + + let charlie_output = tempfile::tempdir().unwrap(); + let leaked = charlie + .core + .receive_targeted_transfer(auth, charlie_output.path().to_string_lossy().into_owned()); + assert!( + leaked.is_err(), + "leaked authorization must not authorize another endpoint" + ); +} + +#[test] +fn invitation_multi_receiver_shares_remain_independently_authorized() { + let source_dir = tempfile::tempdir().unwrap(); + let first_output = tempfile::tempdir().unwrap(); + let second_output = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"shared with both receivers").unwrap(); + let sender = ProtectedNode::new(); + let first_receiver = ProtectedNode::new(); + let second_receiver = ProtectedNode::new(); + let share = sender + .core + .share_files( + vec![ShareSource { + kind: SourceKind::Path, + value: source_path.to_string_lossy().into_owned(), + display_name: Some("shared.txt".to_string()), + is_directory: false, + }], + ShareMetadataInput { + transfer_id: 10_030, + transfer_name: Some("Existing share".to_string()), + sender_name: Some("Sender".to_string()), + access_mode: TransferAccessMode::Public, + }, + ) + .unwrap(); + + for (receiver, output) in [ + (&first_receiver, first_output.path()), + (&second_receiver, second_output.path()), + ] { + receiver + .core + .receive( + share.ticket.clone(), + output.to_string_lossy().into_owned(), + Some("Receiver".to_string()), + ) + .unwrap(); + assert_eq!( + std::fs::read(output.join("shared.txt")).unwrap(), + b"shared with both receivers" + ); + } +}