Build Rust core handshake and Compose UI foundation

crates/vnidrop/CORE_FLOW.md

@@ -0,0 +1,50 @@
+# VniDrop Core Send/Receive Flow
+
+This crate owns the transfer backend. Platform/UI code should pass file handles
+or paths into Rust and react to `CoreEvent` updates; it should not move file
+bytes through Kotlin memory.
+
+## Send
+
+1. `initialize(app_data_dir, event_sink)` starts the Iroh endpoint, blob
+   provider, handshake protocol, SQLite repository, and event hub.
+2. `share_files(sources, metadata)` validates platform sources, streams each
+   file into `iroh-blobs`, stores a collection, and returns a VniDrop ticket.
+3. New VniDrop shares are `ApprovalRequired` by default. A copied ticket is not
+   enough to read bytes until the sender approves the receiver endpoint.
+4. The sender observes receiver requests through `CoreEvent` entries with
+   `phase="approval"` and can query them with
+   `list_receiver_requests(transfer_id)`.
+5. `respond_receiver_request(request_id, accepted, reason)` accepts or refuses a
+   pending request. Accepted requests create a time-limited access session for
+   the receiver endpoint.
+
+## Receive
+
+1. `receive(ticket, output_dir, receiver_name)` parses and validates the ticket.
+2. VniDrop tickets first connect to the handshake ALPN
+   `/vnidrop/handshake/1` and send `RequestTransfer` metadata to the sender.
+3. If approved, the receiver connects to the blobs ALPN, downloads the
+   collection, and streams files to `output_dir`.
+4. If refused, expired, unknown, or cancelled, the receive transfer is marked
+   `failed` or `cancelled` and emits an error/lifecycle event.
+5. Legacy raw `BlobTicket` values do not carry VniDrop metadata, so they bypass
+   the app approval handshake and use the underlying blob ticket directly.
+
+## Core States And Events
+
+- Transfer statuses: `sharing`, `receiving`, `done`, `failed`, `cancelled`,
+  `stopped`.
+- Main event phases: `endpoint`, `import`, `ticket`, `handshake`, `approval`,
+  `access`, `transfer`, `download`, `export`, `lifecycle`, `error`.
+- Events are sent to `CoreEventSink` immediately and persisted through the event
+  hub. `list_events` flushes queued persistence before reading SQLite.
+- `shutdown()` is idempotent and flushes events before stopping the router.
+
+## Platform File Rules
+
+- Desktop uses normal filesystem paths.
+- Android opens SAF/content URIs in Kotlin and passes a borrowed file
+  descriptor; Rust duplicates the descriptor before streaming.
+- iOS starts the security-scoped URL lease in Kotlin and keeps it alive while
+  Rust streams from the accessible file URL/path.

crates/vnidrop/Cargo.toml

@@ -17,6 +17,7 @@ futures-lite = "2.6.1"
 iroh = "1.0.0"
 iroh-blobs = "0.103.0"
 irpc = "0.17.0"
+irpc-iroh = "0.17.0"
 libc = "0.2.186"
 n0-future = "0.3.1"
 num_cpus = "1.17.0"

crates/vnidrop/src/access_policy.rs

@@ -1,11 +1,9 @@
-use std::{
-    collections::{HashMap, HashSet},
-    sync::Arc,
-};
+use std::{collections::HashMap, sync::Arc};
 
 use tokio::sync::RwLock;
 
 use crate::api::TransferAccessMode;
+use crate::util::now_ms;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum AccessDecision {
@@ -16,7 +14,7 @@ pub(crate) enum AccessDecision {
 #[derive(Debug, Default)]
 pub(crate) struct AccessPolicy {
     modes: RwLock<HashMap<u64, TransferAccessMode>>,
-    approved_sessions: RwLock<HashSet<(u64, String)>>,
+    approved_sessions: RwLock<HashMap<(u64, String), ApprovalSession>>,
 }
 
 impl AccessPolicy {
@@ -33,14 +31,24 @@ impl AccessPolicy {
         self.approved_sessions
             .write()
             .await
-            .retain(|(id, _)| *id != transfer_id);
+            .retain(|(id, _), _| *id != transfer_id);
     }
 
     pub(crate) async fn approve_endpoint(&self, transfer_id: u64, endpoint_id: String) {
+        self.approve_endpoint_until(transfer_id, endpoint_id, None)
+            .await;
+    }
+
+    pub(crate) async fn approve_endpoint_until(
+        &self,
+        transfer_id: u64,
+        endpoint_id: String,
+        expires_at: Option<i64>,
+    ) {
         self.approved_sessions
             .write()
             .await
-            .insert((transfer_id, endpoint_id));
+            .insert((transfer_id, endpoint_id), ApprovalSession { expires_at });
     }
 
     pub(crate) async fn decide(
@@ -48,9 +56,6 @@ 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()
@@ -66,19 +71,32 @@ impl AccessPolicy {
                         reason: "missing-endpoint-id",
                     };
                 };
-                if self
-                    .approved_sessions
-                    .read()
-                    .await
-                    .contains(&(transfer_id, endpoint_id.to_string()))
-                {
-                    AccessDecision::Allow
-                } else {
-                    AccessDecision::Deny {
-                        reason: "approval-required",
+                let key = (transfer_id, endpoint_id.to_string());
+                let mut sessions = self.approved_sessions.write().await;
+                match sessions.get(&key) {
+                    Some(session) if session.is_valid(now_ms()) => AccessDecision::Allow,
+                    Some(_) => {
+                        sessions.remove(&key);
+                        AccessDecision::Deny {
+                            reason: "approval-expired",
+                        }
                     }
+                    None => AccessDecision::Deny {
+                        reason: "approval-required",
+                    },
                 }
             }
         }
     }
 }
+
+#[derive(Debug, Clone)]
+struct ApprovalSession {
+    expires_at: Option<i64>,
+}
+
+impl ApprovalSession {
+    fn is_valid(&self, now: i64) -> bool {
+        self.expires_at.is_none_or(|expires_at| expires_at >= now)
+    }
+}

crates/vnidrop/src/api.rs

@@ -122,3 +122,18 @@ pub struct TicketInspection {
     pub blob_ticket: String,
     pub metadata: Option<TransferMetadata>,
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
+pub struct ReceiverRequest {
+    pub id: String,
+    pub transfer_id: u64,
+    pub remote_endpoint_id: String,
+    pub transfer_name: String,
+    pub receiver_name: Option<String>,
+    pub receiver_device_name: Option<String>,
+    pub app_version: String,
+    pub status: String,
+    pub reason: Option<String>,
+    pub requested_at: i64,
+    pub responded_at: Option<i64>,
+}

crates/vnidrop/src/approval.rs

@@ -0,0 +1,221 @@
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::{oneshot, Mutex};
+use uuid::Uuid;
+
+use crate::{
+    access_policy::AccessPolicy,
+    event_hub::EventHub,
+    handshake::{HandshakeResponse, RequestTransfer},
+    repository::{ReceiverRequestInsert, Repository},
+    util::now_ms,
+};
+
+const APPROVAL_TTL_MS: i64 = 10 * 60 * 1000;
+const APPROVAL_WAIT_TIMEOUT: Duration = Duration::from_secs(120);
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct ApprovalDecision {
+    pub(crate) request_id: String,
+    pub(crate) accepted: bool,
+    pub(crate) reason: Option<String>,
+}
+
+#[derive(Clone)]
+pub(crate) struct ApprovalService {
+    repository: Repository,
+    event_hub: Arc<EventHub>,
+    access_policy: Arc<AccessPolicy>,
+    pending: Arc<Mutex<HashMap<String, oneshot::Sender<ApprovalDecision>>>>,
+}
+
+impl ApprovalService {
+    pub(crate) fn new(
+        repository: Repository,
+        event_hub: Arc<EventHub>,
+        access_policy: Arc<AccessPolicy>,
+    ) -> Self {
+        Self {
+            repository,
+            event_hub,
+            access_policy,
+            pending: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    pub(crate) async fn respond(
+        &self,
+        request_id: String,
+        accepted: bool,
+        reason: Option<String>,
+    ) -> anyhow::Result<()> {
+        let sender = self.pending.lock().await.remove(&request_id);
+        let status = if accepted { "accepted" } else { "refused" };
+        self.repository
+            .update_receiver_request_status(&request_id, status, reason.as_deref())
+            .await?;
+
+        if let Some(sender) = sender {
+            let _ = sender.send(ApprovalDecision {
+                request_id,
+                accepted,
+                reason,
+            });
+        }
+
+        Ok(())
+    }
+
+    pub(crate) async fn request_transfer(
+        &self,
+        remote_endpoint_id: String,
+        request: RequestTransfer,
+    ) -> HandshakeResponse {
+        self.event_hub.emit_transfer(
+            request.transfer_id,
+            "send",
+            "handshake",
+            "transfer-requested",
+            json!({
+                "remote_endpoint_id": remote_endpoint_id,
+                "request": request,
+            }),
+        );
+
+        match self
+            .repository
+            .send_exists(request.transfer_id, &request.transfer_hash)
+            .await
+        {
+            Ok(true) => {
+                self.wait_for_sender_decision(remote_endpoint_id, request)
+                    .await
+            }
+            Ok(false) => {
+                self.deny(request.transfer_id, remote_endpoint_id, "unknown-transfer")
+                    .await
+            }
+            Err(error) => {
+                tracing::error!(%error, "failed to validate handshake transfer request");
+                self.deny(request.transfer_id, remote_endpoint_id, "repository-error")
+                    .await
+            }
+        }
+    }
+
+    async fn wait_for_sender_decision(
+        &self,
+        remote_endpoint_id: String,
+        request: RequestTransfer,
+    ) -> HandshakeResponse {
+        let request_id = Uuid::new_v4().to_string();
+        let (tx, rx) = oneshot::channel();
+        self.pending.lock().await.insert(request_id.clone(), tx);
+
+        let insert_result = self
+            .repository
+            .insert_receiver_request(ReceiverRequestInsert {
+                id: &request_id,
+                transfer_id: request.transfer_id,
+                remote_endpoint_id: &remote_endpoint_id,
+                transfer_name: &request.transfer_name,
+                receiver_name: request.receiver_name.as_deref(),
+                receiver_device_name: request.receiver_device_name.as_deref(),
+                app_version: &request.app_version,
+            })
+            .await;
+
+        if let Err(error) = insert_result {
+            self.pending.lock().await.remove(&request_id);
+            tracing::error!(%error, "failed to persist receiver request");
+            return self
+                .deny(request.transfer_id, remote_endpoint_id, "repository-error")
+                .await;
+        }
+
+        self.event_hub.emit_transfer(
+            request.transfer_id,
+            "send",
+            "approval",
+            "receiver-requested",
+            json!({
+                "request_id": request_id,
+                "remote_endpoint_id": remote_endpoint_id,
+                "receiver_name": request.receiver_name,
+                "receiver_device_name": request.receiver_device_name,
+                "transfer_name": request.transfer_name,
+            }),
+        );
+
+        match tokio::time::timeout(APPROVAL_WAIT_TIMEOUT, rx).await {
+            Ok(Ok(decision)) if decision.accepted => {
+                let token = Uuid::new_v4().to_string();
+                let expires_at = now_ms() + APPROVAL_TTL_MS;
+                self.access_policy
+                    .approve_endpoint_until(
+                        request.transfer_id,
+                        remote_endpoint_id.clone(),
+                        Some(expires_at),
+                    )
+                    .await;
+                self.event_hub.emit_transfer(
+                    request.transfer_id,
+                    "send",
+                    "approval",
+                    "receiver-accepted",
+                    json!({
+                        "request_id": decision.request_id,
+                        "remote_endpoint_id": remote_endpoint_id,
+                        "expires_at": expires_at,
+                    }),
+                );
+                HandshakeResponse::Approved { token, expires_at }
+            }
+            Ok(Ok(decision)) => {
+                self.deny(
+                    request.transfer_id,
+                    remote_endpoint_id,
+                    decision
+                        .reason
+                        .unwrap_or_else(|| "sender-refused".to_string()),
+                )
+                .await
+            }
+            Ok(Err(_)) | Err(_) => {
+                self.pending.lock().await.remove(&request_id);
+                let _ = self
+                    .repository
+                    .update_receiver_request_status(
+                        &request_id,
+                        "expired",
+                        Some("approval timed out"),
+                    )
+                    .await;
+                self.deny(request.transfer_id, remote_endpoint_id, "approval-timeout")
+                    .await
+            }
+        }
+    }
+
+    async fn deny(
+        &self,
+        transfer_id: u64,
+        remote_endpoint_id: String,
+        reason: impl Into<String>,
+    ) -> HandshakeResponse {
+        let reason = reason.into();
+        self.event_hub.emit_transfer(
+            transfer_id,
+            "send",
+            "approval",
+            "receiver-refused",
+            json!({
+                "remote_endpoint_id": remote_endpoint_id,
+                "reason": reason,
+            }),
+        );
+        HandshakeResponse::Denied { reason }
+    }
+}

crates/vnidrop/src/handshake.rs

@@ -0,0 +1,120 @@
+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 crate::api::TransferMetadata;
+
+#[derive(Clone)]
+pub(crate) struct HandshakeService {
+    approval: crate::approval::ApprovalService,
+}
+
+impl fmt::Debug for HandshakeService {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("HandshakeService")
+    }
+}
+
+impl HandshakeService {
+    pub(crate) const ALPN: &'static [u8] = b"/vnidrop/handshake/1";
+
+    pub(crate) fn new(approval: crate::approval::ApprovalService) -> Self {
+        Self { approval }
+    }
+
+    pub(crate) fn client(endpoint: Endpoint, addr: EndpointAddr) -> HandshakeClient {
+        HandshakeClient {
+            inner: Client::boxed(IrohLazyRemoteConnection::new(
+                endpoint,
+                addr,
+                Self::ALPN.to_vec(),
+            )),
+        }
+    }
+
+    async fn handle_request(
+        &self,
+        remote_endpoint_id: String,
+        request: RequestTransfer,
+    ) -> HandshakeResponse {
+        self.approval
+            .request_transfer(remote_endpoint_id, request)
+            .await
+    }
+}
+
+impl ProtocolHandler for HandshakeService {
+    async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
+        let remote_endpoint_id = connection.remote_id().to_string();
+
+        while let Some(message) = read_request::<HandshakeProtocol>(&connection).await? {
+            match message {
+                HandshakeMessage::RequestTransfer(message) => {
+                    let WithChannels { inner, tx, .. } = message;
+                    // The receiver-provided name is display data. The trusted
+                    // identity is the endpoint id from the Iroh connection.
+                    let response = self.handle_request(remote_endpoint_id.clone(), inner).await;
+                    let _ = tx.send(response).await;
+                }
+            }
+        }
+
+        connection.closed().await;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct HandshakeClient {
+    inner: Client<HandshakeProtocol>,
+}
+
+impl HandshakeClient {
+    pub(crate) async fn request_transfer(
+        &self,
+        metadata: &TransferMetadata,
+        receiver_name: Option<&str>,
+    ) -> Result<HandshakeResponse, irpc::Error> {
+        self.inner
+            .rpc(RequestTransfer {
+                transfer_id: metadata.transfer_id,
+                transfer_hash: metadata.content_hash.clone(),
+                transfer_name: metadata.transfer_name.clone(),
+                receiver_name: receiver_name.map(ToOwned::to_owned),
+                receiver_device_name: None,
+                app_version: env!("CARGO_PKG_VERSION").to_string(),
+            })
+            .await
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct RequestTransfer {
+    pub(crate) transfer_id: u64,
+    pub(crate) transfer_hash: String,
+    pub(crate) transfer_name: String,
+    pub(crate) receiver_name: Option<String>,
+    pub(crate) receiver_device_name: Option<String>,
+    pub(crate) app_version: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub(crate) enum HandshakeResponse {
+    Approved { token: String, expires_at: i64 },
+    Denied { reason: String },
+}
+
+#[rpc_requests(message = HandshakeMessage)]
+#[derive(Debug, Serialize, Deserialize)]
+enum HandshakeProtocol {
+    #[rpc(tx=oneshot::Sender<HandshakeResponse>)]
+    RequestTransfer(RequestTransfer),
+}

crates/vnidrop/src/lib.rs

@@ -1,8 +1,10 @@
 mod access_policy;
 mod api;
+mod approval;
 mod error;
 mod event_hub;
 mod filesystem;
+mod handshake;
 mod logging;
 mod repository;
 mod runtime;
@@ -11,8 +13,9 @@ mod ticket;
 mod util;
 
 pub use api::{
-    CoreEvent, CoreEventSink, RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource,
-    SourceKind, StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata,
+    CoreEvent, CoreEventSink, ReceiverRequest, RuntimeStatus, ShareMetadataInput, ShareResult,
+    ShareSource, SourceKind, StoredTransfer, TicketInspection, TransferAccessMode,
+    TransferMetadata,
 };
 pub use error::VnidropError;
 pub use runtime::VnidropCore;

crates/vnidrop/src/repository.rs

@@ -6,7 +6,7 @@ use sqlx::{
     Row, SqlitePool,
 };
 
-use crate::api::{CoreEvent, StoredTransfer};
+use crate::api::{CoreEvent, ReceiverRequest, StoredTransfer};
 use crate::util::now_ms;
 
 const SCHEMA_VERSION: i64 = 1;
@@ -27,6 +27,16 @@ pub(crate) struct TransferUpsert<'a> {
     pub(crate) total_size: u64,
 }
 
+pub(crate) struct ReceiverRequestInsert<'a> {
+    pub(crate) id: &'a str,
+    pub(crate) transfer_id: u64,
+    pub(crate) remote_endpoint_id: &'a str,
+    pub(crate) transfer_name: &'a str,
+    pub(crate) receiver_name: Option<&'a str>,
+    pub(crate) receiver_device_name: Option<&'a str>,
+    pub(crate) app_version: &'a str,
+}
+
 impl Repository {
     pub(crate) async fn open(app_data_dir: &Path) -> Result<Self> {
         let db_path = app_data_dir.join("vnidrop.sqlite3");
@@ -87,6 +97,32 @@ impl Repository {
         .execute(&self.pool)
         .await?;
 
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS receiver_requests (
+                id TEXT PRIMARY KEY,
+                transfer_id INTEGER NOT NULL,
+                remote_endpoint_id TEXT NOT NULL,
+                transfer_name TEXT NOT NULL,
+                receiver_name TEXT,
+                receiver_device_name TEXT,
+                app_version TEXT NOT NULL,
+                status TEXT NOT NULL,
+                reason TEXT,
+                requested_at INTEGER NOT NULL,
+                responded_at INTEGER
+            );
+            "#,
+        )
+        .execute(&self.pool)
+        .await?;
+
+        sqlx::query(
+            "CREATE INDEX IF NOT EXISTS idx_receiver_requests_transfer_id ON receiver_requests(transfer_id, requested_at DESC);",
+        )
+        .execute(&self.pool)
+        .await?;
+
         sqlx::query(&format!("PRAGMA user_version = {SCHEMA_VERSION}"))
             .execute(&self.pool)
             .await?;
@@ -171,6 +207,98 @@ impl Repository {
         Ok(())
     }
 
+    pub(crate) async fn insert_receiver_request(
+        &self,
+        request: ReceiverRequestInsert<'_>,
+    ) -> Result<()> {
+        sqlx::query(
+            r#"
+            INSERT INTO receiver_requests (
+                id, transfer_id, remote_endpoint_id, transfer_name,
+                receiver_name, receiver_device_name, app_version, status,
+                reason, requested_at, responded_at
+            )
+            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'requested', NULL, ?8, NULL)
+            "#,
+        )
+        .bind(request.id)
+        .bind(request.transfer_id as i64)
+        .bind(request.remote_endpoint_id)
+        .bind(request.transfer_name)
+        .bind(request.receiver_name)
+        .bind(request.receiver_device_name)
+        .bind(request.app_version)
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    pub(crate) async fn update_receiver_request_status(
+        &self,
+        id: &str,
+        status: &str,
+        reason: Option<&str>,
+    ) -> Result<()> {
+        let result = sqlx::query(
+            r#"
+            UPDATE receiver_requests
+            SET status = ?1, reason = ?2, responded_at = ?3
+            WHERE id = ?4
+              AND status = 'requested'
+            "#,
+        )
+        .bind(status)
+        .bind(reason)
+        .bind(now_ms())
+        .bind(id)
+        .execute(&self.pool)
+        .await?;
+        if result.rows_affected() == 0 {
+            anyhow::bail!("receiver request not found or already handled");
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn list_receiver_requests(
+        &self,
+        transfer_id: u64,
+    ) -> Result<Vec<ReceiverRequest>> {
+        let rows = sqlx::query(
+            r#"
+            SELECT id, transfer_id, remote_endpoint_id, transfer_name,
+                   receiver_name, receiver_device_name, app_version, status,
+                   reason, requested_at, responded_at
+            FROM receiver_requests
+            WHERE transfer_id = ?1
+            ORDER BY requested_at DESC
+            "#,
+        )
+        .bind(transfer_id as i64)
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows.into_iter().map(row_to_receiver_request).collect())
+    }
+
+    pub(crate) async fn send_exists(&self, transfer_id: u64, content_hash: &str) -> Result<bool> {
+        let row = sqlx::query(
+            r#"
+            SELECT EXISTS(
+                SELECT 1 FROM transfers
+                WHERE transfer_id = ?1
+                  AND content_hash = ?2
+                  AND direction = 'send'
+                  AND status = 'sharing'
+            )
+            "#,
+        )
+        .bind(transfer_id as i64)
+        .bind(content_hash)
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(row.get::<i64, _>(0) != 0)
+    }
+
     pub(crate) async fn list_transfers(&self) -> Result<Vec<StoredTransfer>> {
         let rows = sqlx::query(
             r#"
@@ -243,3 +371,19 @@ fn row_to_event(row: sqlx::sqlite::SqliteRow) -> CoreEvent {
         data_json: row.get("data_json"),
     }
 }
+
+fn row_to_receiver_request(row: sqlx::sqlite::SqliteRow) -> ReceiverRequest {
+    ReceiverRequest {
+        id: row.get("id"),
+        transfer_id: row.get::<i64, _>("transfer_id") as u64,
+        remote_endpoint_id: row.get("remote_endpoint_id"),
+        transfer_name: row.get("transfer_name"),
+        receiver_name: row.get("receiver_name"),
+        receiver_device_name: row.get("receiver_device_name"),
+        app_version: row.get("app_version"),
+        status: row.get("status"),
+        reason: row.get("reason"),
+        requested_at: row.get("requested_at"),
+        responded_at: row.get("responded_at"),
+    }
+}

crates/vnidrop/src/runtime.rs

@@ -32,9 +32,10 @@ use tokio::{
 use crate::{
     access_policy::{AccessDecision, AccessPolicy},
     api::{
-        CoreEvent, CoreEventSink, RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource,
-        StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata,
+        CoreEvent, CoreEventSink, ReceiverRequest, RuntimeStatus, ShareMetadataInput, ShareResult,
+        ShareSource, StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata,
     },
+    approval::ApprovalService,
     error::VnidropError,
     event_hub::EventHub,
     filesystem::{
@@ -42,6 +43,7 @@ use crate::{
         read_stream_from_blocking_reader, safe_output_path, wait_for_writer,
         write_stream_to_blocking_writer, TransferImport,
     },
+    handshake::{HandshakeResponse, HandshakeService},
     logging::init_logging,
     repository::{Repository, TransferUpsert},
     secret::load_or_create_secret,
@@ -69,7 +71,8 @@ struct CoreInner {
     router: Router,
     store: FsStore,
     repository: Repository,
-    event_hub: EventHub,
+    event_hub: Arc<EventHub>,
+    approval: ApprovalService,
     access_policy: Arc<AccessPolicy>,
     active_transfers: TokioMutex<HashMap<u64, oneshot::Sender<()>>>,
     active_shares: TokioMutex<HashMap<u64, TempTag>>,
@@ -165,6 +168,26 @@ impl VnidropCore {
             .map_err(VnidropError::permission)
     }
 
+    pub fn list_receiver_requests(
+        &self,
+        transfer_id: u64,
+    ) -> Result<Vec<ReceiverRequest>, VnidropError> {
+        self.runtime
+            .block_on(self.inner.repository.list_receiver_requests(transfer_id))
+            .map_err(VnidropError::repository)
+    }
+
+    pub fn respond_receiver_request(
+        &self,
+        request_id: String,
+        accepted: bool,
+        reason: Option<String>,
+    ) -> Result<(), VnidropError> {
+        self.runtime
+            .block_on(self.inner.approval.respond(request_id, accepted, reason))
+            .map_err(VnidropError::permission)
+    }
+
     pub fn list_transfers(&self) -> Result<Vec<StoredTransfer>, VnidropError> {
         self.runtime
             .block_on(self.inner.repository.list_transfers())
@@ -215,10 +238,15 @@ impl CoreInner {
         // 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 event_hub = Arc::new(EventHub::start(repository.clone(), event_sink));
+        let access_policy = AccessPolicy::new();
+        let approval =
+            ApprovalService::new(repository.clone(), event_hub.clone(), access_policy.clone());
+        let handshake = HandshakeService::new(approval.clone());
         let router = Router::builder(endpoint.clone())
             .accept(iroh_blobs::ALPN, blobs)
+            .accept(HandshakeService::ALPN, handshake)
             .spawn();
-        let event_hub = EventHub::start(repository.clone(), event_sink);
 
         let inner = Arc::new(Self {
             endpoint,
@@ -226,7 +254,8 @@ impl CoreInner {
             store,
             repository,
             event_hub,
-            access_policy: AccessPolicy::new(),
+            approval,
+            access_policy,
             active_transfers: TokioMutex::new(HashMap::new()),
             active_shares: TokioMutex::new(HashMap::new()),
             hash_to_transfer: TokioMutex::new(HashMap::new()),
@@ -320,7 +349,7 @@ impl CoreInner {
             .await
             .insert(import.root_hash.to_string(), metadata.transfer_id);
         self.access_policy
-            .set_mode(metadata.transfer_id, TransferAccessMode::Public)
+            .set_mode(metadata.transfer_id, TransferAccessMode::ApprovalRequired)
             .await;
         self.active_shares
             .lock()
@@ -464,6 +493,15 @@ impl CoreInner {
         tokio::fs::create_dir_all(&output_dir).await?;
 
         self.emit_transfer(transfer_id, "receive", "network", "connecting", json!({}));
+        if let Some(metadata) = &parsed.metadata {
+            self.request_transfer_approval(
+                transfer_id,
+                parsed.blob_ticket.addr().clone(),
+                metadata,
+                receiver_name.as_deref(),
+            )
+            .await?;
+        }
         let connection = self
             .endpoint
             .connect(parsed.blob_ticket.addr().clone(), iroh_blobs::ALPN)
@@ -583,6 +621,49 @@ impl CoreInner {
         Ok(())
     }
 
+    async fn request_transfer_approval(
+        &self,
+        local_transfer_id: u64,
+        addr: iroh::EndpointAddr,
+        metadata: &TransferMetadata,
+        receiver_name: Option<&str>,
+    ) -> Result<()> {
+        self.emit_transfer(
+            local_transfer_id,
+            "receive",
+            "handshake",
+            "approval-requesting",
+            json!({
+                "sender_transfer_id": metadata.transfer_id,
+                "metadata": metadata,
+            }),
+        );
+
+        let client = HandshakeService::client(self.endpoint.clone(), addr);
+        match client
+            .request_transfer(metadata, receiver_name)
+            .await
+            .map_err(|error| anyhow::anyhow!("handshake request failed: {error}"))?
+        {
+            HandshakeResponse::Approved { expires_at, .. } => {
+                self.emit_transfer(
+                    local_transfer_id,
+                    "receive",
+                    "handshake",
+                    "approval-granted",
+                    json!({
+                        "sender_transfer_id": metadata.transfer_id,
+                        "expires_at": expires_at,
+                    }),
+                );
+                Ok(())
+            }
+            HandshakeResponse::Denied { reason } => {
+                anyhow::bail!("transfer request was denied by sender: {reason}")
+            }
+        }
+    }
+
     async fn shutdown(&self) {
         if self.shutdown_started.swap(true, Ordering::SeqCst) {
             return;

crates/vnidrop/src/tests.rs

@@ -17,7 +17,7 @@ mod tests {
             collect_import_files, default_collection_name, path_to_string,
             percent_decode_file_url_path, validated_relative_string,
         },
-        repository::Repository,
+        repository::{ReceiverRequestInsert, Repository},
         runtime::VnidropCore,
         secret::load_or_create_secret,
         ticket::{parse_transfer_ticket, VnidropTicket},
@@ -292,6 +292,42 @@ mod tests {
         assert_eq!(events[0].id, "event-1");
     }
 
+    #[tokio::test]
+    async fn repository_persists_receiver_requests() {
+        let temp = tempfile::tempdir().unwrap();
+        let repository = Repository::open(temp.path()).await.unwrap();
+        repository
+            .insert_receiver_request(ReceiverRequestInsert {
+                id: "request-1",
+                transfer_id: 77,
+                remote_endpoint_id: "node-a",
+                transfer_name: "demo",
+                receiver_name: Some("receiver"),
+                receiver_device_name: Some("phone"),
+                app_version: "0.1.0",
+            })
+            .await
+            .unwrap();
+        repository
+            .update_receiver_request_status("request-1", "accepted", None)
+            .await
+            .unwrap();
+        assert!(repository
+            .update_receiver_request_status("request-1", "refused", Some("late"))
+            .await
+            .is_err());
+        assert!(repository
+            .update_receiver_request_status("missing", "accepted", None)
+            .await
+            .is_err());
+
+        let requests = repository.list_receiver_requests(77).await.unwrap();
+        assert_eq!(requests.len(), 1);
+        assert_eq!(requests[0].status, "accepted");
+        assert_eq!(requests[0].receiver_name.as_deref(), Some("receiver"));
+        assert!(requests[0].responded_at.is_some());
+    }
+
     #[tokio::test]
     async fn access_policy_requires_approved_endpoint_when_locked() {
         let policy = AccessPolicy::new();
@@ -319,6 +355,30 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn access_policy_rejects_expired_approval_sessions() {
+        let policy = AccessPolicy::new();
+        policy
+            .set_mode(100, TransferAccessMode::ApprovalRequired)
+            .await;
+        policy
+            .approve_endpoint_until(100, "node-a".to_string(), Some(crate::util::now_ms() - 1))
+            .await;
+
+        assert_eq!(
+            policy.decide(100, Some("node-a")).await,
+            AccessDecision::Deny {
+                reason: "approval-expired"
+            }
+        );
+        assert_eq!(
+            policy.decide(100, Some("node-a")).await,
+            AccessDecision::Deny {
+                reason: "approval-required"
+            }
+        );
+    }
+
     #[test]
     fn invalid_receive_ticket_is_typed_and_persisted_as_event() {
         let temp = tempfile::tempdir().unwrap();

crates/vnidrop/tests/local_transfer.rs

@@ -1,7 +1,10 @@
-use std::sync::{Arc, Mutex};
+use std::{
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
 
 use vnidrop::{
-    CoreEvent, CoreEventSink, ShareMetadataInput, ShareSource, SourceKind, TransferAccessMode,
+    CoreEvent, CoreEventSink, ReceiverRequest, ShareMetadataInput, ShareSource, SourceKind,
     VnidropCore,
 };
 
@@ -22,6 +25,49 @@ impl RecordingSink {
     }
 }
 
+fn wait_for_receiver_request(sender: &VnidropCore, transfer_id: u64) -> ReceiverRequest {
+    let started = Instant::now();
+    loop {
+        let requests = sender.list_receiver_requests(transfer_id).unwrap();
+        if let Some(request) = requests
+            .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(50));
+    }
+}
+
+fn receive_with_response(
+    sender: &VnidropCore,
+    transfer_id: u64,
+    receiver: Arc<VnidropCore>,
+    ticket: String,
+    output_dir: String,
+    receiver_name: Option<String>,
+    accepted: bool,
+) -> Result<(), String> {
+    let handle = std::thread::spawn(move || {
+        receiver
+            .receive(ticket, output_dir, receiver_name)
+            .map_err(|error| error.to_string())
+    });
+    let request = wait_for_receiver_request(sender, transfer_id);
+    sender
+        .respond_receiver_request(
+            request.id,
+            accepted,
+            (!accepted).then(|| "sender-refused".to_string()),
+        )
+        .unwrap();
+    handle.join().unwrap()
+}
+
 #[test]
 fn two_local_cores_transfer_file() {
     let sender_dir = tempfile::tempdir().unwrap();
@@ -63,13 +109,16 @@ fn two_local_cores_transfer_file() {
         )
         .unwrap();
 
-    receiver
-        .receive(
-            share.ticket,
-            output_dir.path().to_string_lossy().to_string(),
-            Some("receiver".to_string()),
-        )
-        .unwrap();
+    receive_with_response(
+        &sender,
+        share.transfer_id,
+        receiver.clone(),
+        share.ticket,
+        output_dir.path().to_string_lossy().to_string(),
+        Some("receiver".to_string()),
+        true,
+    )
+    .unwrap();
 
     assert_eq!(
         std::fs::read(output_dir.path().join("hello.txt")).unwrap(),
@@ -121,13 +170,16 @@ fn two_local_cores_transfer_directory() {
         )
         .unwrap();
 
-    receiver
-        .receive(
-            share.ticket,
-            output_dir.path().to_string_lossy().to_string(),
-            Some("receiver".to_string()),
-        )
-        .unwrap();
+    receive_with_response(
+        &sender,
+        share.transfer_id,
+        receiver.clone(),
+        share.ticket,
+        output_dir.path().to_string_lossy().to_string(),
+        Some("receiver".to_string()),
+        true,
+    )
+    .unwrap();
 
     assert_eq!(
         std::fs::read(output_dir.path().join("photos").join("cover.txt")).unwrap(),
@@ -189,32 +241,31 @@ fn approval_required_denies_then_allows_receiver() {
             },
         )
         .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!(receive_with_response(
+        &sender,
+        share.transfer_id,
+        receiver.clone(),
+        share.ticket.clone(),
+        denied_output.path().to_string_lossy().to_string(),
+        Some("receiver".to_string()),
+        false,
+    )
+    .is_err());
     assert!(sender_sink
         .events()
         .iter()
-        .any(|event| event.phase == "access" && event.kind == "request-denied"));
+        .any(|event| event.phase == "approval" && event.kind == "receiver-refused"));
 
-    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();
+    receive_with_response(
+        &sender,
+        share.transfer_id,
+        receiver.clone(),
+        share.ticket,
+        allowed_output.path().to_string_lossy().to_string(),
+        Some("receiver".to_string()),
+        true,
+    )
+    .unwrap();
     assert_eq!(
         std::fs::read(allowed_output.path().join("private.txt")).unwrap(),
         b"approved content"