From 193fe7c75792d634464995ee0601a185c163e6e4 Mon Sep 17 00:00:00 2001 From: Hammed Abass Date: Tue, 11 Aug 2026 03:03:43 +0200 Subject: [PATCH] feat(core): save devices through mutual consent and grants Establish PendingOutgoing/PendingIncoming relationships over a token-bound pairing protocol, exchange directional grants with challenge-response proofs and a final ack before Saved, and merge simultaneous initiations without a second prompt. Co-authored-by: Cursor --- .../vnidrop/src/device_relationship/crypto.rs | 134 ++ crates/vnidrop/src/device_relationship/mod.rs | 1306 +++++++++++++++++ crates/vnidrop/src/grant.rs | 36 + crates/vnidrop/src/lib.rs | 1 + crates/vnidrop/src/pairing_eligibility.rs | 61 +- crates/vnidrop/src/repository.rs | 7 +- crates/vnidrop/src/runtime/contacts.rs | 26 +- crates/vnidrop/src/runtime/facade.rs | 21 + crates/vnidrop/src/runtime/mod.rs | 18 + crates/vnidrop/src/secure_secret.rs | 8 + crates/vnidrop/src/tests.rs | 2 + .../vnidrop/src/tests/device_relationship.rs | 317 ++++ crates/vnidrop/src/tests/repository.rs | 4 +- 13 files changed, 1933 insertions(+), 8 deletions(-) create mode 100644 crates/vnidrop/src/device_relationship/crypto.rs create mode 100644 crates/vnidrop/src/device_relationship/mod.rs create mode 100644 crates/vnidrop/src/tests/device_relationship.rs diff --git a/crates/vnidrop/src/device_relationship/crypto.rs b/crates/vnidrop/src/device_relationship/crypto.rs new file mode 100644 index 0000000..0591876 --- /dev/null +++ b/crates/vnidrop/src/device_relationship/crypto.rs @@ -0,0 +1,134 @@ +//! Relationship-grant possession proofs (design §7). + +use crate::{ + error::VnidropError, + grant::{Challenge, GrantId, GrantProof, GrantSecret}, + secure_secret::SecretMaterial, +}; + +const RELATIONSHIP_GRANT_CONTEXT: &[u8] = b"vnidrop-relationship-grant-v1"; + +pub(super) fn encode_relationship_grant_secret( + secret: &GrantSecret, +) -> Result { + // Custody stores only the 32-byte secret; issuer/holder/generation/protocol + // bindings live in the relationship row and are enforced at prove/verify time. + SecretMaterial::new(secret.as_bytes().to_vec()) +} + +pub(super) fn secret_from_material(material: &SecretMaterial) -> Result { + let bytes: [u8; 32] = + material + .to_vec() + .try_into() + .map_err(|_| VnidropError::SecureStorageCorrupted { + reason: "relationship grant secret has invalid length".to_string(), + })?; + Ok(GrantSecret::from_bytes(bytes)) +} + +pub(super) fn prove_relationship_grant( + grant_id: GrantId, + secret: &GrantSecret, + challenge: &Challenge, + issuer: &str, + holder: &str, + generation: u64, + protocol_version: u16, +) -> GrantProof { + let mac = relationship_mac( + secret, + challenge, + issuer, + holder, + generation, + protocol_version, + ); + GrantProof::from_parts(grant_id, mac) +} + +pub(super) fn verify_relationship_grant( + secret: &GrantSecret, + proof: &GrantProof, + challenge: &Challenge, + issuer: &str, + holder: &str, + generation: u64, + protocol_version: u16, +) -> Result<(), &'static str> { + let expected = relationship_mac( + secret, + challenge, + issuer, + holder, + generation, + protocol_version, + ); + if blake3::Hash::from_bytes(expected) != blake3::Hash::from_bytes(*proof.mac()) { + return Err("bad relationship grant proof"); + } + Ok(()) +} + +fn relationship_mac( + secret: &GrantSecret, + challenge: &Challenge, + issuer: &str, + holder: &str, + generation: u64, + protocol_version: u16, +) -> [u8; 32] { + let mut hasher = blake3::Hasher::new_keyed(secret.as_bytes()); + hasher.update(RELATIONSHIP_GRANT_CONTEXT); + hasher.update(challenge.as_bytes()); + hasher.update(&(issuer.len() as u64).to_le_bytes()); + hasher.update(issuer.as_bytes()); + hasher.update(&(holder.len() as u64).to_le_bytes()); + hasher.update(holder.as_bytes()); + hasher.update(&generation.to_le_bytes()); + hasher.update(&protocol_version.to_le_bytes()); + *hasher.finalize().as_bytes() +} + +#[cfg(test)] +mod grant_vectors { + use super::*; + use crate::api::experimental_saved_device_capabilities; + use data_encoding::HEXLOWER; + + #[test] + fn relationship_grant_proof_vectors_are_stable() { + let secret = + GrantSecret::decode("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef") + .unwrap(); + let grant_id = GrantId::decode("0123456789abcdef0123456789abcdef").unwrap(); + let challenge = Challenge::from_bytes([9u8; 32]); + let protocol = experimental_saved_device_capabilities().relationship_protocol_version; + let proof = prove_relationship_grant( + grant_id, &secret, &challenge, "issuer", "holder", 1, protocol, + ); + // Binding and replay resistance: wrong holder or challenge must fail. + verify_relationship_grant(&secret, &proof, &challenge, "issuer", "holder", 1, protocol) + .unwrap(); + let mac_hex = HEXLOWER.encode(proof.mac()); + assert_eq!( + mac_hex, + "e6cc2641183b84fae9e3805761961d69e09d25a1f8ceeaeede952774ddd95d6b" + ); + assert!(verify_relationship_grant( + &secret, &proof, &challenge, "issuer", "other", 1, protocol, + ) + .is_err()); + let other_challenge = Challenge::from_bytes([8u8; 32]); + assert!(verify_relationship_grant( + &secret, + &proof, + &other_challenge, + "issuer", + "holder", + 1, + protocol, + ) + .is_err()); + } +} diff --git a/crates/vnidrop/src/device_relationship/mod.rs b/crates/vnidrop/src/device_relationship/mod.rs new file mode 100644 index 0000000..346d282 --- /dev/null +++ b/crates/vnidrop/src/device_relationship/mod.rs @@ -0,0 +1,1306 @@ +//! Experimental saved-device mutual-consent relationships. +//! +//! Implements design §6/§7: pending outgoing/incoming states, directional grants +//! bound to relationship generation, and Saved only after mutual acknowledgement. + +use std::{collections::HashMap, fmt, sync::Arc, time::Duration}; + +use anyhow::Context; +use data_encoding::HEXLOWER; +use iroh::{ + endpoint::Connection, + protocol::{AcceptError, ProtocolHandler}, + Endpoint, EndpointAddr, EndpointId, +}; +use irpc::{channel::oneshot, rpc_requests, Client, WithChannels}; +use irpc_iroh::{read_request, IrohLazyRemoteConnection}; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use sqlx::{Row, SqlitePool}; +use tokio::sync::Mutex as TokioMutex; + +use crate::{ + api::{DeviceRelationship, DeviceRelationshipState, SavedDevice}, + error::VnidropError, + event_hub::EventHub, + grant::{Challenge, GrantId, GrantProof, GrantSecret}, + pairing_eligibility::PairingEligibilityService, + secure_secret::{SecretCustody, SecretHandle, SecretKind, SecretMaterial}, + ticket::encode_persisted_sender_address, + util::now_ms, +}; + +mod crypto; + +use crypto::{ + encode_relationship_grant_secret, prove_relationship_grant, secret_from_material, + verify_relationship_grant, +}; + +const PENDING_TTL_MS: i64 = 30 * 60 * 1_000; +const PAIRING_RPC_TIMEOUT: Duration = Duration::from_secs(15); + +#[derive(Clone)] +pub(crate) struct DeviceRelationshipService { + pool: SqlitePool, + custody: Option>, + eligibility: PairingEligibilityService, + event_hub: Arc, + local_endpoint_id: String, + endpoint: Endpoint, + peer_locks: Arc>>>>, +} + +impl DeviceRelationshipService { + pub(crate) fn new( + pool: SqlitePool, + custody: Option>, + eligibility: PairingEligibilityService, + event_hub: Arc, + local_endpoint_id: String, + endpoint: Endpoint, + ) -> Self { + Self { + pool, + custody, + eligibility, + event_hub, + local_endpoint_id, + endpoint, + peer_locks: Arc::new(TokioMutex::new(HashMap::new())), + } + } + + async fn lock_peer(&self, peer_endpoint_id: &str) -> Arc> { + let mut locks = self.peer_locks.lock().await; + locks + .entry(peer_endpoint_id.to_string()) + .or_insert_with(|| Arc::new(TokioMutex::new(()))) + .clone() + } + + pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> { + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS device_relationships ( + remote_endpoint_id TEXT PRIMARY KEY, + state TEXT NOT NULL, + generation INTEGER NOT NULL, + minimum_protocol_version INTEGER NOT NULL, + session_id TEXT, + issued_grant_handle TEXT, + held_grant_handle TEXT, + issued_grant_id TEXT, + held_grant_id TEXT, + peer_ack INTEGER NOT NULL DEFAULT 0, + local_ack INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + "#, + ) + .execute(pool) + .await?; + let columns = sqlx::query("PRAGMA table_info(device_relationships)") + .fetch_all(pool) + .await?; + let has = |name: &str| columns.iter().any(|row| row.get::(1) == name); + if !has("issued_grant_id") { + sqlx::query("ALTER TABLE device_relationships ADD COLUMN issued_grant_id TEXT") + .execute(pool) + .await?; + } + if !has("held_grant_id") { + sqlx::query("ALTER TABLE device_relationships ADD COLUMN held_grant_id TEXT") + .execute(pool) + .await?; + } + Ok(()) + } + + /// Drop orphaned relationship grant secrets and disable rows whose secrets are gone. + pub(crate) async fn reconcile(&self) -> Result<(), VnidropError> { + let Some(custody) = &self.custody else { + return Ok(()); + }; + let rows = sqlx::query( + r#" + SELECT remote_endpoint_id, issued_grant_handle, held_grant_handle, state + FROM device_relationships + "#, + ) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + + let mut live_handles = std::collections::HashSet::new(); + for row in rows { + let peer: String = row.get("remote_endpoint_id"); + let state = parse_state(&row.get::("state"))?; + let issued: Option = row.get("issued_grant_handle"); + let held: Option = row.get("held_grant_handle"); + let mut missing = false; + for handle in [&issued, &held].into_iter().flatten() { + live_handles.insert(handle.clone()); + if custody + .load(&SecretHandle::from_stored(handle.clone())) + .await + .is_err() + { + missing = true; + } + } + if missing && state == DeviceRelationshipState::Saved { + // Grants are required for a usable saved device. + self.delete_relationship(&peer).await?; + } + } + + for handle in custody + .list_active_handles(SecretKind::RelationshipGrant) + .await? + { + if !live_handles.contains(handle.as_str()) { + let _ = custody.remove(&handle).await; + } + } + Ok(()) + } + + pub(crate) async fn list(&self) -> Result, VnidropError> { + self.expire_pending().await?; + let rows = sqlx::query( + r#" + SELECT remote_endpoint_id, state, generation, minimum_protocol_version, created_at, updated_at + FROM device_relationships + ORDER BY updated_at DESC + "#, + ) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + rows.into_iter().map(row_to_relationship).collect() + } + + pub(crate) async fn list_saved_devices(&self) -> Result, VnidropError> { + Ok(self + .list() + .await? + .into_iter() + .filter(|entry| entry.state == DeviceRelationshipState::Saved) + .map(|entry| SavedDevice { + endpoint_id: entry.remote_endpoint_id, + local_label: None, + remote_display_name: None, + created_at: entry.created_at, + last_authenticated_at: Some(entry.updated_at), + }) + .collect()) + } + + pub(crate) async fn request_pairing( + self: &Arc, + peer_endpoint_id: String, + ) -> Result { + let peer_lock = self.lock_peer(&peer_endpoint_id).await; + let _guard = peer_lock.lock().await; + + if let Some(existing) = self.find_row(&peer_endpoint_id).await? { + if existing.state == DeviceRelationshipState::Saved { + return Ok(true); + } + } + + let Some(taken) = self.eligibility.take_eligibility(&peer_endpoint_id).await? else { + return Ok(false); + }; + let generation = 1_u64; + let now = now_ms(); + + // Peer already prompted us for the same qualifying session: merge into + // one relationship without a second consent prompt. + if let Some(existing) = self.find_row(&peer_endpoint_id).await? { + if existing.state == DeviceRelationshipState::PendingIncoming + && existing.session_id.as_deref() == Some(taken.session_id.as_str()) + { + self.upsert_relationship(RelationshipUpsert { + remote_endpoint_id: &peer_endpoint_id, + state: DeviceRelationshipState::PendingOutgoing, + generation, + minimum_protocol_version: taken.protocol_version, + session_id: Some(&taken.session_id), + issued_grant_handle: None, + held_grant_handle: None, + issued_grant_id: None, + held_grant_id: None, + peer_ack: false, + local_ack: false, + created_at: existing.created_at, + updated_at: now, + }) + .await?; + self.emit_changed(&peer_endpoint_id, DeviceRelationshipState::PendingOutgoing); + drop(_guard); + self.complete_simultaneous_merge( + &peer_endpoint_id, + generation, + taken.protocol_version, + ) + .await?; + return Ok(true); + } + } + + self.upsert_relationship(RelationshipUpsert { + remote_endpoint_id: &peer_endpoint_id, + state: DeviceRelationshipState::PendingOutgoing, + generation, + minimum_protocol_version: taken.protocol_version, + session_id: Some(&taken.session_id), + issued_grant_handle: None, + held_grant_handle: None, + issued_grant_id: None, + held_grant_id: None, + peer_ack: false, + local_ack: false, + created_at: now, + updated_at: now, + }) + .await?; + self.emit_changed(&peer_endpoint_id, DeviceRelationshipState::PendingOutgoing); + drop(_guard); + + let addr = self.peer_addr(&peer_endpoint_id).await?; + let client = RelationshipClient::connect(self.endpoint.clone(), addr); + let response = match tokio::time::timeout( + PAIRING_RPC_TIMEOUT, + client.pairing_request(PairingRequest { + session_id: taken.session_id.clone(), + capability: taken.capability.to_vec(), + protocol_version: taken.protocol_version, + generation, + }), + ) + .await + { + Ok(Ok(response)) => response, + // Timeout/network failure keeps bounded PendingOutgoing for recovery. + Ok(Err(_)) | Err(_) => return Ok(true), + }; + + match response { + PairingRequestResponse::AwaitingConsent => Ok(true), + PairingRequestResponse::Merged => { + self.complete_simultaneous_merge( + &peer_endpoint_id, + generation, + taken.protocol_version, + ) + .await?; + Ok(true) + } + PairingRequestResponse::AlreadySaved => { + self.activate_saved(&peer_endpoint_id).await?; + Ok(true) + } + PairingRequestResponse::Rejected => { + // Keep pending: a transient reject must not erase recoverable state. + // Explicit decline is handled on the peer via consent=false. + Ok(true) + } + } + } + + pub(crate) async fn respond_to_pairing( + self: &Arc, + peer_endpoint_id: String, + accepted: bool, + ) -> Result { + let Some(row) = self.find_row(&peer_endpoint_id).await? else { + return Ok(false); + }; + if row.state == DeviceRelationshipState::Saved { + return Ok(true); + } + if row.state != DeviceRelationshipState::PendingIncoming { + return Ok(false); + } + if !accepted { + if let Some(session_id) = &row.session_id { + let _ = self + .eligibility + .consume_session(&peer_endpoint_id, session_id) + .await; + } + self.delete_relationship(&peer_endpoint_id).await?; + let addr = self.peer_addr(&peer_endpoint_id).await?; + let client = RelationshipClient::connect(self.endpoint.clone(), addr); + let _ = tokio::time::timeout( + PAIRING_RPC_TIMEOUT, + client.pairing_consent(PairingConsent { + accepted: false, + grant: None, + challenge: None, + generation: row.generation, + protocol_version: row.minimum_protocol_version, + }), + ) + .await; + return Ok(true); + } + + if let Some(session_id) = &row.session_id { + let _ = self + .eligibility + .consume_session(&peer_endpoint_id, session_id) + .await; + } + + let grant = self + .mint_and_store_issued_grant( + &peer_endpoint_id, + row.generation, + row.minimum_protocol_version, + ) + .await?; + let challenge = Challenge::generate(); + let addr = self.peer_addr(&peer_endpoint_id).await?; + let client = RelationshipClient::connect(self.endpoint.clone(), addr); + let response = match tokio::time::timeout( + PAIRING_RPC_TIMEOUT, + client.pairing_consent(PairingConsent { + accepted: true, + grant: Some(grant.clone()), + challenge: Some(challenge.encode()), + generation: row.generation, + protocol_version: row.minimum_protocol_version, + }), + ) + .await + { + Ok(Ok(response)) => response, + // Leave PendingIncoming so the operator can retry consent. + Ok(Err(_)) | Err(_) => return Ok(true), + }; + + match response { + PairingConsentResponse::Completed { + grant: peer_grant, + possession_proof, + ack_challenge, + } => { + self.verify_issued_possession( + &peer_endpoint_id, + &challenge, + &possession_proof, + row.generation, + row.minimum_protocol_version, + ) + .await?; + self.store_held_grant(&peer_endpoint_id, &peer_grant) + .await?; + let ack_challenge = + Challenge::decode(&ack_challenge).map_err(VnidropError::invalid_input)?; + let proof = self + .prove_held_possession( + &peer_endpoint_id, + &ack_challenge, + peer_grant.generation, + peer_grant.protocol_version, + ) + .await?; + match tokio::time::timeout( + PAIRING_RPC_TIMEOUT, + client.pairing_ack(PairingAck { + possession_proof: proof, + challenge: ack_challenge.encode(), + generation: row.generation, + protocol_version: row.minimum_protocol_version, + }), + ) + .await + { + Ok(Ok(PairingAckResponse::Acknowledged | PairingAckResponse::AlreadySaved)) => { + self.set_acks(&peer_endpoint_id, true, true).await?; + self.activate_saved(&peer_endpoint_id).await?; + Ok(true) + } + // Ack lost/failed: grants are stored; pending remains recoverable. + _ => Ok(true), + } + } + PairingConsentResponse::AlreadySaved => { + self.activate_saved(&peer_endpoint_id).await?; + Ok(true) + } + PairingConsentResponse::Rejected => Ok(false), + } + } + + async fn handle_pairing_request( + self: &Arc, + remote_endpoint_id: String, + request: PairingRequest, + ) -> PairingRequestResponse { + let peer_lock = self.lock_peer(&remote_endpoint_id).await; + let _guard = peer_lock.lock().await; + + if let Ok(Some(existing)) = self.find_row(&remote_endpoint_id).await { + if existing.state == DeviceRelationshipState::Saved { + return PairingRequestResponse::AlreadySaved; + } + // Simultaneous initiation: both sides proved eligibility for the same session. + if existing.state == DeviceRelationshipState::PendingOutgoing + && existing.session_id.as_deref() == Some(request.session_id.as_str()) + { + let generation = existing.generation; + let protocol_version = existing.minimum_protocol_version; + let should_lead = self.local_endpoint_id.as_str() < remote_endpoint_id.as_str(); + drop(_guard); + if should_lead { + let _ = self + .complete_simultaneous_merge( + &remote_endpoint_id, + generation, + protocol_version, + ) + .await; + } + return PairingRequestResponse::Merged; + } + } + + let capability = match SecretMaterial::new(request.capability) { + Ok(capability) => capability, + Err(_) => return PairingRequestResponse::Rejected, + }; + let accepted = match self + .eligibility + .validate_presented_capability(&remote_endpoint_id, &request.session_id, &capability) + .await + { + Ok(Some(_)) => true, + Ok(None) | Err(_) => false, + }; + if !accepted { + return PairingRequestResponse::Rejected; + } + + let now = now_ms(); + if self + .upsert_relationship(RelationshipUpsert { + remote_endpoint_id: &remote_endpoint_id, + state: DeviceRelationshipState::PendingIncoming, + generation: request.generation, + minimum_protocol_version: request.protocol_version, + session_id: Some(&request.session_id), + issued_grant_handle: None, + held_grant_handle: None, + issued_grant_id: None, + held_grant_id: None, + peer_ack: false, + local_ack: false, + created_at: now, + updated_at: now, + }) + .await + .is_err() + { + return PairingRequestResponse::Rejected; + } + self.emit_changed( + &remote_endpoint_id, + DeviceRelationshipState::PendingIncoming, + ); + PairingRequestResponse::AwaitingConsent + } + + async fn handle_pairing_consent( + self: &Arc, + remote_endpoint_id: String, + consent: PairingConsent, + ) -> PairingConsentResponse { + let Ok(Some(row)) = self.find_row(&remote_endpoint_id).await else { + return PairingConsentResponse::Rejected; + }; + if row.state == DeviceRelationshipState::Saved { + return PairingConsentResponse::AlreadySaved; + } + if row.state != DeviceRelationshipState::PendingOutgoing { + return PairingConsentResponse::Rejected; + } + if !consent.accepted { + // Peer declined: clear local pending; eligibility was already consumed by requester. + let _ = self.delete_relationship(&remote_endpoint_id).await; + return PairingConsentResponse::Rejected; + } + let (Some(peer_grant), Some(challenge_hex)) = (consent.grant, consent.challenge) else { + return PairingConsentResponse::Rejected; + }; + let Ok(challenge) = Challenge::decode(&challenge_hex) else { + return PairingConsentResponse::Rejected; + }; + if self + .store_held_grant(&remote_endpoint_id, &peer_grant) + .await + .is_err() + { + return PairingConsentResponse::Rejected; + } + let Ok(local_grant) = self + .mint_and_store_issued_grant( + &remote_endpoint_id, + consent.generation, + consent.protocol_version, + ) + .await + else { + return PairingConsentResponse::Rejected; + }; + let Ok(possession_proof) = self + .prove_held_possession( + &remote_endpoint_id, + &challenge, + peer_grant.generation, + peer_grant.protocol_version, + ) + .await + else { + return PairingConsentResponse::Rejected; + }; + let ack_challenge = Challenge::generate(); + if self + .set_acks(&remote_endpoint_id, false, false) + .await + .is_err() + { + return PairingConsentResponse::Rejected; + } + PairingConsentResponse::Completed { + grant: Box::new(local_grant), + possession_proof, + ack_challenge: ack_challenge.encode(), + } + } + + async fn handle_pairing_ack( + &self, + remote_endpoint_id: String, + ack: PairingAck, + ) -> PairingAckResponse { + let Ok(Some(row)) = self.find_row(&remote_endpoint_id).await else { + return PairingAckResponse::Rejected; + }; + if row.state == DeviceRelationshipState::Saved { + return PairingAckResponse::AlreadySaved; + } + if row.state != DeviceRelationshipState::PendingOutgoing + && row.state != DeviceRelationshipState::PendingIncoming + { + return PairingAckResponse::Rejected; + } + // Peer proves possession of our issued grant over the ack challenge we + // sent in Completed; the challenge travels again on the ack message. + let Ok(challenge) = Challenge::decode(&ack.challenge) else { + return PairingAckResponse::Rejected; + }; + if self + .verify_issued_possession( + &remote_endpoint_id, + &challenge, + &ack.possession_proof, + ack.generation, + ack.protocol_version, + ) + .await + .is_err() + { + return PairingAckResponse::Rejected; + } + if self + .set_acks(&remote_endpoint_id, true, true) + .await + .is_err() + { + return PairingAckResponse::Rejected; + } + if self.activate_saved(&remote_endpoint_id).await.is_err() { + return PairingAckResponse::Rejected; + } + PairingAckResponse::Acknowledged + } + + async fn complete_simultaneous_merge( + self: &Arc, + peer_endpoint_id: &str, + generation: u64, + protocol_version: u16, + ) -> Result<(), VnidropError> { + let peer_lock = self.lock_peer(peer_endpoint_id).await; + let _guard = peer_lock.lock().await; + if let Some(row) = self.find_row(peer_endpoint_id).await? { + if row.state == DeviceRelationshipState::Saved { + return Ok(()); + } + // Another merge attempt already minted; do not rotate the issued grant. + if row.issued_grant_handle.is_some() { + return Ok(()); + } + } + // Deterministic role: smaller endpoint ID leads the consent+ack exchange. + if self.local_endpoint_id.as_str() >= peer_endpoint_id { + return Ok(()); + } + let grant = self + .mint_and_store_issued_grant(peer_endpoint_id, generation, protocol_version) + .await?; + let challenge = Challenge::generate(); + drop(_guard); + let addr = self.peer_addr(peer_endpoint_id).await?; + let client = RelationshipClient::connect(self.endpoint.clone(), addr); + let response = match tokio::time::timeout( + PAIRING_RPC_TIMEOUT, + client.pairing_consent(PairingConsent { + accepted: true, + grant: Some(grant), + challenge: Some(challenge.encode()), + generation, + protocol_version, + }), + ) + .await + { + Ok(Ok(response)) => response, + Ok(Err(_)) | Err(_) => return Ok(()), + }; + match response { + PairingConsentResponse::Completed { + grant: peer_grant, + possession_proof, + ack_challenge, + } => { + self.verify_issued_possession( + peer_endpoint_id, + &challenge, + &possession_proof, + generation, + protocol_version, + ) + .await?; + self.store_held_grant(peer_endpoint_id, &peer_grant).await?; + let ack_challenge = + Challenge::decode(&ack_challenge).map_err(VnidropError::invalid_input)?; + let proof = self + .prove_held_possession( + peer_endpoint_id, + &ack_challenge, + peer_grant.generation, + peer_grant.protocol_version, + ) + .await?; + if let Ok(Ok(PairingAckResponse::Acknowledged | PairingAckResponse::AlreadySaved)) = + tokio::time::timeout( + PAIRING_RPC_TIMEOUT, + client.pairing_ack(PairingAck { + possession_proof: proof, + challenge: ack_challenge.encode(), + generation, + protocol_version, + }), + ) + .await + { + self.set_acks(peer_endpoint_id, true, true).await?; + self.activate_saved(peer_endpoint_id).await?; + } + } + PairingConsentResponse::AlreadySaved => { + self.activate_saved(peer_endpoint_id).await?; + } + PairingConsentResponse::Rejected => {} + } + Ok(()) + } + + async fn mint_and_store_issued_grant( + &self, + peer_endpoint_id: &str, + generation: u64, + protocol_version: u16, + ) -> Result { + let custody = + self.custody + .as_ref() + .ok_or_else(|| VnidropError::SecureStorageUnavailable { + reason: "relationship grants require protected custody".to_string(), + })?; + let secret = GrantSecret::generate(); + let grant_id = GrantId::generate(); + let material = encode_relationship_grant_secret(&secret)?; + let handle = custody + .protect(SecretKind::RelationshipGrant, material, None) + .await?; + sqlx::query( + r#" + UPDATE device_relationships + SET issued_grant_handle = ?2, issued_grant_id = ?3, updated_at = ?4 + WHERE remote_endpoint_id = ?1 + "#, + ) + .bind(peer_endpoint_id) + .bind(handle.as_str()) + .bind(grant_id.encode()) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(WireGrant { + grant_id: grant_id.encode(), + secret: secret.encode(), + issuer_endpoint_id: self.local_endpoint_id.clone(), + holder_endpoint_id: peer_endpoint_id.to_string(), + generation, + protocol_version, + }) + } + + async fn store_held_grant( + &self, + peer_endpoint_id: &str, + grant: &WireGrant, + ) -> Result<(), VnidropError> { + if grant.holder_endpoint_id != self.local_endpoint_id + || grant.issuer_endpoint_id != peer_endpoint_id + { + return Err(VnidropError::invalid_input(anyhow::anyhow!( + "relationship grant endpoint binding mismatch" + ))); + } + let custody = + self.custody + .as_ref() + .ok_or_else(|| VnidropError::SecureStorageUnavailable { + reason: "relationship grants require protected custody".to_string(), + })?; + let grant_id = GrantId::decode(&grant.grant_id).map_err(VnidropError::invalid_input)?; + let secret = GrantSecret::decode(&grant.secret).map_err(VnidropError::invalid_input)?; + let material = encode_relationship_grant_secret(&secret)?; + let handle = custody + .protect(SecretKind::RelationshipGrant, material, None) + .await?; + sqlx::query( + r#" + UPDATE device_relationships + SET held_grant_handle = ?2, held_grant_id = ?3, updated_at = ?4 + WHERE remote_endpoint_id = ?1 + "#, + ) + .bind(peer_endpoint_id) + .bind(handle.as_str()) + .bind(grant_id.encode()) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + async fn prove_held_possession( + &self, + peer_endpoint_id: &str, + challenge: &Challenge, + generation: u64, + protocol_version: u16, + ) -> Result { + let custody = + self.custody + .as_ref() + .ok_or_else(|| VnidropError::SecureStorageUnavailable { + reason: "relationship grants require protected custody".to_string(), + })?; + let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("missing relationship for proof")) + })?; + let handle = row.held_grant_handle.ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("missing held grant for proof")) + })?; + let grant_id = GrantId::decode(row.held_grant_id.as_deref().ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("missing held grant id")) + })?) + .map_err(VnidropError::invalid_input)?; + let material = custody.load(&SecretHandle::from_stored(handle)).await?; + let secret = secret_from_material(&material)?; + let proof = prove_relationship_grant( + grant_id, + &secret, + challenge, + peer_endpoint_id, + &self.local_endpoint_id, + generation, + protocol_version, + ); + Ok(WireProof { + grant_id: proof.grant_id.encode(), + mac: HEXLOWER.encode(proof.mac()), + challenge: challenge.encode(), + }) + } + + async fn verify_issued_possession( + &self, + peer_endpoint_id: &str, + challenge: &Challenge, + proof: &WireProof, + generation: u64, + protocol_version: u16, + ) -> Result<(), VnidropError> { + let custody = + self.custody + .as_ref() + .ok_or_else(|| VnidropError::SecureStorageUnavailable { + reason: "relationship grants require protected custody".to_string(), + })?; + let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("missing relationship for verify")) + })?; + let handle = row.issued_grant_handle.ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("missing issued grant for verify")) + })?; + let expected_id = row.issued_grant_id.ok_or_else(|| { + VnidropError::invalid_input(anyhow::anyhow!("missing issued grant id")) + })?; + if proof.grant_id != expected_id { + return Err(VnidropError::invalid_input(anyhow::anyhow!( + "relationship grant id mismatch" + ))); + } + let material = custody.load(&SecretHandle::from_stored(handle)).await?; + let secret = secret_from_material(&material)?; + let grant_id = GrantId::decode(&proof.grant_id).map_err(VnidropError::invalid_input)?; + let mac_bytes = HEXLOWER + .decode(proof.mac.as_bytes()) + .context("invalid proof mac") + .map_err(VnidropError::invalid_input)?; + let mac: [u8; 32] = mac_bytes.try_into().map_err(|_| { + VnidropError::invalid_input(anyhow::anyhow!("invalid proof mac length")) + })?; + let presented = GrantProof::from_parts(grant_id, mac); + verify_relationship_grant( + &secret, + &presented, + challenge, + &self.local_endpoint_id, + peer_endpoint_id, + generation, + protocol_version, + ) + .map_err(|error| VnidropError::invalid_input(anyhow::anyhow!(error)))?; + Ok(()) + } + + async fn set_acks( + &self, + peer_endpoint_id: &str, + local_ack: bool, + peer_ack: bool, + ) -> Result<(), VnidropError> { + sqlx::query( + "UPDATE device_relationships SET local_ack = ?2, peer_ack = ?3, updated_at = ?4 WHERE remote_endpoint_id = ?1", + ) + .bind(peer_endpoint_id) + .bind(i64::from(local_ack)) + .bind(i64::from(peer_ack)) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + async fn activate_saved(&self, peer_endpoint_id: &str) -> Result<(), VnidropError> { + sqlx::query( + "UPDATE device_relationships SET state = ?2, updated_at = ?3 WHERE remote_endpoint_id = ?1", + ) + .bind(peer_endpoint_id) + .bind(state_as_str(DeviceRelationshipState::Saved)) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + self.emit_changed(peer_endpoint_id, DeviceRelationshipState::Saved); + Ok(()) + } + + async fn expire_pending(&self) -> Result<(), VnidropError> { + let cutoff = now_ms() - PENDING_TTL_MS; + let rows = sqlx::query( + r#" + SELECT remote_endpoint_id FROM device_relationships + WHERE state IN ('pending_outgoing', 'pending_incoming') AND updated_at < ?1 + "#, + ) + .bind(cutoff) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + for row in rows { + let peer: String = row.get(0); + self.delete_relationship(&peer).await?; + } + Ok(()) + } + + async fn delete_relationship(&self, peer_endpoint_id: &str) -> Result<(), VnidropError> { + if let Ok(Some(row)) = self.find_row(peer_endpoint_id).await { + if let Some(custody) = &self.custody { + for handle in [row.issued_grant_handle, row.held_grant_handle] + .into_iter() + .flatten() + { + let _ = custody.remove(&SecretHandle::from_stored(handle)).await; + } + } + } + sqlx::query("DELETE FROM device_relationships WHERE remote_endpoint_id = ?1") + .bind(peer_endpoint_id) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + async fn find_row( + &self, + peer_endpoint_id: &str, + ) -> Result, VnidropError> { + let row = sqlx::query( + r#" + SELECT remote_endpoint_id, state, generation, minimum_protocol_version, session_id, + issued_grant_handle, held_grant_handle, issued_grant_id, held_grant_id, + peer_ack, local_ack, created_at, updated_at + FROM device_relationships WHERE remote_endpoint_id = ?1 + "#, + ) + .bind(peer_endpoint_id) + .fetch_optional(&self.pool) + .await + .map_err(VnidropError::repository)?; + row.map(relationship_row_from_sql).transpose() + } + + async fn upsert_relationship(&self, entry: RelationshipUpsert<'_>) -> Result<(), VnidropError> { + sqlx::query( + r#" + INSERT INTO device_relationships ( + remote_endpoint_id, state, generation, minimum_protocol_version, session_id, + issued_grant_handle, held_grant_handle, issued_grant_id, held_grant_id, + peer_ack, local_ack, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13) + ON CONFLICT(remote_endpoint_id) DO UPDATE SET + state = excluded.state, + generation = excluded.generation, + minimum_protocol_version = excluded.minimum_protocol_version, + session_id = excluded.session_id, + issued_grant_handle = COALESCE(excluded.issued_grant_handle, device_relationships.issued_grant_handle), + held_grant_handle = COALESCE(excluded.held_grant_handle, device_relationships.held_grant_handle), + issued_grant_id = COALESCE(excluded.issued_grant_id, device_relationships.issued_grant_id), + held_grant_id = COALESCE(excluded.held_grant_id, device_relationships.held_grant_id), + peer_ack = excluded.peer_ack, + local_ack = excluded.local_ack, + updated_at = excluded.updated_at + "#, + ) + .bind(entry.remote_endpoint_id) + .bind(state_as_str(entry.state)) + .bind(entry.generation as i64) + .bind(i64::from(entry.minimum_protocol_version)) + .bind(entry.session_id) + .bind(entry.issued_grant_handle) + .bind(entry.held_grant_handle) + .bind(entry.issued_grant_id) + .bind(entry.held_grant_id) + .bind(i64::from(entry.peer_ack)) + .bind(i64::from(entry.local_ack)) + .bind(entry.created_at) + .bind(entry.updated_at) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + async fn peer_addr(&self, peer_endpoint_id: &str) -> Result { + let parsed: EndpointId = peer_endpoint_id + .parse() + .context("unusable peer endpoint id") + .map_err(VnidropError::invalid_input)?; + if let Some(info) = self.endpoint.remote_info(parsed).await { + let mut addr = EndpointAddr::from(parsed); + addr.addrs = info.addrs().map(|entry| entry.addr().clone()).collect(); + let _ = encode_persisted_sender_address(&addr); + return Ok(addr); + } + Ok(EndpointAddr::from(parsed)) + } + + fn emit_changed(&self, peer_endpoint_id: &str, state: DeviceRelationshipState) { + self.event_hub.emit_endpoint( + "pairing", + "relationship-changed", + json!({ + "peer_endpoint_id": peer_endpoint_id, + "state": state_as_str(state), + }), + ); + } +} + +#[derive(Clone)] +pub(crate) struct RelationshipProtocol { + relationships: Arc, +} + +impl RelationshipProtocol { + pub(crate) const ALPN: &'static [u8] = b"/vnidrop/relationship/1"; + + pub(crate) fn new(relationships: Arc) -> Self { + Self { relationships } + } +} + +impl fmt::Debug for RelationshipProtocol { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("RelationshipProtocol") + } +} + +impl ProtocolHandler for RelationshipProtocol { + async fn accept(&self, connection: Connection) -> Result<(), AcceptError> { + let remote_endpoint_id = connection.remote_id().to_string(); + while let Some(message) = read_request::(&connection).await? { + match message { + RelationshipMessage::PairingRequest(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self + .relationships + .handle_pairing_request(remote_endpoint_id.clone(), inner) + .await; + let _ = tx.send(response).await; + } + RelationshipMessage::PairingConsent(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self + .relationships + .handle_pairing_consent(remote_endpoint_id.clone(), inner) + .await; + let _ = tx.send(response).await; + } + RelationshipMessage::PairingAck(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self + .relationships + .handle_pairing_ack(remote_endpoint_id.clone(), inner) + .await; + let _ = tx.send(response).await; + } + } + } + connection.closed().await; + Ok(()) + } +} + +struct RelationshipClient { + inner: Client, +} + +impl RelationshipClient { + fn connect(endpoint: Endpoint, addr: EndpointAddr) -> Self { + Self { + inner: Client::boxed(IrohLazyRemoteConnection::new( + endpoint, + addr, + RelationshipProtocol::ALPN.to_vec(), + )), + } + } + + async fn pairing_request( + &self, + request: PairingRequest, + ) -> Result { + self.inner.rpc(request).await + } + + async fn pairing_consent( + &self, + consent: PairingConsent, + ) -> Result { + self.inner.rpc(consent).await + } + + async fn pairing_ack(&self, ack: PairingAck) -> Result { + self.inner.rpc(ack).await + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct PairingRequest { + session_id: String, + capability: Vec, + protocol_version: u16, + generation: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +enum PairingRequestResponse { + AwaitingConsent, + Merged, + AlreadySaved, + Rejected, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct PairingConsent { + accepted: bool, + grant: Option, + challenge: Option, + generation: u64, + protocol_version: u16, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +enum PairingConsentResponse { + Completed { + grant: Box, + possession_proof: WireProof, + ack_challenge: String, + }, + AlreadySaved, + Rejected, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct PairingAck { + possession_proof: WireProof, + challenge: String, + generation: u64, + protocol_version: u16, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +enum PairingAckResponse { + Acknowledged, + AlreadySaved, + Rejected, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct WireGrant { + grant_id: String, + secret: String, + issuer_endpoint_id: String, + holder_endpoint_id: String, + generation: u64, + protocol_version: u16, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct WireProof { + grant_id: String, + mac: String, + challenge: String, +} + +#[rpc_requests(message = RelationshipMessage)] +#[derive(Debug, Serialize, Deserialize)] +#[allow( + clippy::enum_variant_names, + reason = "Pairing* names mirror the wire RPC surface" +)] +enum RelationshipMessages { + #[rpc(tx = oneshot::Sender)] + PairingRequest(PairingRequest), + #[rpc(tx = oneshot::Sender)] + PairingConsent(PairingConsent), + #[rpc(tx = oneshot::Sender)] + PairingAck(PairingAck), +} + +struct RelationshipUpsert<'a> { + remote_endpoint_id: &'a str, + state: DeviceRelationshipState, + generation: u64, + minimum_protocol_version: u16, + session_id: Option<&'a str>, + issued_grant_handle: Option<&'a str>, + held_grant_handle: Option<&'a str>, + issued_grant_id: Option<&'a str>, + held_grant_id: Option<&'a str>, + peer_ack: bool, + local_ack: bool, + created_at: i64, + updated_at: i64, +} + +struct RelationshipRow { + state: DeviceRelationshipState, + generation: u64, + minimum_protocol_version: u16, + session_id: Option, + issued_grant_handle: Option, + held_grant_handle: Option, + issued_grant_id: Option, + held_grant_id: Option, + created_at: i64, +} + +fn state_as_str(state: DeviceRelationshipState) -> &'static str { + match state { + DeviceRelationshipState::PendingOutgoing => "pending_outgoing", + DeviceRelationshipState::PendingIncoming => "pending_incoming", + DeviceRelationshipState::Saved => "saved", + DeviceRelationshipState::Revoked => "revoked", + DeviceRelationshipState::Blocked => "blocked", + } +} + +fn parse_state(value: &str) -> Result { + match value { + "pending_outgoing" => Ok(DeviceRelationshipState::PendingOutgoing), + "pending_incoming" => Ok(DeviceRelationshipState::PendingIncoming), + "saved" => Ok(DeviceRelationshipState::Saved), + "revoked" => Ok(DeviceRelationshipState::Revoked), + "blocked" => Ok(DeviceRelationshipState::Blocked), + _ => Err(VnidropError::Internal { + reason: "unknown device relationship state".to_string(), + }), + } +} + +fn row_to_relationship(row: sqlx::sqlite::SqliteRow) -> Result { + Ok(DeviceRelationship { + remote_endpoint_id: row.get("remote_endpoint_id"), + state: parse_state(&row.get::("state"))?, + generation: row.get::("generation") as u64, + minimum_protocol_version: row.get::("minimum_protocol_version") as u16, + created_at: row.get("created_at"), + updated_at: row.get("updated_at"), + }) +} + +fn relationship_row_from_sql( + row: sqlx::sqlite::SqliteRow, +) -> Result { + Ok(RelationshipRow { + state: parse_state(&row.get::("state"))?, + generation: row.get::("generation") as u64, + minimum_protocol_version: row.get::("minimum_protocol_version") as u16, + session_id: row.get("session_id"), + issued_grant_handle: row.get("issued_grant_handle"), + held_grant_handle: row.get("held_grant_handle"), + issued_grant_id: row.get("issued_grant_id"), + held_grant_id: row.get("held_grant_id"), + created_at: row.get("created_at"), + }) +} diff --git a/crates/vnidrop/src/grant.rs b/crates/vnidrop/src/grant.rs index a79b70a..80165f1 100644 --- a/crates/vnidrop/src/grant.rs +++ b/crates/vnidrop/src/grant.rs @@ -64,6 +64,14 @@ impl GrantSecret { Self(random_bytes()) } + pub(crate) fn as_bytes(&self) -> &[u8; GRANT_SECRET_LEN] { + &self.0 + } + + pub(crate) fn from_bytes(bytes: [u8; GRANT_SECRET_LEN]) -> Self { + Self(bytes) + } + pub(crate) fn encode(&self) -> String { HEXLOWER.encode(&self.0) } @@ -96,10 +104,28 @@ impl Challenge { Self(random_bytes()) } + pub(crate) fn as_bytes(&self) -> &[u8; CHALLENGE_LEN] { + &self.0 + } + #[cfg(test)] pub(crate) fn from_bytes(bytes: [u8; CHALLENGE_LEN]) -> Self { Self(bytes) } + + pub(crate) fn encode(&self) -> String { + HEXLOWER.encode(&self.0) + } + + pub(crate) fn decode(value: &str) -> Result { + let bytes = HEXLOWER + .decode(value.as_bytes()) + .context("invalid challenge encoding")?; + let bytes: [u8; CHALLENGE_LEN] = bytes + .try_into() + .map_err(|_| anyhow::anyhow!("invalid challenge length"))?; + Ok(Self(bytes)) + } } impl fmt::Debug for Challenge { @@ -115,6 +141,16 @@ pub(crate) struct GrantProof { mac: [u8; PROOF_LEN], } +impl GrantProof { + pub(crate) fn from_parts(grant_id: GrantId, mac: [u8; PROOF_LEN]) -> Self { + Self { grant_id, mac } + } + + pub(crate) fn mac(&self) -> &[u8; PROOF_LEN] { + &self.mac + } +} + impl fmt::Debug for GrantProof { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("GrantProof") diff --git a/crates/vnidrop/src/lib.rs b/crates/vnidrop/src/lib.rs index db48fbe..9c68b6c 100644 --- a/crates/vnidrop/src/lib.rs +++ b/crates/vnidrop/src/lib.rs @@ -2,6 +2,7 @@ mod access_policy; mod api; mod approval; mod contacts; +mod device_relationship; mod error; mod event_hub; mod filesystem; diff --git a/crates/vnidrop/src/pairing_eligibility.rs b/crates/vnidrop/src/pairing_eligibility.rs index 9de55f6..1ce7bdd 100644 --- a/crates/vnidrop/src/pairing_eligibility.rs +++ b/crates/vnidrop/src/pairing_eligibility.rs @@ -148,24 +148,54 @@ impl PairingEligibilityService { /// /// Returns `false` when eligibility is missing/expired (silent reject). A /// successful start consumes the single-use eligibility for that session. + #[allow( + dead_code, + reason = "retained for eligibility-only callers; mutual consent uses take_eligibility" + )] pub(crate) async fn request_pairing( &self, peer_endpoint_id: &str, ) -> Result { + Ok(self.take_eligibility(peer_endpoint_id).await?.is_some()) + } + + /// Takes and consumes eligibility for a peer, returning the capability material. + pub(crate) async fn take_eligibility( + &self, + peer_endpoint_id: &str, + ) -> Result, VnidropError> { self.expire_due(true).await?; let entries = self .repository .list_pairing_eligibilities_for_peer(peer_endpoint_id) .await?; let Some(entry) = entries.into_iter().next() else { - return Ok(false); + return Ok(None); }; if entry.expires_at <= now_ms() { self.delete_entry_silent(&entry).await?; - return Ok(false); + return Ok(None); } + let Some(custody) = &self.custody else { + self.delete_entry_silent(&entry).await?; + return Ok(None); + }; + let capability = match custody + .load(&SecretHandle::from_stored(entry.secret_handle.clone())) + .await + { + Ok(material) => material, + Err(_) => { + self.delete_entry_silent(&entry).await?; + return Ok(None); + } + }; self.delete_entry(&entry).await?; - Ok(true) + Ok(Some(TakenEligibility { + session_id: entry.session_id, + protocol_version: entry.protocol_version, + capability, + })) } /// Validates an inbound eligibility presentation without prompts or events on failure. @@ -189,6 +219,24 @@ impl PairingEligibilityService { Ok(true) } + /// Consumes eligibility for one session without requiring the capability bytes. + pub(crate) async fn consume_session( + &self, + peer_endpoint_id: &str, + session_id: &str, + ) -> Result<(), VnidropError> { + if let Some(entry) = self + .repository + .find_pairing_eligibility_by_session(session_id) + .await? + { + if entry.peer_endpoint_id == peer_endpoint_id { + self.delete_entry(&entry).await?; + } + } + Ok(()) + } + pub(crate) async fn decline(&self, peer_endpoint_id: &str) -> Result<(), VnidropError> { self.remove_for_peer(peer_endpoint_id).await } @@ -304,6 +352,13 @@ pub(crate) struct PairingEligibilityInsert<'a> { pub(crate) expires_at: i64, } +#[derive(Debug, Clone)] +pub(crate) struct TakenEligibility { + pub(crate) session_id: String, + pub(crate) protocol_version: u16, + pub(crate) capability: SecretMaterial, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct PairingEligibilityRecord { pub(crate) peer_endpoint_id: String, diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index d715dfe..7b58375 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -26,7 +26,7 @@ use crate::{ util::now_ms, }; -const SCHEMA_VERSION: i64 = 11; +const SCHEMA_VERSION: i64 = 13; #[derive(Debug, Clone)] pub(crate) struct Repository { @@ -323,6 +323,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?; sqlx::query( r#" CREATE TABLE IF NOT EXISTS pairing_eligibilities ( @@ -358,6 +359,10 @@ impl Repository { ContactStore::new(self.pool.clone()) } + pub(crate) fn sqlite_pool(&self) -> SqlitePool { + self.pool.clone() + } + #[allow( dead_code, reason = "the private custody seam is activated by platform credential adapters" diff --git a/crates/vnidrop/src/runtime/contacts.rs b/crates/vnidrop/src/runtime/contacts.rs index d5e7052..94bba7a 100644 --- a/crates/vnidrop/src/runtime/contacts.rs +++ b/crates/vnidrop/src/runtime/contacts.rs @@ -59,8 +59,30 @@ impl CoreInner { &self, peer_endpoint_id: String, ) -> Result { - self.pairing_eligibility - .request_pairing(&peer_endpoint_id) + self.device_relationships + .request_pairing(peer_endpoint_id) + .await + } + + pub(super) async fn list_device_relationships( + &self, + ) -> Result, crate::error::VnidropError> { + self.device_relationships.list().await + } + + pub(super) async fn list_saved_devices( + &self, + ) -> Result, crate::error::VnidropError> { + self.device_relationships.list_saved_devices().await + } + + pub(super) async fn respond_to_device_pairing( + self: &Arc, + peer_endpoint_id: String, + accepted: bool, + ) -> Result { + self.device_relationships + .respond_to_pairing(peer_endpoint_id, accepted) .await } diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index 58fae6b..3c69861 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -397,6 +397,27 @@ impl VnidropCore { self.block_on(self.inner.request_saved_device_pairing(peer_endpoint_id)) } + pub fn list_device_relationships( + &self, + ) -> Result, VnidropError> { + self.block_on(self.inner.list_device_relationships()) + } + + pub fn list_saved_devices(&self) -> Result, VnidropError> { + self.block_on(self.inner.list_saved_devices()) + } + + pub fn respond_to_device_pairing( + &self, + peer_endpoint_id: String, + accepted: bool, + ) -> Result { + self.block_on( + self.inner + .respond_to_device_pairing(peer_endpoint_id, accepted), + ) + } + /// Devices the user has chosen to remember. pub fn list_contacts(&self) -> Result, VnidropError> { self.block_on(self.inner.list_contacts()) diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index e2baca7..1890ae3 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -56,6 +56,7 @@ use crate::{ access_policy::{mode_from_storage, AccessPolicy}, api::{CoreEvent, CoreEventSink, CoreLimits, CoreRelayMode}, approval::ApprovalService, + device_relationship::{DeviceRelationshipService, RelationshipProtocol}, event_hub::EventHub, handshake::HandshakeService, logging::init_logging, @@ -106,6 +107,7 @@ pub(super) struct CoreInner { pub(super) approval: ApprovalService, pub(super) pairing: PairingService, pub(super) pairing_eligibility: PairingEligibilityService, + pub(super) device_relationships: Arc, pub(super) offers: OfferInbox, /// Endpoint → last poll time, for the rate limit above. pub(super) last_polled: TokioMutex>, @@ -385,6 +387,14 @@ impl CoreInner { tracing::warn!(%error, "failed to sweep dead grants"); } let offers = OfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize); + let device_relationships = Arc::new(DeviceRelationshipService::new( + repository.sqlite_pool(), + secret_custody.clone(), + pairing_eligibility.clone(), + event_hub.clone(), + endpoint.id().to_string(), + endpoint.clone(), + )); let router = Router::builder(endpoint.clone()) .accept(iroh_blobs::ALPN, blobs) .accept(HandshakeService::ALPN, handshake) @@ -392,6 +402,10 @@ impl CoreInner { OfferService::ALPN, OfferService::new(pairing.clone(), offers.clone(), endpoint.id().to_string()), ) + .accept( + RelationshipProtocol::ALPN, + RelationshipProtocol::new(device_relationships.clone()), + ) .spawn(); let inner = Arc::new(Self { @@ -405,6 +419,7 @@ impl CoreInner { approval, pairing, pairing_eligibility, + device_relationships, offers, last_polled: TokioMutex::new(HashMap::new()), relay_mode, @@ -436,6 +451,9 @@ impl CoreInner { if let Err(error) = inner.pairing_eligibility.reconcile().await { tracing::warn!(%error, "failed to reconcile pairing eligibility"); } + if let Err(error) = inner.device_relationships.reconcile().await { + tracing::warn!(%error, "failed to reconcile device relationships"); + } Ok(inner) } diff --git a/crates/vnidrop/src/secure_secret.rs b/crates/vnidrop/src/secure_secret.rs index bd27cb8..9ba1db0 100644 --- a/crates/vnidrop/src/secure_secret.rs +++ b/crates/vnidrop/src/secure_secret.rs @@ -52,6 +52,14 @@ impl SecretMaterial { let bytes: [u8; SECRET_BYTES] = self.0.try_into().expect("validated length"); SecretKey::from_bytes(&bytes) } + + pub(crate) fn as_bytes(&self) -> &[u8] { + &self.0 + } + + pub(crate) fn to_vec(&self) -> Vec { + self.0.clone() + } } impl fmt::Debug for SecretMaterial { diff --git a/crates/vnidrop/src/tests.rs b/crates/vnidrop/src/tests.rs index b0f9865..4ef6972 100644 --- a/crates/vnidrop/src/tests.rs +++ b/crates/vnidrop/src/tests.rs @@ -4,6 +4,8 @@ mod access_policy_tests; mod contact_polling_tests; #[path = "tests/contacts.rs"] mod contacts_tests; +#[path = "tests/device_relationship.rs"] +mod device_relationship_tests; #[path = "tests/error.rs"] mod error_tests; #[path = "tests/filesystem.rs"] diff --git a/crates/vnidrop/src/tests/device_relationship.rs b/crates/vnidrop/src/tests/device_relationship.rs new file mode 100644 index 0000000..dece2fc --- /dev/null +++ b/crates/vnidrop/src/tests/device_relationship.rs @@ -0,0 +1,317 @@ +use std::{ + path::Path, + sync::{Arc, Mutex}, + time::{Duration, Instant}, +}; + +use crate::{ + secure_secret::FaultInjectingSecretStore, CoreEvent, CoreEventSink, DeviceRelationshipState, + ShareMetadataInput, ShareSource, SourceKind, TransferAccessMode, VnidropCore, +}; + +struct RecordingSink { + events: Mutex>, +} + +impl CoreEventSink for RecordingSink { + fn on_event(&self, event: CoreEvent) { + self.events.lock().unwrap().push(event); + } +} + +struct ProtectedNode { + _data_dir: tempfile::TempDir, + core: Arc, +} + +impl ProtectedNode { + fn new() -> Self { + let data_dir = tempfile::tempdir().unwrap(); + let sink = Arc::new(RecordingSink { + events: Mutex::new(Vec::new()), + }); + let store = Arc::new(FaultInjectingSecretStore::default()); + let core = VnidropCore::initialize_with_test_secret_store( + data_dir.path().to_string_lossy().into_owned(), + sink, + store, + ) + .expect("protected test core"); + Self { + _data_dir: data_dir, + core, + } + } +} + +impl Drop for ProtectedNode { + fn drop(&mut self) { + self.core.shutdown(); + } +} + +fn share_path(core: &VnidropCore, source: &Path, transfer_id: u64) -> crate::ShareResult { + core.share_files( + vec![ShareSource { + kind: SourceKind::Path, + value: source.to_string_lossy().into_owned(), + display_name: Some("hello.txt".to_string()), + is_directory: false, + }], + ShareMetadataInput { + transfer_id, + transfer_name: Some("hello.txt".to_string()), + sender_name: Some("sender".to_string()), + access_mode: TransferAccessMode::ApprovalRequired, + }, + ) + .unwrap() +} + +fn wait_for_receiver_request(sender: &VnidropCore, transfer_id: u64) -> crate::ReceiverRequest { + let started = Instant::now(); + loop { + if let Some(request) = sender + .list_receiver_requests(transfer_id) + .unwrap() + .into_iter() + .find(|request| request.status == "requested") + { + return request; + } + assert!( + started.elapsed() < Duration::from_secs(15), + "timed out waiting for receiver request" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +fn complete_transfer(sender: &ProtectedNode, receiver: &ProtectedNode, transfer_id: u64) { + let source_dir = tempfile::tempdir().unwrap(); + let output_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("hello.txt"); + std::fs::write(&source_path, b"mutual consent").unwrap(); + let share = share_path(&sender.core, &source_path, transfer_id); + let output_dir = output_dir.path().to_string_lossy().to_string(); + let receiver_core = receiver.core.clone(); + let ticket = share.ticket.clone(); + let handle = std::thread::spawn(move || { + receiver_core.receive(ticket, output_dir, Some("receiver".to_string())) + }); + let request = wait_for_receiver_request(&sender.core, share.transfer_id); + sender + .core + .respond_receiver_request(request.id, true, None) + .unwrap(); + handle.join().unwrap().unwrap(); + + let started = Instant::now(); + let peer = receiver.core.status().endpoint_id.clone(); + loop { + if sender + .core + .list_pairing_eligibilities() + .unwrap() + .iter() + .any(|entry| entry.peer_endpoint_id == peer) + { + break; + } + assert!( + started.elapsed() < Duration::from_secs(10), + "eligibility never appeared" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +fn wait_for_relationship( + core: &VnidropCore, + peer: &str, + state: DeviceRelationshipState, +) -> crate::DeviceRelationship { + let started = Instant::now(); + loop { + if let Some(relationship) = core + .list_device_relationships() + .unwrap() + .into_iter() + .find(|entry| entry.remote_endpoint_id == peer && entry.state == state) + { + return relationship; + } + assert!( + started.elapsed() < Duration::from_secs(15), + "relationship {peer} never reached {state:?}" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +#[test] +fn mutual_consent_reaches_saved_after_both_grants_and_acknowledgement() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + let alice_id = alice.core.status().endpoint_id.clone(); + let bob_id = bob.core.status().endpoint_id.clone(); + + complete_transfer(&alice, &bob, 80_001); + + assert!(alice + .core + .request_saved_device_pairing(bob_id.clone()) + .unwrap()); + + wait_for_relationship( + &alice.core, + &bob_id, + DeviceRelationshipState::PendingOutgoing, + ); + wait_for_relationship( + &bob.core, + &alice_id, + DeviceRelationshipState::PendingIncoming, + ); + assert!( + alice.core.list_saved_devices().unwrap().is_empty(), + "pending outgoing must not surface as a saved device" + ); + assert!( + bob.core.list_saved_devices().unwrap().is_empty(), + "pending incoming must not surface as a saved device" + ); + + 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); + + let alice_saved = alice.core.list_saved_devices().unwrap(); + let bob_saved = bob.core.list_saved_devices().unwrap(); + assert_eq!(alice_saved.len(), 1); + assert_eq!(alice_saved[0].endpoint_id, bob_id); + assert_eq!(bob_saved.len(), 1); + assert_eq!(bob_saved[0].endpoint_id, alice_id); +} + +#[test] +fn declining_pending_incoming_consumes_eligibility_and_never_saves() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + let alice_id = alice.core.status().endpoint_id.clone(); + let bob_id = bob.core.status().endpoint_id.clone(); + complete_transfer(&alice, &bob, 80_010); + + 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(), false) + .unwrap()); + assert!(bob.core.list_saved_devices().unwrap().is_empty()); + assert!(alice.core.list_saved_devices().unwrap().is_empty()); + assert!(bob + .core + .list_pairing_eligibilities() + .unwrap() + .iter() + .all(|entry| entry.peer_endpoint_id != alice_id)); + // Declined eligibility cannot prompt again without a new qualifying transfer. + assert!(!alice.core.request_saved_device_pairing(bob_id).unwrap()); +} + +#[test] +fn repeated_consent_is_idempotent_and_does_not_duplicate_saved_rows() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + let alice_id = alice.core.status().endpoint_id.clone(); + let bob_id = bob.core.status().endpoint_id.clone(); + complete_transfer(&alice, &bob, 80_020); + + 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); + + assert!(bob.core.respond_to_device_pairing(alice_id, true).unwrap()); + assert_eq!(alice.core.list_saved_devices().unwrap().len(), 1); + assert_eq!(bob.core.list_saved_devices().unwrap().len(), 1); + assert_eq!(alice.core.list_device_relationships().unwrap().len(), 1); + assert_eq!(bob.core.list_device_relationships().unwrap().len(), 1); +} + +#[test] +fn simultaneous_initiation_merges_into_one_relationship_per_side() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + let alice_id = alice.core.status().endpoint_id.clone(); + let bob_id = bob.core.status().endpoint_id.clone(); + complete_transfer(&alice, &bob, 80_030); + + let alice_core = alice.core.clone(); + let bob_core = bob.core.clone(); + let bob_id_clone = bob_id.clone(); + let alice_id_clone = alice_id.clone(); + let alice_handle = + std::thread::spawn(move || alice_core.request_saved_device_pairing(bob_id_clone)); + let bob_handle = + std::thread::spawn(move || bob_core.request_saved_device_pairing(alice_id_clone)); + let alice_ok = alice_handle.join().unwrap().unwrap(); + let bob_ok = bob_handle.join().unwrap().unwrap(); + assert!(alice_ok || bob_ok); + + wait_for_relationship(&alice.core, &bob_id, DeviceRelationshipState::Saved); + wait_for_relationship(&bob.core, &alice_id, DeviceRelationshipState::Saved); + assert_eq!(alice.core.list_device_relationships().unwrap().len(), 1); + assert_eq!(bob.core.list_device_relationships().unwrap().len(), 1); + assert_eq!(alice.core.list_saved_devices().unwrap().len(), 1); + assert_eq!(bob.core.list_saved_devices().unwrap().len(), 1); +} + +#[test] +fn request_timeout_leaves_recoverable_pending_not_saved() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + let bob_id = bob.core.status().endpoint_id.clone(); + complete_transfer(&alice, &bob, 80_040); + + // Shut down Bob so Alice's pairing request cannot complete on the wire. + bob.core.shutdown(); + + assert!(alice + .core + .request_saved_device_pairing(bob_id.clone()) + .unwrap()); + wait_for_relationship( + &alice.core, + &bob_id, + DeviceRelationshipState::PendingOutgoing, + ); + assert!( + alice.core.list_saved_devices().unwrap().is_empty(), + "timed-out pairing must not surface as saved" + ); +} diff --git a/crates/vnidrop/src/tests/repository.rs b/crates/vnidrop/src/tests/repository.rs index d963bf0..d444e79 100644 --- a/crates/vnidrop/src/tests/repository.rs +++ b/crates/vnidrop/src/tests/repository.rs @@ -66,7 +66,7 @@ async fn received_artifacts_survive_history_deletion() { async fn persists_transfers_and_events_across_reopen() { let temp = tempfile::tempdir().unwrap(); let repository = Repository::open(temp.path()).await.unwrap(); - assert_eq!(repository.schema_version().await.unwrap(), 11); + assert_eq!(repository.schema_version().await.unwrap(), 13); repository .insert_transfer(transfer( 7, @@ -645,7 +645,7 @@ async fn migrates_schema_v2_identity_without_losing_transfer() { pool.close().await; let repository = Repository::open(temp.path()).await.unwrap(); - assert_eq!(repository.schema_version().await.unwrap(), 11); + assert_eq!(repository.schema_version().await.unwrap(), 13); let stored = repository.list_transfers().await.unwrap().remove(0); assert_eq!(stored.transfer_id, 7); assert_eq!(stored.local_id, "legacy-7-send");