mirror of
https://github.com/sudosylabs/vnidrop.git
synced 2026-08-12 05:29:57 +02:00
feat(core): complete one approved targeted transfer between saved devices
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -805,6 +805,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,
|
||||
@@ -1029,7 +1098,10 @@ impl DeviceRelationshipService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn peer_addr(&self, peer_endpoint_id: &str) -> Result<EndpointAddr, VnidropError> {
|
||||
pub(crate) async fn peer_addr(
|
||||
&self,
|
||||
peer_endpoint_id: &str,
|
||||
) -> Result<EndpointAddr, VnidropError> {
|
||||
let parsed: EndpointId = peer_endpoint_id
|
||||
.parse()
|
||||
.context("unusable peer endpoint id")
|
||||
@@ -1206,10 +1278,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,
|
||||
}
|
||||
|
||||
#[rpc_requests(message = RelationshipMessage)]
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -418,6 +418,65 @@ impl VnidropCore {
|
||||
)
|
||||
}
|
||||
|
||||
/// 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<ShareSource>,
|
||||
transfer_name: Option<String>,
|
||||
) -> Result<crate::api::TargetedTransfer, VnidropError> {
|
||||
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<crate::api::PendingTargetedOffer> {
|
||||
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<Option<String>, 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<Option<crate::api::TargetedTransfer>, VnidropError> {
|
||||
self.block_on(self.inner.get_targeted_transfer(id))
|
||||
}
|
||||
|
||||
pub fn list_targeted_transfers(
|
||||
&self,
|
||||
) -> Result<Vec<crate::api::TargetedTransfer>, VnidropError> {
|
||||
self.block_on(self.inner.list_targeted_transfers())
|
||||
}
|
||||
|
||||
/// Devices the user has chosen to remember.
|
||||
pub fn list_contacts(&self) -> Result<Vec<ContactSummary>, VnidropError> {
|
||||
self.block_on(self.inner.list_contacts())
|
||||
|
||||
@@ -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<DeviceRelationshipService>,
|
||||
pub(super) offers: OfferInbox,
|
||||
pub(super) targeted_offers: TargetedOfferInbox,
|
||||
/// Endpoint → last poll time, for the rate limit above.
|
||||
pub(super) last_polled: TokioMutex<HashMap<String, i64>>,
|
||||
pub(super) limits: CoreLimits,
|
||||
@@ -387,6 +391,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(),
|
||||
@@ -406,6 +412,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 {
|
||||
@@ -421,6 +436,7 @@ impl CoreInner {
|
||||
pairing_eligibility,
|
||||
device_relationships,
|
||||
offers,
|
||||
targeted_offers,
|
||||
last_polled: TokioMutex::new(HashMap::new()),
|
||||
relay_mode,
|
||||
custom_relay_urls: relay_urls,
|
||||
|
||||
339
crates/vnidrop/src/runtime/targeted.rs
Normal file
339
crates/vnidrop/src/runtime/targeted.rs
Normal file
@@ -0,0 +1,339 @@
|
||||
//! 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<PendingTargetedOffer> {
|
||||
self.targeted_offers.list().await
|
||||
}
|
||||
|
||||
pub(super) async fn get_targeted_transfer(
|
||||
&self,
|
||||
id: String,
|
||||
) -> Result<Option<TargetedTransfer>, VnidropError> {
|
||||
self.targeted_store().get(&id).await
|
||||
}
|
||||
|
||||
pub(super) async fn list_targeted_transfers(
|
||||
&self,
|
||||
) -> Result<Vec<TargetedTransfer>, VnidropError> {
|
||||
self.targeted_store().list().await
|
||||
}
|
||||
|
||||
/// Cancel in-flight targeted transfers involving `peer` (for forget/block).
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "hook for ticket 09 forget/block to call without waiting on that work"
|
||||
)]
|
||||
pub(crate) async fn cancel_targeted_transfers_for_peer(
|
||||
&self,
|
||||
peer_endpoint_id: &str,
|
||||
) -> Result<u64, VnidropError> {
|
||||
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<Option<String>, 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<Self>,
|
||||
receiver_endpoint_id: String,
|
||||
sources: Vec<ShareSource>,
|
||||
transfer_name: Option<String>,
|
||||
) -> Result<TargetedTransfer, VnidropError> {
|
||||
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<Self>,
|
||||
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<BlobTicket, String>;
|
||||
}
|
||||
|
||||
impl BlobTicketParse for BlobTicket {
|
||||
fn from_str_compat(value: &str) -> Result<BlobTicket, String> {
|
||||
use std::str::FromStr;
|
||||
BlobTicket::from_str(value).map_err(|error| error.to_string())
|
||||
}
|
||||
}
|
||||
140
crates/vnidrop/src/targeted_transfer/auth.rs
Normal file
140
crates/vnidrop/src/targeted_transfer/auth.rs
Normal file
@@ -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<Self, VnidropError> {
|
||||
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<String, VnidropError> {
|
||||
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<Self, VnidropError> {
|
||||
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())
|
||||
}
|
||||
}
|
||||
234
crates/vnidrop/src/targeted_transfer/inbox.rs
Normal file
234
crates/vnidrop/src/targeted_transfer/inbox.rs
Normal file
@@ -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<bool>,
|
||||
}
|
||||
|
||||
struct AuthWaiter {
|
||||
auth: oneshot::Sender<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct TargetedOfferInbox {
|
||||
event_hub: Arc<EventHub>,
|
||||
pending: Arc<Mutex<HashMap<String, PendingTargetedOfferRecord>>>,
|
||||
decisions: Arc<Mutex<HashMap<String, DecisionWaiter>>>,
|
||||
auths: Arc<Mutex<HashMap<String, AuthWaiter>>>,
|
||||
max_pending: usize,
|
||||
}
|
||||
|
||||
impl TargetedOfferInbox {
|
||||
pub(crate) fn new(event_hub: Arc<EventHub>, 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<PendingTargetedOffer> {
|
||||
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<Option<String>, 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<String> = {
|
||||
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()
|
||||
}
|
||||
267
crates/vnidrop/src/targeted_transfer/mod.rs
Normal file
267
crates/vnidrop/src/targeted_transfer/mod.rs
Normal file
@@ -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<Option<TargetedTransfer>, 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<Option<TargetedTransferRow>, 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<Vec<TargetedTransfer>, 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<u64, VnidropError> {
|
||||
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<TargetedTransfer, VnidropError> {
|
||||
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::<i64, _>("file_count") as u64,
|
||||
total_size: row.get::<i64, _>("total_size") as u64,
|
||||
state: parse_state(&row.get::<String, _>("state"))?,
|
||||
created_at: row.get("created_at"),
|
||||
updated_at: row.get("updated_at"),
|
||||
})
|
||||
}
|
||||
|
||||
fn row_to_full(row: sqlx::sqlite::SqliteRow) -> Result<TargetedTransferRow, VnidropError> {
|
||||
Ok(TargetedTransferRow {
|
||||
id: row.get("id"),
|
||||
protocol_transfer_id: row.get::<i64, _>("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::<i64, _>("file_count") as u64,
|
||||
total_size: row.get::<i64, _>("total_size") as u64,
|
||||
state: parse_state(&row.get::<String, _>("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<TargetedTransferState, VnidropError> {
|
||||
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}"
|
||||
))),
|
||||
}
|
||||
}
|
||||
297
crates/vnidrop/src/targeted_transfer/protocol.rs
Normal file
297
crates/vnidrop/src/targeted_transfer/protocol.rs
Normal file
@@ -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<DeviceRelationshipService>,
|
||||
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<DeviceRelationshipService>,
|
||||
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::<TargetedTransferMessages>(&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<TargetedTransferMessages>,
|
||||
}
|
||||
|
||||
impl TargetedTransferClient {
|
||||
pub(crate) async fn request_challenge(&self) -> Result<Challenge, irpc::Error> {
|
||||
Ok(self.inner.rpc(RequestChallenge).await?.challenge)
|
||||
}
|
||||
|
||||
pub(crate) async fn submit_offer(
|
||||
&self,
|
||||
offer: SubmitTargetedOffer,
|
||||
) -> Result<TargetedOfferResponse, irpc::Error> {
|
||||
self.inner.rpc(offer).await
|
||||
}
|
||||
|
||||
pub(crate) async fn deliver_authorization(
|
||||
&self,
|
||||
delivery: DeliverTargetedAuthorization,
|
||||
) -> Result<DeliverAuthorizationResponse, irpc::Error> {
|
||||
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<ChallengeResponse>)]
|
||||
RequestChallenge(RequestChallenge),
|
||||
#[rpc(tx = oneshot::Sender<TargetedOfferResponse>)]
|
||||
SubmitTargetedOffer(SubmitTargetedOffer),
|
||||
#[rpc(tx = oneshot::Sender<DeliverAuthorizationResponse>)]
|
||||
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}"))
|
||||
}
|
||||
@@ -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"]
|
||||
|
||||
471
crates/vnidrop/src/tests/targeted_transfer.rs
Normal file
471
crates/vnidrop/src/tests/targeted_transfer.rs
Normal file
@@ -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<Vec<CoreEvent>>,
|
||||
}
|
||||
|
||||
impl CoreEventSink for RecordingSink {
|
||||
fn on_event(&self, event: CoreEvent) {
|
||||
self.events.lock().unwrap().push(event);
|
||||
}
|
||||
}
|
||||
|
||||
struct ProtectedNode {
|
||||
_data_dir: tempfile::TempDir,
|
||||
core: Arc<VnidropCore>,
|
||||
}
|
||||
|
||||
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::<String>));
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user