diff --git a/crates/vnidrop/src/access_policy.rs b/crates/vnidrop/src/access_policy.rs index eb9d7fb..d30eb09 100644 --- a/crates/vnidrop/src/access_policy.rs +++ b/crates/vnidrop/src/access_policy.rs @@ -29,13 +29,6 @@ impl AccessPolicy { self.modes.write().await.insert(transfer_id, mode); } - pub(crate) async fn allows_without_approval(&self, transfer_id: u64) -> bool { - matches!( - self.modes.read().await.get(&transfer_id), - Some(TransferAccessMode::Public) - ) - } - pub(crate) async fn remove_transfer(&self, transfer_id: u64) { self.modes.write().await.remove(&transfer_id); self.approved_sessions diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index 2e87d02..b2470ed 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -477,6 +477,21 @@ pub struct ContactSummary { pub can_send: bool, } +/// A transfer a paired device is offering. +/// +/// The ticket is deliberately absent: it is a capability, and it is handed over +/// only when the user accepts. +#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] +pub struct IncomingOffer { + pub offer_id: String, + pub from_endpoint_id: String, + pub sender_display_name: Option, + pub transfer_name: String, + pub file_count: u64, + pub total_bytes: u64, + pub received_at: i64, +} + /// A device offering to be remembered, awaiting the local user's decision. #[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] pub struct PendingPairing { diff --git a/crates/vnidrop/src/approval.rs b/crates/vnidrop/src/approval.rs index c0f273d..d310409 100644 --- a/crates/vnidrop/src/approval.rs +++ b/crates/vnidrop/src/approval.rs @@ -6,7 +6,7 @@ use tokio::sync::{oneshot, Mutex}; use uuid::Uuid; use crate::{ - access_policy::{AccessPolicy, APPROVAL_SESSION_TTL_MS}, + access_policy::{AccessDecision, AccessPolicy, APPROVAL_SESSION_TTL_MS}, event_hub::EventHub, handshake::{ DeliveryFailureReceipt, DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, @@ -198,10 +198,15 @@ impl ApprovalService { .await { Ok(true) => { + // An existing access session means this endpoint was already + // authorised: either the share is public, or the sender pushed + // this transfer to them. Prompting again would ask the sender + // to approve a transfer they themselves initiated. if self .access_policy - .allows_without_approval(request.transfer_id) + .decide(request.transfer_id, Some(&remote_endpoint_id)) .await + == AccessDecision::Allow { self.allow_without_sender_decision(remote_endpoint_id, request) .await diff --git a/crates/vnidrop/src/contacts.rs b/crates/vnidrop/src/contacts.rs index 1baea1b..61af7b3 100644 --- a/crates/vnidrop/src/contacts.rs +++ b/crates/vnidrop/src/contacts.rs @@ -8,15 +8,18 @@ //! tickets: never logged, never emitted in an event, never returned across the //! UniFFI boundary. -// A few operations are exercised only by unit tests until send-to-contact -// offers consume them. Remove this once that lands. -#![allow(dead_code)] - use anyhow::{Context, Result}; use sqlx::{Row, SqlitePool}; use crate::grant::{parse_secret, GrantId, HeldGrant, IssuedGrant}; +/// How long a dead grant is kept before being swept. +/// +/// A revoked grant stays as a tombstone so a returning peer is told `Revoked` +/// rather than `Unknown`; after this long, a peer that has not come back is +/// unlikely to, and the row is noise. +pub(crate) const DEAD_GRANT_RETENTION_MS: i64 = 30 * 24 * 60 * 60 * 1_000; + /// A device the user has transferred with and chosen to remember. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct Contact { diff --git a/crates/vnidrop/src/grant.rs b/crates/vnidrop/src/grant.rs index 68341fd..a79b70a 100644 --- a/crates/vnidrop/src/grant.rs +++ b/crates/vnidrop/src/grant.rs @@ -9,10 +9,6 @@ //! This module is pure: no storage, no network, no clock of its own. Callers //! supply `now_ms` so expiry and renewal stay testable. -// A few helpers are exercised only by unit tests until send-to-contact offers -// consume them. Remove this once that lands. -#![allow(dead_code)] - use std::fmt; use anyhow::{bail, Context, Result}; @@ -230,10 +226,6 @@ impl IssuedGrant { self.expires_at .is_some_and(|expires_at| expires_at < now_ms) } - - pub(crate) fn is_active(&self, now_ms: i64) -> bool { - self.revoked_at.is_none() && !self.is_expired(now_ms) - } } /// A grant as held by the party it was issued to: the capability used to reach diff --git a/crates/vnidrop/src/lib.rs b/crates/vnidrop/src/lib.rs index da4ee95..5937e85 100644 --- a/crates/vnidrop/src/lib.rs +++ b/crates/vnidrop/src/lib.rs @@ -9,6 +9,7 @@ mod grant; mod handshake; mod logging; mod offer; +mod offer_inbox; mod pairing; mod repository; mod runtime; @@ -20,10 +21,10 @@ mod util; pub use api::{ clear_inactive_transfer_cache, default_core_limits, default_core_network_config, ContactSummary, CoreEvent, CoreEventSink, CoreLimits, CoreNetworkConfig, CoreRelayMode, - CoreStorageUsage, GrantLifetimeSetting, PendingPairing, PublishedOutput, ReceiveOutputSink, - ReceiveOutputSinkV2, ReceivedArtifact, ReceivedLocatorKind, ReceiverRequest, RuntimeStatus, - ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer, TicketInspection, - TransferAccessMode, TransferMetadata, + CoreStorageUsage, GrantLifetimeSetting, IncomingOffer, PendingPairing, PublishedOutput, + ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedArtifact, ReceivedLocatorKind, ReceiverRequest, + RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer, + TicketInspection, TransferAccessMode, TransferMetadata, }; pub use error::VnidropError; pub use runtime::VnidropCore; diff --git a/crates/vnidrop/src/offer.rs b/crates/vnidrop/src/offer.rs index 1c4f601..44d902c 100644 --- a/crates/vnidrop/src/offer.rs +++ b/crates/vnidrop/src/offer.rs @@ -22,13 +22,18 @@ use irpc_iroh::{read_request, IrohLazyRemoteConnection}; use serde::{Deserialize, Serialize}; use crate::{ - grant::{Challenge, GrantId}, + grant::{Challenge, GrantId, GrantProof}, + offer_inbox::OfferInbox, pairing::PairingService, }; #[derive(Clone)] pub(crate) struct OfferService { pairing: PairingService, + inbox: OfferInbox, + /// This device's endpoint id. Grants we issued are bound to it, so proofs + /// must be verified against it rather than against whatever a peer claims. + self_endpoint_id: String, } impl fmt::Debug for OfferService { @@ -40,8 +45,16 @@ impl fmt::Debug for OfferService { impl OfferService { pub(crate) const ALPN: &'static [u8] = b"/vnidrop/offer/1"; - pub(crate) fn new(pairing: PairingService) -> Self { - Self { pairing } + pub(crate) fn new( + pairing: PairingService, + inbox: OfferInbox, + self_endpoint_id: String, + ) -> Self { + Self { + pairing, + inbox, + self_endpoint_id, + } } pub(crate) fn client(endpoint: Endpoint, addr: EndpointAddr) -> OfferClient { @@ -55,6 +68,46 @@ impl OfferService { } } +impl OfferService { + /// Validate the grant, then hand the offer to the local user. + /// + /// A refusal names the grant failure so the peer can drop a dead entry; + /// `Unknown` covers both "never issued" and "blocked", which is what keeps + /// blocking undetectable. + async fn handle_offer( + &self, + remote_endpoint_id: &str, + challenge: &Challenge, + offer: SubmitOffer, + ) -> OfferResponse { + if let Err(rejection) = self + .pairing + .verify_and_renew( + &offer.proof, + challenge, + &self.self_endpoint_id, + remote_endpoint_id, + ) + .await + { + return OfferResponse::Refused { + reason: rejection.as_str().to_string(), + }; + } + + self.inbox + .submit( + remote_endpoint_id.to_string(), + offer.transfer_name, + offer.sender_display_name, + offer.file_count, + offer.total_bytes, + offer.ticket, + ) + .await + } +} + impl ProtocolHandler for OfferService { /// Accepts inbound connections from paired peers. /// @@ -91,6 +144,13 @@ impl ProtocolHandler for OfferService { .await; let _ = tx.send(response).await; } + OfferMessage::SubmitOffer(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self + .handle_offer(&remote_endpoint_id, &challenge, inner) + .await; + let _ = tx.send(response).await; + } } } @@ -105,6 +165,17 @@ pub(crate) struct OfferClient { } impl OfferClient { + pub(crate) async fn request_challenge(&self) -> Result { + Ok(self.inner.rpc(RequestChallenge).await?.challenge) + } + + pub(crate) async fn submit_offer( + &self, + offer: SubmitOffer, + ) -> Result { + self.inner.rpc(offer).await + } + pub(crate) async fn deliver_grant( &self, grant: DeliverGrant, @@ -180,6 +251,31 @@ pub(crate) enum RevocationResponse { Unknown, } +/// Hand a paired device a ticket for content it may fetch. +/// +/// The ticket is a capability, so this is sent only over a connection where the +/// grant proof has already been presented. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct SubmitOffer { + pub(crate) proof: GrantProof, + pub(crate) ticket: String, + pub(crate) transfer_name: String, + pub(crate) sender_display_name: Option, + pub(crate) file_count: u64, + pub(crate) total_bytes: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum OfferResponse { + /// The receiving user agreed. They fetch the content themselves next. + Accepted, + /// The receiving user said no, or never answered. + Declined { reason: String }, + /// The grant did not validate. Names the reason so a peer holding a dead + /// grant can clear it. + Refused { reason: String }, +} + #[rpc_requests(message = OfferMessage)] #[derive(Debug, Serialize, Deserialize)] enum OfferProtocol { @@ -189,4 +285,6 @@ enum OfferProtocol { DeliverGrant(DeliverGrant), #[rpc(tx=oneshot::Sender)] RevokeGrant(RevokeGrant), + #[rpc(tx=oneshot::Sender)] + SubmitOffer(SubmitOffer), } diff --git a/crates/vnidrop/src/offer_inbox.rs b/crates/vnidrop/src/offer_inbox.rs new file mode 100644 index 0000000..58071be --- /dev/null +++ b/crates/vnidrop/src/offer_inbox.rs @@ -0,0 +1,222 @@ +//! Incoming transfer offers from paired devices. +//! +//! An offer is only a delivery mechanism for a ticket: it replaces the QR code, +//! not the transfer. Accepting hands the ticket to the platform layer, which +//! runs the ordinary receive with its own destination rules. +//! +//! Nothing here is persisted. An offer is a live connection to a running app, +//! so a restart correctly loses it rather than resurrecting a prompt whose +//! sender is long gone. + +use std::{collections::HashMap, sync::Arc, time::Duration}; + +use serde_json::json; +use tokio::sync::{oneshot, Mutex}; +use uuid::Uuid; + +use crate::{event_hub::EventHub, offer::OfferResponse, util::now_ms}; + +/// How long the sender waits for the receiving user to decide. +const OFFER_WAIT_TIMEOUT: Duration = Duration::from_secs(120); + +/// Quiet period after a decline, so a paired device cannot re-prompt on a loop. +/// A contact is not a stranger, but it is not unlimited either. +const DECLINE_COOLDOWN_MS: i64 = 60 * 1_000; + +#[derive(Debug, Clone)] +pub(crate) struct PendingOffer { + pub(crate) offer_id: String, + pub(crate) from_endpoint_id: String, + pub(crate) sender_display_name: Option, + pub(crate) transfer_name: String, + pub(crate) file_count: u64, + pub(crate) total_bytes: u64, + pub(crate) received_at: i64, + /// Released to the caller only once the local user accepts. + ticket: String, +} + +struct Waiter { + endpoint_id: String, + responder: oneshot::Sender, +} + +#[derive(Clone)] +pub(crate) struct OfferInbox { + event_hub: Arc, + pending: Arc>>, + waiters: Arc>>, + /// Endpoint → time before which new offers are refused. + cooldowns: Arc>>, + max_pending: usize, +} + +impl OfferInbox { + pub(crate) fn new(event_hub: Arc, max_pending: usize) -> Self { + Self { + event_hub, + pending: Arc::new(Mutex::new(HashMap::new())), + waiters: Arc::new(Mutex::new(HashMap::new())), + cooldowns: Arc::new(Mutex::new(HashMap::new())), + max_pending, + } + } + + /// Surface an offer and block until the local user decides. + /// + /// The caller has already proven a live grant, so this is a known device; + /// the limits here bound nuisance rather than attack. + pub(crate) async fn submit( + &self, + from_endpoint_id: String, + transfer_name: String, + sender_display_name: Option, + file_count: u64, + total_bytes: u64, + ticket: String, + ) -> OfferResponse { + let now = now_ms(); + { + let mut cooldowns = self.cooldowns.lock().await; + cooldowns.retain(|_, until| *until > now); + if cooldowns.contains_key(&from_endpoint_id) { + return OfferResponse::Declined { + reason: "declined-recently".to_string(), + }; + } + } + + let offer_id = Uuid::new_v4().to_string(); + let (tx, rx) = oneshot::channel(); + { + let mut pending = self.pending.lock().await; + if pending.len() >= self.max_pending { + return OfferResponse::Declined { + reason: "too-many-pending-offers".to_string(), + }; + } + // One prompt per device at a time: a second offer would stack + // notifications for the same sender. + if pending + .values() + .any(|offer| offer.from_endpoint_id == from_endpoint_id) + { + return OfferResponse::Declined { + reason: "offer-already-pending".to_string(), + }; + } + pending.insert( + offer_id.clone(), + PendingOffer { + offer_id: offer_id.clone(), + from_endpoint_id: from_endpoint_id.clone(), + sender_display_name: sender_display_name.clone(), + transfer_name: transfer_name.clone(), + file_count, + total_bytes, + received_at: now, + ticket, + }, + ); + } + self.waiters.lock().await.insert( + offer_id.clone(), + Waiter { + endpoint_id: from_endpoint_id.clone(), + responder: tx, + }, + ); + + // The ticket is deliberately absent: an event is a log record, and a + // ticket is a capability. + self.event_hub.emit_endpoint( + "offer", + "offer-received", + json!({ + "offer_id": offer_id, + "from_endpoint_id": from_endpoint_id, + "sender_display_name": sender_display_name, + "transfer_name": transfer_name, + "file_count": file_count, + "total_bytes": total_bytes, + }), + ); + + match tokio::time::timeout(OFFER_WAIT_TIMEOUT, rx).await { + Ok(Ok(true)) => OfferResponse::Accepted, + Ok(Ok(false)) => OfferResponse::Declined { + reason: "receiver-declined".to_string(), + }, + // Dropped responder or timeout: clear the prompt so it cannot + // linger after the sender has given up. + Ok(Err(_)) | Err(_) => { + self.discard(&offer_id).await; + OfferResponse::Declined { + reason: "no-response".to_string(), + } + } + } + } + + pub(crate) async fn list(&self) -> Vec { + self.pending.lock().await.values().cloned().collect() + } + + /// Record the local user's decision. + /// + /// Returns the ticket on acceptance: it leaves the core at the moment of + /// consent and not before, so a declined offer never hands over a + /// capability. The caller then runs the ordinary receive with it. + pub(crate) async fn respond(&self, offer_id: &str, accepted: bool) -> Option { + let offer = self.pending.lock().await.remove(offer_id)?; + let waiter = self.waiters.lock().await.remove(offer_id); + + if !accepted { + self.cooldowns.lock().await.insert( + offer.from_endpoint_id.clone(), + now_ms() + DECLINE_COOLDOWN_MS, + ); + } + if let Some(waiter) = waiter { + let _ = waiter.responder.send(accepted); + } + self.event_hub.emit_endpoint( + "offer", + if accepted { + "offer-accepted" + } else { + "offer-declined" + }, + json!({ + "offer_id": offer_id, + "from_endpoint_id": offer.from_endpoint_id, + }), + ); + + accepted.then_some(offer.ticket) + } + + /// Drop every prompt from a device, used when it is forgotten or blocked + /// while an offer is on screen. + pub(crate) async fn discard_from(&self, endpoint_id: &str) { + let ids: Vec = { + let pending = self.pending.lock().await; + pending + .values() + .filter(|offer| offer.from_endpoint_id == endpoint_id) + .map(|offer| offer.offer_id.clone()) + .collect() + }; + for offer_id in ids { + self.discard(&offer_id).await; + } + } + + async fn discard(&self, offer_id: &str) { + self.pending.lock().await.remove(offer_id); + if let Some(waiter) = self.waiters.lock().await.remove(offer_id) { + let _ = waiter.responder.send(false); + let _ = waiter.endpoint_id; + } + } +} diff --git a/crates/vnidrop/src/pairing.rs b/crates/vnidrop/src/pairing.rs index cb5f352..71e2749 100644 --- a/crates/vnidrop/src/pairing.rs +++ b/crates/vnidrop/src/pairing.rs @@ -13,7 +13,9 @@ use tokio::sync::Mutex; use crate::{ contacts::ContactStore, event_hub::EventHub, - grant::{GrantLifetime, GrantSecret, HeldGrant, IssuedGrant}, + grant::{ + Challenge, GrantLifetime, GrantProof, GrantRejection, GrantSecret, HeldGrant, IssuedGrant, + }, offer::{DeliverGrant, GrantDeliveryResponse, RevocationResponse, RevokeGrant}, util::now_ms, }; @@ -288,6 +290,55 @@ impl PairingService { Ok(grant) } + /// Validate a proof a peer presented, and push the idle deadline forward. + /// + /// The grant record is ours: we issued it, so we are the only party that + /// can decide it is still alive. A blocked endpoint is answered `Unknown`, + /// the same as one we never issued to. + pub(crate) async fn verify_and_renew( + &self, + proof: &GrantProof, + challenge: &Challenge, + issuer_endpoint_id: &str, + remote_endpoint_id: &str, + ) -> Result<(), GrantRejection> { + if self + .contacts + .is_blocked(remote_endpoint_id) + .await + .unwrap_or(false) + { + return Err(GrantRejection::Unknown); + } + let grant = self + .contacts + .find_issued_grant(proof.grant_id) + .await + .map_err(|_| GrantRejection::Unknown)? + .ok_or(GrantRejection::Unknown)?; + + let now = now_ms(); + let lifetime = self.grant_lifetime().await; + let renewed = grant.accept( + proof, + challenge, + issuer_endpoint_id, + remote_endpoint_id, + now, + lifetime, + )?; + // A failed renewal is not grounds to refuse a peer that just proved + // possession; the grant stays valid until its existing deadline. + if let Err(error) = self + .contacts + .renew_issued_grant(proof.grant_id, renewed) + .await + { + tracing::warn!(%error, "failed to renew grant deadline"); + } + Ok(()) + } + fn drop_expired(&self, pending: &mut HashMap, now_ms: i64) { let window = CONSENT_WINDOW.as_millis() as i64; pending.retain(|_, entry| now_ms - entry.received_at < window); diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index e1d46ea..00e0cd7 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -326,8 +326,6 @@ impl Repository { /// Device history, grants, and the block list. Shares this pool so the /// tables migrate together with the rest of the schema. - // Reached from tests until the offer protocol lands. - #[allow(dead_code)] pub(crate) fn contacts(&self) -> ContactStore { ContactStore::new(self.pool.clone()) } diff --git a/crates/vnidrop/src/runtime/contacts.rs b/crates/vnidrop/src/runtime/contacts.rs index cd736dc..3891101 100644 --- a/crates/vnidrop/src/runtime/contacts.rs +++ b/crates/vnidrop/src/runtime/contacts.rs @@ -12,10 +12,15 @@ use serde_json::json; use super::CoreInner; use crate::{ - api::{ContactSummary, GrantLifetimeSetting, PendingPairing}, + api::{ + ContactSummary, GrantLifetimeSetting, IncomingOffer, PendingPairing, ShareMetadataInput, + ShareResult, ShareSource, TransferAccessMode, + }, error::VnidropError, - grant::GrantId, - offer::{DeliverGrant, GrantDeliveryResponse, OfferService, RevokeGrant}, + grant::{GrantId, HeldGrant}, + offer::{ + DeliverGrant, GrantDeliveryResponse, OfferResponse, OfferService, RevokeGrant, SubmitOffer, + }, ticket::{encode_persisted_sender_address, parse_persisted_sender_address}, util::now_ms, }; @@ -136,6 +141,150 @@ impl CoreInner { } } + /// Share content and push the ticket straight to a paired device. + /// + /// Two things make this one prompt rather than two: the share is created + /// with the ticket never leaving this device except over the authenticated + /// offer connection, and the target endpoint is pre-authorised so the + /// handshake it runs next does not ask us to approve a transfer we started. + pub(super) async fn send_to_contact( + self: &Arc, + endpoint_id: String, + sources: Vec, + mut metadata: ShareMetadataInput, + ) -> Result { + let store = self.repository.contacts(); + let grant = store + .held_grant_for(&endpoint_id) + .await + .map_err(VnidropError::repository)? + .ok_or_else(|| { + VnidropError::permission(anyhow::anyhow!( + "no live grant for this device; pair with it again" + )) + })?; + + // Invariant: an offer-created share is never public. The recipient is a + // specific device, so serving it to anyone holding the ticket would + // widen access beyond what the user asked for. + metadata.access_mode = TransferAccessMode::ApprovalRequired; + let sender_name = metadata.sender_name.clone(); + let share = self.share_files(sources, metadata).await?; + + let outcome = self + .deliver_offer(&endpoint_id, &grant, &share, sender_name.as_deref()) + .await + .inspect_err(|_| { + // Nothing to serve if the offer never landed. + let inner = self.clone(); + let transfer_id = share.transfer_id; + tokio::spawn(async move { + let _ = inner.cancel_idle_or_share(transfer_id).await; + }); + })?; + + match outcome { + OfferResponse::Accepted => { + store + .touch_transfer(&endpoint_id, now_ms()) + .await + .map_err(VnidropError::repository)?; + self.remember_addr(&endpoint_id).await; + self.emit_transfer( + share.transfer_id, + "send", + "offer", + "offer-accepted", + json!({ "peer_endpoint_id": endpoint_id }), + ); + Ok(share) + } + OfferResponse::Declined { reason } | OfferResponse::Refused { reason } => { + let _ = self.cancel_idle_or_share(share.transfer_id).await; + self.emit_transfer( + share.transfer_id, + "send", + "offer", + "offer-refused", + json!({ "peer_endpoint_id": endpoint_id, "reason": reason }), + ); + // A refusal naming a dead grant is the peer telling us to stop + // believing we can reach them. + if matches!(reason.as_str(), "revoked" | "unknown" | "expired") { + let _ = store.delete_held_grant(grant.grant_id).await; + } + Err(VnidropError::permission(anyhow::anyhow!( + "device did not accept the transfer: {reason}" + )) + .into()) + } + } + } + + async fn deliver_offer( + self: &Arc, + endpoint_id: &str, + grant: &HeldGrant, + share: &ShareResult, + sender_name: Option<&str>, + ) -> Result { + let addr = self.contact_addr(endpoint_id).await?; + let client = OfferService::client(self.endpoint.clone(), addr); + let challenge = client + .request_challenge() + .await + .context("device is not reachable") + .map_err(VnidropError::transfer)?; + + // Authorise before offering: the receiver may dial back the instant it + // accepts, and an unauthorised endpoint would be refused by the + // provider. + self.access_policy + .approve_endpoint(share.transfer_id, endpoint_id.to_string()) + .await; + + client + .submit_offer(SubmitOffer { + proof: grant.prove(&challenge, &self.endpoint.id().to_string()), + ticket: share.ticket.clone(), + transfer_name: share.transfer_name.clone(), + sender_display_name: sender_name.map(ToOwned::to_owned), + file_count: share.file_count, + total_bytes: share.total_size, + }) + .await + .context("failed to deliver the offer") + .map_err(VnidropError::transfer) + .map_err(Into::into) + } + + pub(super) async fn list_pending_offers(&self) -> Vec { + self.offers + .list() + .await + .into_iter() + .map(|offer| IncomingOffer { + offer_id: offer.offer_id, + from_endpoint_id: offer.from_endpoint_id, + sender_display_name: offer.sender_display_name, + transfer_name: offer.transfer_name, + file_count: offer.file_count, + total_bytes: offer.total_bytes, + received_at: offer.received_at, + }) + .collect() + } + + /// Answer an incoming offer. Returns the ticket when accepted, so the + /// platform layer can run the ordinary receive with its own destination. + pub(super) async fn respond_to_offer( + &self, + offer_id: String, + accepted: bool, + ) -> Option { + self.offers.respond(&offer_id, accepted).await + } + pub(super) async fn respond_to_pairing( &self, endpoint_id: String, @@ -162,6 +311,9 @@ impl CoreInner { .delete_contact(&endpoint_id) .await .map_err(VnidropError::repository)?; + // A prompt on screen from a device we just forgot would be actionable + // with a grant that no longer exists. + self.offers.discard_from(&endpoint_id).await; self.emit_endpoint( "contacts", "contact-forgotten", @@ -171,6 +323,33 @@ impl CoreInner { Ok(()) } + /// Forget every device at once, alongside the existing history-clearing + /// actions. Every peer loses access; each is notified best effort. + pub(super) async fn forget_all_contacts(self: &Arc) -> Result { + let store = self.repository.contacts(); + let contacts = store + .list_contacts() + .await + .map_err(VnidropError::repository)?; + let revoked = store + .delete_all_contacts() + .await + .map_err(VnidropError::repository)?; + for contact in &contacts { + self.offers.discard_from(&contact.endpoint_id).await; + } + self.emit_endpoint( + "contacts", + "contacts-cleared", + json!({ "contacts": contacts.len(), "revoked": revoked.len() }), + ); + for contact in contacts.iter() { + self.notify_revoked(contact.endpoint_id.clone(), revoked.clone()) + .await; + } + Ok(revoked.len() as u64) + } + pub(super) async fn block_contact(self: &Arc, endpoint_id: String) -> Result<()> { let store = self.repository.contacts(); let revoked = store @@ -185,6 +364,7 @@ impl CoreInner { .delete_contact(&endpoint_id) .await .map_err(VnidropError::repository)?; + self.offers.discard_from(&endpoint_id).await; self.emit_endpoint( "contacts", "contact-blocked", diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index 5a80bd1..17d886c 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -7,9 +7,9 @@ use super::CoreInner; use crate::{ api::{ ContactSummary, CoreEvent, CoreEventSink, CoreLimits, CoreNetworkConfig, CoreStorageUsage, - GrantLifetimeSetting, PendingPairing, ReceiveOutputSink, ReceiveOutputSinkV2, - ReceivedArtifact, ReceiverRequest, RuntimeStatus, ShareMetadataInput, ShareResult, - ShareSource, StoredTransfer, TicketInspection, TransferAccessMode, + GrantLifetimeSetting, IncomingOffer, PendingPairing, ReceiveOutputSink, + ReceiveOutputSinkV2, ReceivedArtifact, ReceiverRequest, RuntimeStatus, ShareMetadataInput, + ShareResult, ShareSource, StoredTransfer, TicketInspection, TransferAccessMode, }, error::VnidropError, filesystem::platform_path, @@ -277,6 +277,34 @@ impl VnidropCore { .map_err(VnidropError::repository) } + /// Share content and push it straight to a paired device. + /// + /// Only the receiving user is prompted: this device authorised the target + /// when it created the offer. + pub fn send_to_contact( + &self, + endpoint_id: String, + sources: Vec, + metadata: ShareMetadataInput, + ) -> Result { + self.block_on(self.inner.send_to_contact(endpoint_id, sources, metadata)) + .map_err(VnidropError::transfer) + } + + /// Transfers paired devices are offering, awaiting this user's decision. + pub fn list_pending_offers(&self) -> Vec { + self.block_on(self.inner.list_pending_offers()) + } + + /// Accept or decline an incoming offer. + /// + /// Returns the ticket when accepted, which the caller passes to `receive` + /// with its own destination. Declining returns none: a refused offer never + /// yields a capability. + pub fn respond_to_offer(&self, offer_id: String, accepted: bool) -> Option { + self.block_on(self.inner.respond_to_offer(offer_id, accepted)) + } + /// Devices offering to be remembered, awaiting the local user's decision. pub fn list_pending_pairings(&self) -> Vec { self.block_on(self.inner.list_pending_pairings()) @@ -316,6 +344,12 @@ impl VnidropCore { .map_err(VnidropError::repository) } + /// Forget every device at once. Returns how many grants were revoked. + pub fn forget_all_contacts(&self) -> Result { + self.block_on(self.inner.forget_all_contacts()) + .map_err(VnidropError::repository) + } + /// Refuse a device outright. Unlike forgetting, the peer is told nothing. pub fn block_contact(&self, endpoint_id: String) -> Result<(), VnidropError> { self.block_on(self.inner.block_contact(endpoint_id)) diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index 90c2950..4611387 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -58,6 +58,7 @@ use crate::{ handshake::HandshakeService, logging::init_logging, offer::OfferService, + offer_inbox::OfferInbox, pairing::PairingService, repository::Repository, secret::load_or_create_secret, @@ -95,6 +96,7 @@ pub(super) struct CoreInner { pub(super) event_hub: Arc, pub(super) approval: ApprovalService, pub(super) pairing: PairingService, + pub(super) offers: OfferInbox, pub(super) limits: CoreLimits, pub(super) relay_mode: CoreRelayMode, pub(super) custom_relay_urls: Vec, @@ -332,10 +334,22 @@ impl CoreInner { limits.max_pending_offers as usize, limits.max_metadata_bytes, ); + // Sweep grants dead long enough that no peer still needs the tombstone. + if let Err(error) = repository + .contacts() + .purge_dead_grants(crate::util::now_ms() - crate::contacts::DEAD_GRANT_RETENTION_MS) + .await + { + tracing::warn!(%error, "failed to sweep dead grants"); + } + let offers = OfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize); let router = Router::builder(endpoint.clone()) .accept(iroh_blobs::ALPN, blobs) .accept(HandshakeService::ALPN, handshake) - .accept(OfferService::ALPN, OfferService::new(pairing.clone())) + .accept( + OfferService::ALPN, + OfferService::new(pairing.clone(), offers.clone(), endpoint.id().to_string()), + ) .spawn(); let inner = Arc::new(Self { @@ -347,6 +361,7 @@ impl CoreInner { event_hub, approval, pairing, + offers, relay_mode, custom_relay_urls: relay_urls, transfer_slots: Semaphore::new(limits.max_concurrent_transfers as usize), diff --git a/crates/vnidrop/src/tests/grant.rs b/crates/vnidrop/src/tests/grant.rs index 74f45ee..b982052 100644 --- a/crates/vnidrop/src/tests/grant.rs +++ b/crates/vnidrop/src/tests/grant.rs @@ -185,21 +185,6 @@ fn never_lifetime_produces_no_deadline() { .expect("proof accepted"); assert_eq!(renewed, None); - assert!(grant.is_active(now + 10_000 * DAY_MS)); -} - -#[test] -fn is_active_tracks_revocation_and_expiry() { - let now = 1_700_000_000_000; - let grant = issued(now); - assert!(grant.is_active(now)); - - let mut revoked = issued(now); - revoked.revoked_at = Some(now); - assert!(!revoked.is_active(now)); - - let expires_at = grant.expires_at.expect("default lifetime expires"); - assert!(!grant.is_active(expires_at + 1)); } #[test] diff --git a/crates/vnidrop/tests/offer.rs b/crates/vnidrop/tests/offer.rs new file mode 100644 index 0000000..09c42ec --- /dev/null +++ b/crates/vnidrop/tests/offer.rs @@ -0,0 +1,371 @@ +//! Send-to-contact offers between two real nodes. + +mod support; + +use std::{ + path::Path, + sync::Arc, + time::{Duration, Instant}, +}; + +use support::{RecordingSink, TestNode}; +use vnidrop::{ + IncomingOffer, ShareMetadataInput, ShareResult, ShareSource, SourceKind, TransferAccessMode, + VnidropCore, VnidropError, +}; + +fn endpoint_id(node: &TestNode) -> String { + node.core.status().endpoint_id +} + +/// Establish a one-way relationship: `issuer` becomes reachable by `holder`. +fn pair(issuer: &TestNode, holder: &TestNode) { + let issuer_id = endpoint_id(issuer); + issuer + .core + .allow_device_to_reach_me(endpoint_id(holder), Some("Issuer".to_string())) + .expect("grant delivered"); + + let started = Instant::now(); + while !holder + .core + .list_pending_pairings() + .iter() + .any(|pending| pending.endpoint_id == issuer_id) + { + assert!( + started.elapsed() < Duration::from_secs(10), + "pairing offer never surfaced" + ); + std::thread::sleep(Duration::from_millis(25)); + } + holder + .core + .respond_to_pairing(issuer_id, true) + .expect("consent recorded"); +} + +fn sources(path: &Path) -> Vec { + vec![ShareSource { + kind: SourceKind::Path, + value: path.to_string_lossy().to_string(), + display_name: Some("shared.txt".to_string()), + is_directory: false, + }] +} + +fn metadata(transfer_id: u64) -> ShareMetadataInput { + ShareMetadataInput { + transfer_id, + transfer_name: Some("shared.txt".to_string()), + sender_name: Some("Sender".to_string()), + access_mode: TransferAccessMode::ApprovalRequired, + } +} + +/// Send in the background: the call blocks until the receiver decides. +fn send_in_background( + core: Arc, + to: String, + path: &Path, + transfer_id: u64, +) -> std::thread::JoinHandle> { + let sources = sources(path); + std::thread::spawn(move || core.send_to_contact(to, sources, metadata(transfer_id))) +} + +fn wait_for_offer(core: &VnidropCore) -> IncomingOffer { + let started = Instant::now(); + loop { + if let Some(offer) = core.list_pending_offers().into_iter().next() { + return offer; + } + assert!( + started.elapsed() < Duration::from_secs(10), + "offer never surfaced on the receiver" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +/// The whole point: the receiver is asked exactly once, the sender not at all. +#[test] +fn an_accepted_offer_transfers_without_prompting_the_sender() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"offered content").unwrap(); + let output_dir = tempfile::tempdir().unwrap(); + + let sender = TestNode::new(); + let receiver = TestNode::new(); + // The sender must be able to reach the receiver, so the receiver issues. + pair(&receiver, &sender); + + let handle = send_in_background( + sender.core.arc(), + endpoint_id(&receiver), + &source_path, + 4_001, + ); + + let offer = wait_for_offer(&receiver.core); + assert_eq!(offer.from_endpoint_id, endpoint_id(&sender)); + assert_eq!(offer.transfer_name, "shared.txt"); + assert_eq!(offer.file_count, 1); + assert_eq!(offer.sender_display_name.as_deref(), Some("Sender")); + + let ticket = receiver + .core + .respond_to_offer(offer.offer_id, true) + .expect("accepting yields the ticket"); + let share = handle.join().unwrap().expect("offer accepted"); + + receiver + .core + .receive( + ticket, + output_dir.path().to_string_lossy().to_string(), + Some("Receiver".to_string()), + ) + .expect("receive completes"); + + assert_eq!( + std::fs::read(output_dir.path().join("shared.txt")).unwrap(), + b"offered content" + ); + + // The sender was never asked: the only receiver request on its side was + // recorded as already approved. + let requests = sender + .core + .list_receiver_requests(share.transfer_id) + .unwrap(); + assert_eq!(requests.len(), 1); + assert!( + matches!(requests[0].status.as_str(), "accepted" | "completed"), + "sender should not have been prompted, got status {}", + requests[0].status + ); + assert!(requests[0].reason.is_none()); +} + +/// Declining yields no ticket and stops the share. +#[test] +fn a_declined_offer_yields_no_ticket() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"offered content").unwrap(); + + let sender = TestNode::new(); + let receiver = TestNode::new(); + pair(&receiver, &sender); + + let handle = send_in_background( + sender.core.arc(), + endpoint_id(&receiver), + &source_path, + 4_002, + ); + let offer = wait_for_offer(&receiver.core); + + assert!( + receiver + .core + .respond_to_offer(offer.offer_id, false) + .is_none(), + "a declined offer must not hand over a ticket" + ); + + let outcome = handle.join().unwrap(); + assert!(outcome.is_err(), "sender should see the refusal"); + assert!(receiver.core.list_pending_offers().is_empty()); +} + +/// A device with no grant cannot offer at all. +#[test] +fn sending_without_a_grant_is_refused_locally() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + + let sender = TestNode::new(); + let receiver = TestNode::new(); + + let outcome = sender.core.send_to_contact( + endpoint_id(&receiver), + sources(&source_path), + metadata(4_003), + ); + + assert!(outcome.is_err(), "no grant means nothing to send with"); + assert!(receiver.core.list_pending_offers().is_empty()); +} + +/// After the peer revokes, the offer is refused and the dead grant is dropped. +#[test] +fn a_revoked_grant_cannot_be_used_to_offer() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + + let sender = TestNode::new(); + let receiver = TestNode::new(); + pair(&receiver, &sender); + // The receiver decides it no longer wants to hear from the sender. + receiver + .core + .forget_contact(endpoint_id(&sender)) + .expect("forgotten"); + + let outcome = sender.core.send_to_contact( + endpoint_id(&receiver), + sources(&source_path), + metadata(4_004), + ); + + assert!(outcome.is_err()); + assert!(receiver.core.list_pending_offers().is_empty()); + let contacts = sender.core.list_contacts().unwrap(); + assert!( + contacts.iter().all(|contact| !contact.can_send), + "a refusal naming a dead grant must clear the sender's belief it can reach them" + ); +} + +/// An offer-created share is never public, whatever the caller asked for. +#[test] +fn an_offer_share_is_never_public() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + + let sender = TestNode::new(); + let receiver = TestNode::new(); + pair(&receiver, &sender); + + let core = sender.core.arc(); + let to = endpoint_id(&receiver); + let sources = sources(&source_path); + let handle = std::thread::spawn(move || { + core.send_to_contact( + to, + sources, + ShareMetadataInput { + transfer_id: 4_005, + transfer_name: Some("shared.txt".to_string()), + sender_name: None, + // Deliberately asking for the wider mode. + access_mode: TransferAccessMode::Public, + }, + ) + }); + + let offer = wait_for_offer(&receiver.core); + receiver.core.respond_to_offer(offer.offer_id, true); + let share = handle.join().unwrap().expect("offer accepted"); + + let stored = sender + .core + .list_transfers() + .unwrap() + .into_iter() + .find(|transfer| transfer.transfer_id == share.transfer_id) + .expect("share recorded"); + assert_eq!(stored.access_mode, TransferAccessMode::ApprovalRequired); +} + +/// A second offer while one is on screen is refused rather than stacked. +#[test] +fn only_one_offer_per_device_is_pending_at_a_time() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + + let sender = TestNode::new(); + let receiver = TestNode::new(); + pair(&receiver, &sender); + + let first = send_in_background( + sender.core.arc(), + endpoint_id(&receiver), + &source_path, + 4_006, + ); + wait_for_offer(&receiver.core); + + let second = sender.core.send_to_contact( + endpoint_id(&receiver), + sources(&source_path), + metadata(4_007), + ); + assert!(second.is_err(), "a second prompt must not stack"); + assert_eq!(receiver.core.list_pending_offers().len(), 1); + + let offer = receiver.core.list_pending_offers().remove(0); + receiver.core.respond_to_offer(offer.offer_id, false); + let _ = first.join().unwrap(); +} + +/// Forgetting a device clears any prompt it left on screen, which would +/// otherwise be actionable with a grant that no longer exists. +#[test] +fn forgetting_a_device_clears_its_pending_offer() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + + let sender = TestNode::new(); + let receiver = TestNode::new(); + pair(&receiver, &sender); + + let handle = send_in_background( + sender.core.arc(), + endpoint_id(&receiver), + &source_path, + 4_008, + ); + wait_for_offer(&receiver.core); + + receiver + .core + .forget_contact(endpoint_id(&sender)) + .expect("forgotten"); + + assert!(receiver.core.list_pending_offers().is_empty()); + assert!(handle.join().unwrap().is_err()); +} + +/// The ordinary QR path still prompts the sender: pre-authorisation applies +/// only to transfers the sender pushed. +#[test] +fn an_ordinary_ticket_receive_still_prompts_the_sender() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + let output_dir = tempfile::tempdir().unwrap(); + + let sender_dir = tempfile::tempdir().unwrap(); + let sink = Arc::new(RecordingSink::default()); + let sender = support::CoreGuard::start(sender_dir.path(), sink); + let receiver = TestNode::new(); + + let share = sender + .share_files(sources(&source_path), metadata(4_009)) + .expect("shared"); + + let core = receiver.core.arc(); + let ticket = share.ticket.clone(); + let output = output_dir.path().to_string_lossy().to_string(); + let handle = + std::thread::spawn(move || core.receive(ticket, output, Some("Receiver".to_string()))); + + let request = support::wait_for_receiver_request(&sender, share.transfer_id); + assert_eq!( + request.status, "requested", + "an unsolicited ticket receive must still ask the sender" + ); + sender + .respond_receiver_request(request.id, true, None) + .unwrap(); + handle.join().unwrap().expect("receive completes"); +}