From 68363fb680e5c08fc9b46f0c90bc81a8f67bd28b Mon Sep 17 00:00:00 2001 From: Hammed Abass Date: Fri, 3 Jul 2026 15:21:22 +0200 Subject: [PATCH] Harden Rust transfer core --- crates/vnidrop/src/access_policy.rs | 3 + crates/vnidrop/src/error.rs | 58 +++++- crates/vnidrop/src/event_hub.rs | 130 +++++++++++++ crates/vnidrop/src/filesystem.rs | 12 +- crates/vnidrop/src/lib.rs | 4 +- crates/vnidrop/src/repository.rs | 55 ++++-- crates/vnidrop/src/runtime.rs | 242 ++++++++++++++++--------- crates/vnidrop/src/tests.rs | 173 ++++++++++++++++-- crates/vnidrop/src/ticket.rs | 24 ++- crates/vnidrop/tests/local_transfer.rs | 199 +++++++++++++++++++- 10 files changed, 772 insertions(+), 128 deletions(-) create mode 100644 crates/vnidrop/src/event_hub.rs diff --git a/crates/vnidrop/src/access_policy.rs b/crates/vnidrop/src/access_policy.rs index 1547774..2604b53 100644 --- a/crates/vnidrop/src/access_policy.rs +++ b/crates/vnidrop/src/access_policy.rs @@ -48,6 +48,9 @@ impl AccessPolicy { transfer_id: u64, endpoint_id: Option<&str>, ) -> AccessDecision { + // This is intentionally only the provider-side gate for milestone one. + // A later handshake can add receiver-request/sender-approval events on + // top without weakening the default public sharing behavior. match self .modes .read() diff --git a/crates/vnidrop/src/error.rs b/crates/vnidrop/src/error.rs index b6dc915..a431309 100644 --- a/crates/vnidrop/src/error.rs +++ b/crates/vnidrop/src/error.rs @@ -2,13 +2,63 @@ use std::io; #[derive(Debug, thiserror::Error, uniffi::Error)] pub enum VnidropError { - #[error("{reason}")] - Generic { reason: String }, + #[error("initialization failed: {reason}")] + Initialization { reason: String }, + #[error("ticket error: {reason}")] + Ticket { reason: String }, + #[error("filesystem error: {reason}")] + Filesystem { reason: String }, + #[error("transfer error: {reason}")] + Transfer { reason: String }, + #[error("permission error: {reason}")] + Permission { reason: String }, + #[error("repository error: {reason}")] + Repository { reason: String }, + #[error("internal error: {reason}")] + Internal { reason: String }, +} + +impl VnidropError { + pub(crate) fn initialization(error: impl Into) -> Self { + Self::Initialization { + reason: error.into().to_string(), + } + } + + pub(crate) fn ticket(error: impl Into) -> Self { + Self::Ticket { + reason: error.into().to_string(), + } + } + + pub(crate) fn filesystem(error: impl Into) -> Self { + Self::Filesystem { + reason: error.into().to_string(), + } + } + + pub(crate) fn transfer(error: impl Into) -> Self { + Self::Transfer { + reason: error.into().to_string(), + } + } + + pub(crate) fn permission(error: impl Into) -> Self { + Self::Permission { + reason: error.into().to_string(), + } + } + + pub(crate) fn repository(error: impl Into) -> Self { + Self::Repository { + reason: error.into().to_string(), + } + } } impl From for VnidropError { fn from(error: anyhow::Error) -> Self { - Self::Generic { + Self::Internal { reason: error.to_string(), } } @@ -16,7 +66,7 @@ impl From for VnidropError { impl From for VnidropError { fn from(error: io::Error) -> Self { - Self::Generic { + Self::Filesystem { reason: error.to_string(), } } diff --git a/crates/vnidrop/src/event_hub.rs b/crates/vnidrop/src/event_hub.rs new file mode 100644 index 0000000..22ec182 --- /dev/null +++ b/crates/vnidrop/src/event_hub.rs @@ -0,0 +1,130 @@ +use std::sync::{Arc, Mutex}; + +use serde_json::Value; +use tokio::{ + sync::{mpsc, oneshot, Mutex as TokioMutex}, + task::JoinHandle, +}; + +use crate::{ + api::{CoreEvent, CoreEventSink}, + repository::Repository, + util::now_ms, +}; + +enum EventCommand { + Persist(CoreEvent), + Flush(oneshot::Sender<()>), + Shutdown(oneshot::Sender<()>), +} + +pub(crate) struct EventHub { + sink: Arc, + tx: mpsc::UnboundedSender, + join: TokioMutex>>, + sequence: Mutex, +} + +impl EventHub { + pub(crate) fn start(repository: Repository, sink: Arc) -> Self { + let (tx, mut rx) = mpsc::unbounded_channel(); + let join = tokio::spawn(async move { + while let Some(command) = rx.recv().await { + match command { + EventCommand::Persist(event) => { + if let Err(error) = repository.insert_event(&event).await { + tracing::warn!(%error, event_id = %event.id, "failed to persist core event"); + } + } + EventCommand::Flush(done) => { + let _ = done.send(()); + } + EventCommand::Shutdown(done) => { + let _ = done.send(()); + break; + } + } + } + }); + + Self { + sink, + tx, + join: TokioMutex::new(Some(join)), + sequence: Mutex::new(1), + } + } + + pub(crate) fn emit_endpoint(&self, phase: &str, kind: &str, data: Value) { + self.emit("endpoint", None, None, phase, kind, data); + } + + pub(crate) fn emit_transfer( + &self, + transfer_id: u64, + direction: &str, + phase: &str, + kind: &str, + data: Value, + ) { + self.emit( + "transfer", + Some(transfer_id), + Some(direction.to_string()), + phase, + kind, + data, + ); + } + + pub(crate) async fn flush(&self) { + let (tx, rx) = oneshot::channel(); + if self.tx.send(EventCommand::Flush(tx)).is_ok() { + let _ = rx.await; + } + } + + pub(crate) async fn shutdown(&self) { + let (tx, rx) = oneshot::channel(); + if self.tx.send(EventCommand::Shutdown(tx)).is_ok() { + let _ = rx.await; + } + if let Some(join) = self.join.lock().await.take() { + let _ = join.await; + } + } + + fn emit( + &self, + scope: &str, + transfer_id: Option, + direction: Option, + phase: &str, + kind: &str, + data: Value, + ) { + let timestamp = now_ms(); + let mut sequence = self.sequence.lock().expect("event sequence lock poisoned"); + let id = format!("{timestamp}-{}", *sequence); + *sequence += 1; + drop(sequence); + + // Compose observes this event synchronously, while SQLite persistence is + // serialized through the queue. That keeps the UI responsive without + // losing the ability to flush persisted history during shutdown/tests. + let event = CoreEvent { + id, + timestamp, + scope: scope.to_string(), + transfer_id, + direction, + phase: phase.to_string(), + kind: kind.to_string(), + data_json: data.to_string(), + }; + if self.tx.send(EventCommand::Persist(event.clone())).is_err() { + tracing::warn!(event_id = %event.id, "event persistence queue is closed"); + } + self.sink.on_event(event); + } +} diff --git a/crates/vnidrop/src/filesystem.rs b/crates/vnidrop/src/filesystem.rs index 7eac03e..0180e14 100644 --- a/crates/vnidrop/src/filesystem.rs +++ b/crates/vnidrop/src/filesystem.rs @@ -122,10 +122,16 @@ pub(crate) fn collect_import_files(sources: Vec) -> Result Result { + if source.value.trim().is_empty() { + anyhow::bail!("source path must not be empty"); + } platform_path(&source.value) } pub(crate) fn platform_path(value: &str) -> Result { + if value.trim().is_empty() { + anyhow::bail!("path must not be empty"); + } if let Some(without_scheme) = value.strip_prefix("file://") { return Ok(PathBuf::from(percent_decode_file_url_path(without_scheme)?)); } @@ -230,7 +236,11 @@ pub(crate) async fn wait_for_writer( } pub(crate) fn validated_relative_string(name: &str) -> Result { - path_to_string(Path::new(name), true) + let value = path_to_string(Path::new(name), true)?; + if value.trim().is_empty() { + anyhow::bail!("relative path must not be empty"); + } + Ok(value) } pub(crate) fn path_to_string(path: impl AsRef, must_be_relative: bool) -> Result { diff --git a/crates/vnidrop/src/lib.rs b/crates/vnidrop/src/lib.rs index e268451..802e2fa 100644 --- a/crates/vnidrop/src/lib.rs +++ b/crates/vnidrop/src/lib.rs @@ -1,6 +1,7 @@ mod access_policy; mod api; mod error; +mod event_hub; mod filesystem; mod logging; mod repository; @@ -19,4 +20,5 @@ pub use runtime::VnidropCore; uniffi::setup_scaffolding!(); #[cfg(test)] -mod tests; +#[path = "tests.rs"] +mod core_tests; diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index 15d429a..e45c001 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -9,11 +9,24 @@ use sqlx::{ use crate::api::{CoreEvent, StoredTransfer}; use crate::util::now_ms; +const SCHEMA_VERSION: i64 = 1; + #[derive(Debug, Clone)] pub(crate) struct Repository { pool: SqlitePool, } +pub(crate) struct TransferUpsert<'a> { + pub(crate) transfer_id: u64, + pub(crate) direction: &'a str, + pub(crate) status: &'a str, + pub(crate) transfer_name: Option<&'a str>, + pub(crate) content_hash: Option<&'a str>, + pub(crate) ticket: Option<&'a str>, + pub(crate) file_count: u64, + pub(crate) total_size: u64, +} + impl Repository { pub(crate) async fn open(app_data_dir: &Path) -> Result { let db_path = app_data_dir.join("vnidrop.sqlite3"); @@ -30,6 +43,8 @@ impl Repository { } async fn ensure_schema(&self) -> Result<()> { + // The app owns this SQLite file. Keep migrations explicit so future + // desktop/mobile releases can move user history forward in place. sqlx::query( r#" CREATE TABLE IF NOT EXISTS transfers ( @@ -71,20 +86,22 @@ impl Repository { ) .execute(&self.pool) .await?; + + sqlx::query(&format!("PRAGMA user_version = {SCHEMA_VERSION}")) + .execute(&self.pool) + .await?; Ok(()) } - pub(crate) async fn upsert_transfer( - &self, - transfer_id: u64, - direction: &str, - status: &str, - transfer_name: Option<&str>, - content_hash: Option<&str>, - ticket: Option<&str>, - file_count: u64, - total_size: u64, - ) -> Result<()> { + #[cfg(test)] + pub(crate) async fn schema_version(&self) -> Result { + let row = sqlx::query("PRAGMA user_version") + .fetch_one(&self.pool) + .await?; + Ok(row.get(0)) + } + + pub(crate) async fn upsert_transfer(&self, transfer: TransferUpsert<'_>) -> Result<()> { let now = now_ms(); sqlx::query( r#" @@ -104,14 +121,14 @@ impl Repository { updated_at = excluded.updated_at; "#, ) - .bind(transfer_id as i64) - .bind(direction) - .bind(status) - .bind(transfer_name) - .bind(content_hash) - .bind(ticket) - .bind(file_count as i64) - .bind(total_size as i64) + .bind(transfer.transfer_id as i64) + .bind(transfer.direction) + .bind(transfer.status) + .bind(transfer.transfer_name) + .bind(transfer.content_hash) + .bind(transfer.ticket) + .bind(transfer.file_count as i64) + .bind(transfer.total_size as i64) .bind(now) .execute(&self.pool) .await?; diff --git a/crates/vnidrop/src/runtime.rs b/crates/vnidrop/src/runtime.rs index b7a67f9..1881006 100644 --- a/crates/vnidrop/src/runtime.rs +++ b/crates/vnidrop/src/runtime.rs @@ -3,7 +3,10 @@ use std::{ fs::File, io, path::{Path, PathBuf}, - sync::{Arc, Mutex}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, }; use anyhow::{Context, Result}; @@ -21,7 +24,10 @@ use iroh_blobs::{ }; use n0_future::BufferedStreamExt; use serde_json::json; -use tokio::sync::{mpsc, oneshot, Mutex as TokioMutex}; +use tokio::{ + sync::{mpsc, oneshot, Mutex as TokioMutex}, + task::JoinHandle, +}; use crate::{ access_policy::{AccessDecision, AccessPolicy}, @@ -30,18 +36,26 @@ use crate::{ StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata, }, error::VnidropError, + event_hub::EventHub, filesystem::{ collect_import_files, default_collection_name, platform_path, read_stream_from_blocking_reader, safe_output_path, wait_for_writer, write_stream_to_blocking_writer, TransferImport, }, logging::init_logging, - repository::Repository, + repository::{Repository, TransferUpsert}, secret::load_or_create_secret, ticket::{parse_transfer_ticket, ParsedTransferTicket, VnidropTicket}, - util::{non_empty, now_ms, unique_transfer_id}, + util::{non_empty, unique_transfer_id}, }; +const STATUS_SHARING: &str = "sharing"; +const STATUS_RECEIVING: &str = "receiving"; +const STATUS_DONE: &str = "done"; +const STATUS_CANCELLED: &str = "cancelled"; +const STATUS_STOPPED: &str = "stopped"; +const STATUS_FAILED: &str = "failed"; + #[derive(uniffi::Object)] pub struct VnidropCore { runtime: tokio::runtime::Runtime, @@ -49,17 +63,20 @@ pub struct VnidropCore { } struct CoreInner { + // Kotlin owns app lifecycle and platform file picking; this Rust object + // owns the Iroh endpoint, blob store, transfer history, and byte streaming. endpoint: Endpoint, router: Router, store: FsStore, repository: Repository, + event_hub: EventHub, access_policy: Arc, - event_sink: Arc, active_transfers: TokioMutex>>, active_shares: TokioMutex>, hash_to_transfer: TokioMutex>, connection_endpoints: TokioMutex>, - sequence: Mutex, + provider_task: TokioMutex>>, + shutdown_started: AtomicBool, } #[uniffi::export] @@ -74,7 +91,9 @@ impl VnidropCore { .thread_name("vnidrop") .build()?; let app_data_dir = PathBuf::from(app_data_dir); - let inner = runtime.block_on(CoreInner::start(app_data_dir, event_sink))?; + let inner = runtime + .block_on(CoreInner::start(app_data_dir, event_sink)) + .map_err(VnidropError::initialization)?; Ok(Arc::new(Self { runtime, inner })) } @@ -89,7 +108,7 @@ impl VnidropCore { ) -> Result { self.runtime .block_on(self.inner.share_files(sources, metadata)) - .map_err(Into::into) + .map_err(VnidropError::transfer) } pub fn receive( @@ -98,16 +117,29 @@ impl VnidropCore { output_dir: String, receiver_name: Option, ) -> Result<(), VnidropError> { - let output_dir = platform_path(&output_dir)?; + if let Err(error) = + parse_transfer_ticket(&ticket).context("failed to parse transfer ticket") + { + self.runtime.block_on(async { + self.inner.emit_endpoint( + "error", + "invalid-ticket", + json!({ "reason": error.to_string() }), + ); + self.inner.event_hub.flush().await; + }); + return Err(VnidropError::ticket(error)); + } + let output_dir = platform_path(&output_dir).map_err(VnidropError::filesystem)?; self.runtime .block_on(self.inner.receive(ticket, output_dir, receiver_name)) - .map_err(Into::into) + .map_err(VnidropError::transfer) } pub fn cancel_transfer(&self, transfer_id: u64) -> Result<(), VnidropError> { self.runtime .block_on(self.inner.cancel_transfer(transfer_id)) - .map_err(Into::into) + .map_err(VnidropError::transfer) } pub fn set_transfer_access_mode( @@ -117,7 +149,7 @@ impl VnidropCore { ) -> Result<(), VnidropError> { self.runtime .block_on(self.inner.set_transfer_access_mode(transfer_id, mode)) - .map_err(Into::into) + .map_err(VnidropError::permission) } pub fn approve_endpoint_for_transfer( @@ -130,23 +162,25 @@ impl VnidropCore { self.inner .approve_endpoint_for_transfer(transfer_id, endpoint_id), ) - .map_err(Into::into) + .map_err(VnidropError::permission) } pub fn list_transfers(&self) -> Result, VnidropError> { self.runtime .block_on(self.inner.repository.list_transfers()) - .map_err(Into::into) + .map_err(VnidropError::repository) } pub fn list_events(&self, transfer_id: Option) -> Result, VnidropError> { self.runtime - .block_on(self.inner.repository.list_events(transfer_id)) - .map_err(Into::into) + .block_on(self.inner.list_events(transfer_id)) + .map_err(VnidropError::repository) } pub fn inspect_ticket(&self, ticket: String) -> Result { - let parsed = parse_transfer_ticket(&ticket).context("failed to parse transfer ticket")?; + let parsed = parse_transfer_ticket(&ticket) + .context("failed to parse transfer ticket") + .map_err(VnidropError::ticket)?; Ok(TicketInspection { kind: if parsed.metadata.is_some() { "vnidrop".to_string() @@ -177,24 +211,28 @@ impl CoreInner { .await?; endpoint.online().await; + // Provider events are where the sender sees remote readers. The core + // uses them for send progress and for the current approval gate. let (events, event_rx) = EventSender::channel(128, EventMask::ALL_READONLY); let blobs = BlobsProtocol::new(&store, Some(events)); let router = Router::builder(endpoint.clone()) .accept(iroh_blobs::ALPN, blobs) .spawn(); + let event_hub = EventHub::start(repository.clone(), event_sink); let inner = Arc::new(Self { endpoint, router, store, repository, + event_hub, access_policy: AccessPolicy::new(), - event_sink, active_transfers: TokioMutex::new(HashMap::new()), active_shares: TokioMutex::new(HashMap::new()), hash_to_transfer: TokioMutex::new(HashMap::new()), connection_endpoints: TokioMutex::new(HashMap::new()), - sequence: Mutex::new(1), + provider_task: TokioMutex::new(None), + shutdown_started: AtomicBool::new(false), }); inner.emit_endpoint( @@ -206,7 +244,7 @@ impl CoreInner { "store_root": store_root.to_string_lossy(), }), ); - inner.spawn_provider_event_task(event_rx); + inner.spawn_provider_event_task(event_rx).await; Ok(inner) } @@ -223,6 +261,29 @@ impl CoreInner { self: &Arc, sources: Vec, metadata: ShareMetadataInput, + ) -> Result { + let transfer_id = metadata.transfer_id; + let result = self.share_files_inner(sources, metadata).await; + if let Err(error) = &result { + self.emit_transfer( + transfer_id, + "send", + "error", + "failed", + json!({ "reason": error.to_string() }), + ); + let _ = self + .repository + .update_transfer_status(transfer_id, STATUS_FAILED) + .await; + } + result + } + + async fn share_files_inner( + self: &Arc, + sources: Vec, + metadata: ShareMetadataInput, ) -> Result { if sources.is_empty() { anyhow::bail!("at least one source is required"); @@ -266,16 +327,16 @@ impl CoreInner { .await .insert(metadata.transfer_id, import.tag); self.repository - .upsert_transfer( - metadata.transfer_id, - "send", - "sharing", - Some(&transfer_name), - Some(&import.root_hash.to_string()), - Some(&ticket), - import.file_count, - import.total_size, - ) + .upsert_transfer(TransferUpsert { + transfer_id: metadata.transfer_id, + direction: "send", + status: STATUS_SHARING, + transfer_name: Some(&transfer_name), + content_hash: Some(&import.root_hash.to_string()), + ticket: Some(&ticket), + file_count: import.file_count, + total_size: import.total_size, + }) .await?; self.emit_transfer( @@ -308,12 +369,25 @@ impl CoreInner { output_dir: PathBuf, receiver_name: Option, ) -> Result<()> { - let parsed = parse_transfer_ticket(&ticket).context("failed to parse transfer ticket")?; + let parsed = match parse_transfer_ticket(&ticket).context("failed to parse transfer ticket") + { + Ok(parsed) => parsed, + Err(error) => { + self.emit_endpoint( + "error", + "invalid-ticket", + json!({ "reason": error.to_string() }), + ); + return Err(error); + } + }; let transfer_id = parsed .metadata .as_ref() .map(|metadata| metadata.transfer_id) .unwrap_or_else(unique_transfer_id); + // Cancellation is cooperative: it stops our receive future and marks + // local state while lower-level Iroh work unwinds naturally. let (shutdown_tx, mut shutdown_rx) = oneshot::channel(); self.active_transfers .lock() @@ -326,6 +400,19 @@ impl CoreInner { }; self.active_transfers.lock().await.remove(&transfer_id); + if let Err(error) = &result { + self.emit_transfer( + transfer_id, + "receive", + "error", + "failed", + json!({ "reason": error.to_string() }), + ); + let _ = self + .repository + .update_transfer_status(transfer_id, STATUS_FAILED) + .await; + } result } @@ -336,7 +423,8 @@ impl CoreInner { output_dir: PathBuf, receiver_name: Option, ) -> Result<()> { - let metadata_json = serde_json::to_value(&parsed.metadata).unwrap_or_else(|_| json!(null)); + let metadata_json = + serde_json::to_value(&parsed.metadata).unwrap_or(serde_json::Value::Null); self.emit_transfer( transfer_id, "receive", @@ -348,30 +436,30 @@ impl CoreInner { }), ); self.repository - .upsert_transfer( + .upsert_transfer(TransferUpsert { transfer_id, - "receive", - "receiving", - parsed + direction: "receive", + status: STATUS_RECEIVING, + transfer_name: parsed .metadata .as_ref() .map(|metadata| metadata.transfer_name.as_str()), - parsed + content_hash: parsed .metadata .as_ref() .map(|metadata| metadata.content_hash.as_str()), - None, - parsed + ticket: None, + file_count: parsed .metadata .as_ref() .map(|metadata| metadata.file_count) .unwrap_or_default(), - parsed + total_size: parsed .metadata .as_ref() .map(|metadata| metadata.total_size) .unwrap_or_default(), - ) + }) .await?; tokio::fs::create_dir_all(&output_dir).await?; @@ -419,7 +507,7 @@ impl CoreInner { self.export_collection(transfer_id, total_files, output_dir, collection) .await?; self.repository - .update_transfer_status(transfer_id, "done") + .update_transfer_status(transfer_id, STATUS_DONE) .await?; self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({})); Ok(()) @@ -436,7 +524,7 @@ impl CoreInner { json!({}), ); self.repository - .update_transfer_status(transfer_id, "cancelled") + .update_transfer_status(transfer_id, STATUS_CANCELLED) .await?; return Ok(()); } @@ -453,7 +541,7 @@ impl CoreInner { .retain(|_, id| *id != transfer_id); self.access_policy.remove_transfer(transfer_id).await; self.repository - .update_transfer_status(transfer_id, "stopped") + .update_transfer_status(transfer_id, STATUS_STOPPED) .await?; self.emit_transfer(transfer_id, "send", "lifecycle", "share-stopped", json!({})); return Ok(()); @@ -496,7 +584,13 @@ impl CoreInner { } async fn shutdown(&self) { + if self.shutdown_started.swap(true, Ordering::SeqCst) { + return; + } self.emit_endpoint("shutdown", "service-shutdown", json!({})); + // Flush before stopping the router so the app can show the shutdown + // event even if the process exits soon after Compose disposes the core. + self.event_hub.flush().await; if let Err(error) = self.router.shutdown().await { self.emit_endpoint( "shutdown", @@ -504,6 +598,11 @@ impl CoreInner { json!({ "error": error.to_string() }), ); } + if let Some(task) = self.provider_task.lock().await.take() { + task.abort(); + let _ = task.await; + } + self.event_hub.shutdown().await; } async fn import_sources( @@ -693,13 +792,14 @@ impl CoreInner { Ok(()) } - fn spawn_provider_event_task(self: &Arc, mut rx: mpsc::Receiver) { + async fn spawn_provider_event_task(self: &Arc, mut rx: mpsc::Receiver) { let core = self.clone(); - tokio::spawn(async move { + let task = tokio::spawn(async move { while let Some(message) = rx.recv().await { core.handle_provider_message(message).await; } }); + *self.provider_task.lock().await = Some(task); } async fn handle_provider_message(self: &Arc, message: ProviderMessage) { @@ -917,6 +1017,9 @@ impl CoreInner { request_id: u64, mut rx: irpc::channel::mpsc::Receiver, ) { + // Request update tasks are tied to individual provider streams. Router + // shutdown closes those streams; only the long-lived provider receiver + // is tracked directly for explicit shutdown. let core = self.clone(); tokio::spawn(async move { while let Ok(Some(update)) = rx.recv().await { @@ -965,7 +1068,7 @@ impl CoreInner { } fn emit_endpoint(&self, phase: &str, kind: &str, data: serde_json::Value) { - self.emit("endpoint", None, None, phase, kind, data); + self.event_hub.emit_endpoint(phase, kind, data); } fn emit_transfer( @@ -976,47 +1079,12 @@ impl CoreInner { kind: &str, data: serde_json::Value, ) { - self.emit( - "transfer", - Some(transfer_id), - Some(direction.to_string()), - phase, - kind, - data, - ); + self.event_hub + .emit_transfer(transfer_id, direction, phase, kind, data); } - fn emit( - &self, - scope: &str, - transfer_id: Option, - direction: Option, - phase: &str, - kind: &str, - data: serde_json::Value, - ) { - let timestamp = now_ms(); - let mut sequence = self.sequence.lock().expect("event sequence lock poisoned"); - let id = format!("{timestamp}-{}", *sequence); - *sequence += 1; - drop(sequence); - let event = CoreEvent { - id, - timestamp, - scope: scope.to_string(), - transfer_id, - direction, - phase: phase.to_string(), - kind: kind.to_string(), - data_json: data.to_string(), - }; - let repository = self.repository.clone(); - let event_for_repository = event.clone(); - tokio::spawn(async move { - if let Err(error) = repository.insert_event(&event_for_repository).await { - tracing::warn!(%error, "failed to persist core event"); - } - }); - self.event_sink.on_event(event); + async fn list_events(&self, transfer_id: Option) -> Result> { + self.event_hub.flush().await; + self.repository.list_events(transfer_id).await } } diff --git a/crates/vnidrop/src/tests.rs b/crates/vnidrop/src/tests.rs index fff122e..12217db 100644 --- a/crates/vnidrop/src/tests.rs +++ b/crates/vnidrop/src/tests.rs @@ -4,15 +4,18 @@ mod tests { use std::os::fd::AsRawFd; use std::{io::Read, path::Path, sync::Arc}; + use data_encoding::BASE64URL_NOPAD; use iroh::SecretKey; use iroh_blobs::{ticket::BlobTicket, BlobFormat, Hash}; + use serde_json::json; use crate::{ access_policy::{AccessDecision, AccessPolicy}, api::{CoreEvent, CoreEventSink, ShareSource, SourceKind, TransferMetadata}, + error::VnidropError, filesystem::{ - collect_import_files, path_to_string, percent_decode_file_url_path, - validated_relative_string, + collect_import_files, default_collection_name, path_to_string, + percent_decode_file_url_path, validated_relative_string, }, repository::Repository, runtime::VnidropCore, @@ -77,6 +80,58 @@ mod tests { assert!(parse_transfer_ticket("not-a-ticket").is_err()); } + #[test] + fn ticket_rejects_unsupported_versions_and_mismatched_hashes() { + let secret = SecretKey::generate(); + let addr = iroh::EndpointAddr::new(secret.public()); + let blob_ticket = BlobTicket::new(addr, Hash::new([5; 32]), BlobFormat::HashSeq); + let payload = json!({ + "version": 2, + "blob_ticket": blob_ticket.to_string(), + "metadata": { + "version": 1, + "transfer_id": 7, + "transfer_name": "bad version", + "sender_name": null, + "created_at": 1, + "content_hash": blob_ticket.hash().to_string(), + "file_count": 1, + "total_size": 10 + } + }); + let encoded = format!( + "vnd1:{}", + BASE64URL_NOPAD.encode(payload.to_string().as_bytes()) + ); + assert!(parse_transfer_ticket(&encoded) + .unwrap_err() + .to_string() + .contains("unsupported VniDrop ticket version")); + + let payload = json!({ + "version": 1, + "blob_ticket": blob_ticket.to_string(), + "metadata": { + "version": 1, + "transfer_id": 7, + "transfer_name": "bad hash", + "sender_name": null, + "created_at": 1, + "content_hash": Hash::new([6; 32]).to_string(), + "file_count": 1, + "total_size": 10 + } + }); + let encoded = format!( + "vnd1:{}", + BASE64URL_NOPAD.encode(payload.to_string().as_bytes()) + ); + assert!(parse_transfer_ticket(&encoded) + .unwrap_err() + .to_string() + .contains("metadata hash does not match")); + } + #[tokio::test] async fn secret_persists() { let temp = tempfile::tempdir().unwrap(); @@ -90,6 +145,7 @@ mod tests { assert!(path_to_string(Path::new("../escape"), true).is_err()); assert!(path_to_string(Path::new("/absolute"), true).is_err()); assert!(validated_relative_string("bad\\name").is_err()); + assert!(validated_relative_string("").is_err()); } #[test] @@ -121,6 +177,61 @@ mod tests { assert_eq!(content, "fd-backed import"); } + #[cfg(unix)] + #[test] + fn file_descriptor_source_rejects_invalid_values() { + assert!(collect_import_files(vec![ShareSource { + kind: SourceKind::FileDescriptor, + value: "not-an-fd".to_string(), + display_name: Some("from-fd.txt".to_string()), + is_directory: false, + }]) + .is_err()); + + assert!(collect_import_files(vec![ShareSource { + kind: SourceKind::FileDescriptor, + value: "-1".to_string(), + display_name: Some("from-fd.txt".to_string()), + is_directory: false, + }]) + .is_err()); + } + + #[test] + fn android_content_uri_must_be_opened_by_platform_code() { + let error = collect_import_files(vec![ShareSource { + kind: SourceKind::AndroidContentUri, + value: "content://media/item".to_string(), + display_name: Some("from-uri.txt".to_string()), + is_directory: false, + }]) + .unwrap_err() + .to_string(); + assert!(error.contains("ParcelFileDescriptor")); + } + + #[test] + fn directory_sources_preserve_safe_relative_names() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("picked"); + std::fs::create_dir_all(root.join("nested")).unwrap(); + std::fs::write(root.join("nested").join("a.txt"), b"a").unwrap(); + std::fs::write(root.join("b.txt"), b"b").unwrap(); + + let mut files = collect_import_files(vec![ShareSource { + kind: SourceKind::Path, + value: root.to_string_lossy().to_string(), + display_name: Some("Album".to_string()), + is_directory: true, + }]) + .unwrap(); + files.sort_by(|a, b| a.collection_name.cmp(&b.collection_name)); + + assert_eq!(default_collection_name(&files), "Album"); + assert_eq!(files[0].collection_name, "Album/b.txt"); + assert_eq!(files[1].collection_name, "Album/nested/a.txt"); + } + #[test] fn can_initialize_core() { let temp = tempfile::tempdir().unwrap(); @@ -138,17 +249,18 @@ mod tests { async fn repository_persists_transfers_and_events() { let temp = tempfile::tempdir().unwrap(); let repository = Repository::open(temp.path()).await.unwrap(); + assert_eq!(repository.schema_version().await.unwrap(), 1); repository - .upsert_transfer( - 7, - "send", - "sharing", - Some("demo"), - Some("hash"), - Some("ticket"), - 1, - 12, - ) + .upsert_transfer(crate::repository::TransferUpsert { + transfer_id: 7, + direction: "send", + status: "sharing", + transfer_name: Some("demo"), + content_hash: Some("hash"), + ticket: Some("ticket"), + file_count: 1, + total_size: 12, + }) .await .unwrap(); repository @@ -172,6 +284,12 @@ mod tests { let events = repository.list_events(Some(7)).await.unwrap(); assert_eq!(events.len(), 1); assert_eq!(events[0].kind, "created"); + + let reopened = Repository::open(temp.path()).await.unwrap(); + let transfers = reopened.list_transfers().await.unwrap(); + assert_eq!(transfers.len(), 1); + let events = reopened.list_events(Some(7)).await.unwrap(); + assert_eq!(events[0].id, "event-1"); } #[tokio::test] @@ -193,5 +311,36 @@ mod tests { policy.decide(99, Some("node-a")).await, AccessDecision::Allow ); + assert_eq!( + policy.decide(99, None).await, + AccessDecision::Deny { + reason: "missing-endpoint-id" + } + ); + } + + #[test] + fn invalid_receive_ticket_is_typed_and_persisted_as_event() { + let temp = tempfile::tempdir().unwrap(); + let core = VnidropCore::initialize( + temp.path().to_string_lossy().to_string(), + Arc::new(TestSink), + ) + .unwrap(); + + let error = core + .receive( + "not-a-ticket".to_string(), + temp.path().to_string_lossy().to_string(), + None, + ) + .unwrap_err(); + assert!(matches!(error, VnidropError::Ticket { .. })); + + let events = core.list_events(None).unwrap(); + assert!(events + .iter() + .any(|event| event.phase == "error" && event.kind == "invalid-ticket")); + core.shutdown(); } } diff --git a/crates/vnidrop/src/ticket.rs b/crates/vnidrop/src/ticket.rs index af94e72..665f744 100644 --- a/crates/vnidrop/src/ticket.rs +++ b/crates/vnidrop/src/ticket.rs @@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize}; use crate::api::TransferMetadata; const VNIDROP_TICKET_PREFIX: &str = "vnd1:"; +const VNIDROP_TICKET_VERSION: u8 = 1; #[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct VnidropTicket { @@ -19,7 +20,7 @@ pub(crate) struct VnidropTicket { impl VnidropTicket { pub(crate) fn new(blob_ticket: BlobTicket, metadata: TransferMetadata) -> Self { Self { - version: 1, + version: VNIDROP_TICKET_VERSION, blob_ticket: blob_ticket.to_string(), metadata, } @@ -55,8 +56,26 @@ pub(crate) fn parse_transfer_ticket(value: &str) -> Result let normalized = normalize_ticket_input(value); if normalized.starts_with(VNIDROP_TICKET_PREFIX) { let ticket = VnidropTicket::decode(&normalized)?; + if ticket.version != VNIDROP_TICKET_VERSION { + anyhow::bail!("unsupported VniDrop ticket version {}", ticket.version); + } + if ticket.metadata.version != VNIDROP_TICKET_VERSION { + anyhow::bail!( + "unsupported VniDrop metadata version {}", + ticket.metadata.version + ); + } + if ticket.metadata.transfer_id == 0 { + anyhow::bail!("VniDrop ticket metadata is missing a valid transfer id"); + } + if ticket.metadata.transfer_name.trim().is_empty() { + anyhow::bail!("VniDrop ticket metadata is missing a transfer name"); + } let blob_ticket = BlobTicket::from_str(&ticket.blob_ticket) .context("invalid BlobTicket inside VniDrop ticket")?; + if ticket.metadata.content_hash != blob_ticket.hash().to_string() { + anyhow::bail!("VniDrop ticket metadata hash does not match BlobTicket hash"); + } return Ok(ParsedTransferTicket { blob_ticket, metadata: Some(ticket.metadata), @@ -71,5 +90,8 @@ pub(crate) fn parse_transfer_ticket(value: &str) -> Result } fn normalize_ticket_input(value: &str) -> String { + // Tickets are commonly copied from text views or chat apps that insert line + // breaks. Strip whitespace only; other corrupt characters should still be + // rejected by the base64 or BlobTicket decoders. value.chars().filter(|char| !char.is_whitespace()).collect() } diff --git a/crates/vnidrop/tests/local_transfer.rs b/crates/vnidrop/tests/local_transfer.rs index 2b10584..077a8cc 100644 --- a/crates/vnidrop/tests/local_transfer.rs +++ b/crates/vnidrop/tests/local_transfer.rs @@ -1,6 +1,9 @@ use std::sync::{Arc, Mutex}; -use vnidrop::{CoreEvent, CoreEventSink, ShareMetadataInput, ShareSource, SourceKind, VnidropCore}; +use vnidrop::{ + CoreEvent, CoreEventSink, ShareMetadataInput, ShareSource, SourceKind, TransferAccessMode, + VnidropCore, +}; #[derive(Default)] struct RecordingSink { @@ -13,6 +16,12 @@ impl CoreEventSink for RecordingSink { } } +impl RecordingSink { + fn events(&self) -> Vec { + self.events.lock().unwrap().clone() + } +} + #[test] fn two_local_cores_transfer_file() { let sender_dir = tempfile::tempdir().unwrap(); @@ -21,9 +30,11 @@ fn two_local_cores_transfer_file() { let source_path = sender_dir.path().join("hello.txt"); std::fs::write(&source_path, b"hello from vnidrop").unwrap(); + let sender_sink = Arc::new(RecordingSink::default()); + let receiver_sink = Arc::new(RecordingSink::default()); let sender = VnidropCore::initialize( sender_dir.path().join("core").to_string_lossy().to_string(), - Arc::new(RecordingSink::default()), + sender_sink.clone(), ) .unwrap(); let receiver = VnidropCore::initialize( @@ -32,7 +43,7 @@ fn two_local_cores_transfer_file() { .join("core") .to_string_lossy() .to_string(), - Arc::new(RecordingSink::default()), + receiver_sink.clone(), ) .unwrap(); @@ -68,3 +79,185 @@ fn two_local_cores_transfer_file() { sender.shutdown(); receiver.shutdown(); } + +#[test] +fn two_local_cores_transfer_directory() { + let sender_dir = tempfile::tempdir().unwrap(); + let receiver_dir = tempfile::tempdir().unwrap(); + let output_dir = tempfile::tempdir().unwrap(); + let source_root = sender_dir.path().join("photos"); + std::fs::create_dir_all(source_root.join("nested")).unwrap(); + std::fs::write(source_root.join("cover.txt"), b"cover").unwrap(); + std::fs::write(source_root.join("nested").join("inside.txt"), b"inside").unwrap(); + + let sender = VnidropCore::initialize( + sender_dir.path().join("core").to_string_lossy().to_string(), + Arc::new(RecordingSink::default()), + ) + .unwrap(); + let receiver = VnidropCore::initialize( + receiver_dir + .path() + .join("core") + .to_string_lossy() + .to_string(), + Arc::new(RecordingSink::default()), + ) + .unwrap(); + + let share = sender + .share_files( + vec![ShareSource { + kind: SourceKind::Path, + value: source_root.to_string_lossy().to_string(), + display_name: Some("photos".to_string()), + is_directory: true, + }], + ShareMetadataInput { + transfer_id: 8, + transfer_name: Some("photos".to_string()), + sender_name: Some("sender".to_string()), + }, + ) + .unwrap(); + + receiver + .receive( + share.ticket, + output_dir.path().to_string_lossy().to_string(), + Some("receiver".to_string()), + ) + .unwrap(); + + assert_eq!( + std::fs::read(output_dir.path().join("photos").join("cover.txt")).unwrap(), + b"cover" + ); + assert_eq!( + std::fs::read( + output_dir + .path() + .join("photos") + .join("nested") + .join("inside.txt") + ) + .unwrap(), + b"inside" + ); + + sender.shutdown(); + receiver.shutdown(); +} + +#[test] +fn approval_required_denies_then_allows_receiver() { + let sender_dir = tempfile::tempdir().unwrap(); + let receiver_dir = tempfile::tempdir().unwrap(); + let denied_output = tempfile::tempdir().unwrap(); + let allowed_output = tempfile::tempdir().unwrap(); + let source_path = sender_dir.path().join("private.txt"); + std::fs::write(&source_path, b"approved content").unwrap(); + + let sender_sink = Arc::new(RecordingSink::default()); + let sender = VnidropCore::initialize( + sender_dir.path().join("core").to_string_lossy().to_string(), + sender_sink.clone(), + ) + .unwrap(); + let receiver = VnidropCore::initialize( + receiver_dir + .path() + .join("core") + .to_string_lossy() + .to_string(), + Arc::new(RecordingSink::default()), + ) + .unwrap(); + + let share = sender + .share_files( + vec![ShareSource { + kind: SourceKind::Path, + value: source_path.to_string_lossy().to_string(), + display_name: Some("private.txt".to_string()), + is_directory: false, + }], + ShareMetadataInput { + transfer_id: 9, + transfer_name: Some("private".to_string()), + sender_name: None, + }, + ) + .unwrap(); + sender + .set_transfer_access_mode(share.transfer_id, TransferAccessMode::ApprovalRequired) + .unwrap(); + + assert!(receiver + .receive( + share.ticket.clone(), + denied_output.path().to_string_lossy().to_string(), + Some("receiver".to_string()), + ) + .is_err()); + assert!(sender_sink + .events() + .iter() + .any(|event| event.phase == "access" && event.kind == "request-denied")); + + sender + .approve_endpoint_for_transfer(share.transfer_id, receiver.status().endpoint_id) + .unwrap(); + receiver + .receive( + share.ticket, + allowed_output.path().to_string_lossy().to_string(), + Some("receiver".to_string()), + ) + .unwrap(); + assert_eq!( + std::fs::read(allowed_output.path().join("private.txt")).unwrap(), + b"approved content" + ); + + sender.shutdown(); + receiver.shutdown(); +} + +#[test] +fn cancelling_share_updates_status_and_events() { + let sender_dir = tempfile::tempdir().unwrap(); + let source_path = sender_dir.path().join("cancel.txt"); + std::fs::write(&source_path, b"cancel me").unwrap(); + let sink = Arc::new(RecordingSink::default()); + let sender = VnidropCore::initialize( + sender_dir.path().join("core").to_string_lossy().to_string(), + sink.clone(), + ) + .unwrap(); + + let share = sender + .share_files( + vec![ShareSource { + kind: SourceKind::Path, + value: source_path.to_string_lossy().to_string(), + display_name: Some("cancel.txt".to_string()), + is_directory: false, + }], + ShareMetadataInput { + transfer_id: 10, + transfer_name: Some("cancel".to_string()), + sender_name: None, + }, + ) + .unwrap(); + sender.cancel_transfer(share.transfer_id).unwrap(); + + let transfers = sender.list_transfers().unwrap(); + assert_eq!(transfers[0].status, "stopped"); + assert!(sink + .events() + .iter() + .any(|event| event.kind == "share-stopped")); + sender.shutdown(); +}