diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index 37c7bd0..36896e5 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -288,10 +288,22 @@ pub struct CoreLimits { pub max_metadata_bytes: u64, pub max_events: u64, pub max_pending_approvals: u64, - /// Incoming pairing offers awaiting the local user's decision. + /// Incoming pairing / targeted offers awaiting the local user's decision. pub max_pending_offers: u64, pub max_concurrent_transfers: u64, pub event_queue_capacity: u64, + /// Cap on Saved + pending mutual-consent relationships (design §14). + pub max_saved_devices: u64, + /// Quiet period after a decline or repeated malformed control-plane traffic. + pub identity_cooldown_ms: u64, + /// Malformed control-plane messages from one identity before cooldown. + pub malformed_strike_limit: u64, + /// Pairing RPC / acknowledgement wait bound (milliseconds). + pub pairing_timeout_ms: u64, + /// Pre-approval offer decision wait bound (milliseconds). + pub offer_timeout_ms: u64, + /// Connection establishment bound for targeted transfers (milliseconds). + pub connection_timeout_ms: u64, } impl Default for CoreLimits { @@ -313,6 +325,12 @@ impl Default for CoreLimits { max_pending_offers: 16, max_concurrent_transfers: 8, event_queue_capacity: 1_024, + max_saved_devices: 256, + identity_cooldown_ms: 60_000, + malformed_strike_limit: 5, + pairing_timeout_ms: 15_000, + offer_timeout_ms: 120_000, + connection_timeout_ms: 30_000, } } } @@ -331,6 +349,12 @@ impl CoreLimits { ("max_pending_offers", self.max_pending_offers), ("max_concurrent_transfers", self.max_concurrent_transfers), ("event_queue_capacity", self.event_queue_capacity), + ("max_saved_devices", self.max_saved_devices), + ("identity_cooldown_ms", self.identity_cooldown_ms), + ("malformed_strike_limit", self.malformed_strike_limit), + ("pairing_timeout_ms", self.pairing_timeout_ms), + ("offer_timeout_ms", self.offer_timeout_ms), + ("connection_timeout_ms", self.connection_timeout_ms), ]; for (name, value) in positive { if value == 0 { @@ -342,6 +366,8 @@ impl CoreLimits { ("max_pending_offers", self.max_pending_offers), ("max_concurrent_transfers", self.max_concurrent_transfers), ("event_queue_capacity", self.event_queue_capacity), + ("max_saved_devices", self.max_saved_devices), + ("malformed_strike_limit", self.malformed_strike_limit), ] { usize::try_from(value) .with_context(|| format!("core limit {name} exceeds platform capacity"))?; diff --git a/crates/vnidrop/src/control_plane.rs b/crates/vnidrop/src/control_plane.rs new file mode 100644 index 0000000..8a9965f --- /dev/null +++ b/crates/vnidrop/src/control_plane.rs @@ -0,0 +1,280 @@ +//! Saved-device control-plane hardening (design §14). +//! +//! Bounds hostile / noisy peers without imposing quotas on transfers the +//! receiver has already accepted. + +use std::{ + collections::HashMap, + sync::{Arc, Mutex}, +}; + +use data_encoding::HEXLOWER; +use serde_json::{Map, Value}; + +use crate::util::now_ms; + +/// Per-identity quiet period after declines or repeated malformed traffic. +#[derive(Clone)] +pub(crate) struct IdentityCooldown { + inner: Arc>, + cooldown_ms: i64, + strike_limit: u32, +} + +#[derive(Default)] +struct CooldownInner { + until: HashMap, + strikes: HashMap, +} + +impl IdentityCooldown { + pub(crate) fn new(cooldown_ms: u64, strike_limit: u64) -> Self { + Self { + inner: Arc::new(Mutex::new(CooldownInner::default())), + cooldown_ms: cooldown_ms as i64, + strike_limit: strike_limit as u32, + } + } + + pub(crate) fn is_cooling(&self, identity: &str) -> bool { + let now = now_ms(); + let mut state = self + .inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.until.retain(|_, until| *until > now); + state.until.contains_key(identity) + } + + pub(crate) fn record_decline(&self, identity: &str) { + let until = now_ms() + self.cooldown_ms; + let mut state = self + .inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.until.insert(identity.to_string(), until); + state.strikes.remove(identity); + } + + /// Count a malformed / spoofed / ineligible control-plane message. + /// + /// Trips cooldown once the strike limit is reached. Returns whether the + /// identity is now cooling (including an already-active cooldown). + pub(crate) fn record_malformed(&self, identity: &str) -> bool { + if self.is_cooling(identity) { + return true; + } + let mut state = self + .inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let strikes = state.strikes.entry(identity.to_string()).or_insert(0); + *strikes = strikes.saturating_add(1); + if *strikes >= self.strike_limit { + state + .until + .insert(identity.to_string(), now_ms() + self.cooldown_ms); + state.strikes.remove(identity); + return true; + } + false + } + + pub(crate) fn clear_strikes(&self, identity: &str) { + let mut state = self + .inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.strikes.remove(identity); + } +} + +const REDACTED: &str = "[redacted]"; + +/// Keys whose values must never appear in production events / diagnostics. +fn is_sensitive_key(key: &str) -> bool { + matches!( + key, + "endpoint_id" + | "peer_endpoint_id" + | "sender_endpoint_id" + | "receiver_endpoint_id" + | "from_endpoint_id" + | "remote_endpoint_id" + | "local_endpoint_id" + | "ticket" + | "blob_ticket" + | "authorization" + | "capability" + | "secret" + | "grant" + | "grant_id" + | "proof" + | "mac" + | "filename" + | "file_name" + | "path" + | "display_name" + | "transfer_name" + | "sender_display_name" + | "remote_display_name" + | "address" + | "addrs" + | "relay_url" + | "relay_urls" + ) +} + +/// Stable fingerprint so diagnostics can correlate without leaking raw values. +pub(crate) fn fingerprint(value: &str) -> String { + let digest = blake3::hash(value.as_bytes()); + let hex = HEXLOWER.encode(digest.as_bytes()); + format!("", &hex[..8]) +} + +pub(crate) fn redact_json(value: Value) -> Value { + match value { + Value::Object(map) => Value::Object(redact_object(map)), + Value::Array(items) => Value::Array(items.into_iter().map(redact_json).collect()), + other => other, + } +} + +fn redact_object(map: Map) -> Map { + map.into_iter() + .map(|(key, value)| { + if is_sensitive_key(&key) { + let redacted = match value { + Value::String(raw) if !raw.is_empty() => Value::String(fingerprint(&raw)), + Value::Null => Value::Null, + _ => Value::String(REDACTED.to_string()), + }; + (key, redacted) + } else { + (key, redact_json(value)) + } + }) + .collect() +} + +/// Scrub ticket-like and long opaque blobs from free-form error / log text. +pub(crate) fn redact_text(input: &str) -> String { + let mut out = String::with_capacity(input.len()); + let mut rest = input; + while let Some(idx) = rest.find("vnd1:") { + out.push_str(&rest[..idx]); + out.push_str(REDACTED); + rest = &rest[idx + 5..]; + // Skip the remainder of the ticket token (non-whitespace). + let end = rest + .find(|c: char| c.is_whitespace() || c == '"' || c == '\'') + .unwrap_or(rest.len()); + rest = &rest[end..]; + } + out.push_str(rest); + // Collapse long hex runs that look like endpoint ids / grant material. + collapse_long_hex(&out) +} + +fn collapse_long_hex(input: &str) -> String { + let mut out = String::with_capacity(input.len()); + let bytes = input.as_bytes(); + let mut i = 0; + while i < bytes.len() { + if bytes[i].is_ascii_hexdigit() { + let start = i; + while i < bytes.len() && bytes[i].is_ascii_hexdigit() { + i += 1; + } + let len = i - start; + if len >= 32 { + out.push_str(REDACTED); + } else { + out.push_str(&input[start..i]); + } + } else { + out.push(bytes[i] as char); + i += 1; + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + /// Documented non-enforcement: accepted transfers have no per-device quota. + const ACCEPTED_TRANSFER_QUOTA_FIELDS: &[&str] = &[ + "max_per_device_files", + "max_per_device_bytes", + "max_per_device_bandwidth", + "max_per_device_transfers", + ]; + + #[test] + fn cooldown_trips_after_strike_limit_and_isolates_identities() { + let guard = IdentityCooldown::new(60_000, 3); + assert!(!guard.record_malformed("a")); + assert!(!guard.record_malformed("a")); + assert!(guard.record_malformed("a")); + assert!(guard.is_cooling("a")); + assert!(!guard.is_cooling("b")); + guard.record_decline("b"); + assert!(guard.is_cooling("b")); + assert!(guard.is_cooling("a")); + } + + #[test] + fn redaction_scrubs_sensitive_event_fields() { + let raw = json!({ + "transfer_id": "ok-to-keep", + "sender_endpoint_id": "abc123endpointid000000000000000000000000000000000000000000000000", + "transfer_name": "secret.pdf", + "ticket": "vnd1:deadbeef", + "file_count": 2, + "nested": { "capability": [1, 2, 3], "state": "saved" } + }); + let redacted = redact_json(raw); + let obj = redacted.as_object().unwrap(); + assert_eq!(obj.get("transfer_id").unwrap(), "ok-to-keep"); + assert_eq!(obj.get("file_count").unwrap(), 2); + let sender = obj.get("sender_endpoint_id").unwrap().as_str().unwrap(); + assert!(sender.starts_with(", + max_saved_devices: u64, + pairing_timeout: Duration, peer_locks: Arc>>>>, } @@ -75,6 +76,8 @@ impl DeviceRelationshipService { endpoint: Endpoint, relay_mode: CoreRelayMode, custom_relay_urls: Vec, + max_saved_devices: u64, + pairing_timeout_ms: u64, ) -> Self { Self { pool, @@ -85,10 +88,36 @@ impl DeviceRelationshipService { endpoint, relay_mode, custom_relay_urls, + max_saved_devices, + pairing_timeout: Duration::from_millis(pairing_timeout_ms), peer_locks: Arc::new(TokioMutex::new(HashMap::new())), } } + /// Slots consumed by Saved or in-flight mutual-consent relationships. + async fn relationship_slots_used(&self) -> Result { + let row = sqlx::query( + r#" + SELECT COUNT(*) AS n FROM device_relationships + WHERE state IN ('saved', 'pending_outgoing', 'pending_incoming') + "#, + ) + .fetch_one(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(row.get::("n") as u64) + } + + async fn can_create_new_relationship( + &self, + peer_endpoint_id: &str, + ) -> Result { + if self.find_row(peer_endpoint_id).await?.is_some() { + return Ok(true); + } + Ok(self.relationship_slots_used().await? < self.max_saved_devices) + } + pub(super) async fn lock_peer(&self, peer_endpoint_id: &str) -> Arc> { let mut locks = self.peer_locks.lock().await; locks @@ -254,6 +283,10 @@ impl DeviceRelationshipService { } } + if !self.can_create_new_relationship(&peer_endpoint_id).await? { + return Ok(false); + } + let Some(taken) = self.eligibility.take_eligibility(&peer_endpoint_id).await? else { return Ok(false); }; @@ -316,7 +349,7 @@ impl DeviceRelationshipService { 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, + self.pairing_timeout, client.pairing_request(PairingRequest { session_id: taken.session_id.clone(), capability: taken.capability.to_vec(), @@ -382,7 +415,7 @@ impl DeviceRelationshipService { let addr = self.peer_addr(&peer_endpoint_id).await?; let client = RelationshipClient::connect(self.endpoint.clone(), addr); let _ = tokio::time::timeout( - PAIRING_RPC_TIMEOUT, + self.pairing_timeout, client.pairing_consent(PairingConsent { accepted: false, grant: None, @@ -413,7 +446,7 @@ impl DeviceRelationshipService { 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, + self.pairing_timeout, client.pairing_consent(PairingConsent { accepted: true, grant: Some(grant.clone()), @@ -456,7 +489,7 @@ impl DeviceRelationshipService { ) .await?; match tokio::time::timeout( - PAIRING_RPC_TIMEOUT, + self.pairing_timeout, client.pairing_ack(PairingAck { possession_proof: proof, challenge: ack_challenge.encode(), @@ -542,6 +575,11 @@ impl DeviceRelationshipService { return PairingRequestResponse::Rejected; } + match self.can_create_new_relationship(&remote_endpoint_id).await { + Ok(true) => {} + Ok(false) | Err(_) => return PairingRequestResponse::Rejected, + } + let now = now_ms(); if self .upsert_relationship(RelationshipUpsert { @@ -726,7 +764,7 @@ impl DeviceRelationshipService { 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, + self.pairing_timeout, client.pairing_consent(PairingConsent { accepted: true, grant: Some(grant), @@ -767,7 +805,7 @@ impl DeviceRelationshipService { .await?; if let Ok(Ok(PairingAckResponse::Acknowledged | PairingAckResponse::AlreadySaved)) = tokio::time::timeout( - PAIRING_RPC_TIMEOUT, + self.pairing_timeout, client.pairing_ack(PairingAck { possession_proof: proof, challenge: ack_challenge.encode(), @@ -1483,7 +1521,7 @@ impl DeviceRelationshipService { }; let client = RelationshipClient::connect(self.endpoint.clone(), addr); let _ = tokio::time::timeout( - PAIRING_RPC_TIMEOUT, + self.pairing_timeout, client.revoke_notice(RevokeNotice { generation, issued_grant_id, diff --git a/crates/vnidrop/src/error.rs b/crates/vnidrop/src/error.rs index 5f4486d..f06ed0d 100644 --- a/crates/vnidrop/src/error.rs +++ b/crates/vnidrop/src/error.rs @@ -55,7 +55,7 @@ impl VnidropError { pub(crate) fn ticket(error: impl Into) -> Self { Self::Ticket { - reason: error.into().to_string(), + reason: crate::control_plane::redact_text(&error.into().to_string()), } } @@ -168,7 +168,7 @@ impl VnidropError { } fn classify(error: anyhow::Error, fallback: impl FnOnce(String) -> Self) -> Self { - let reason = error.to_string(); + let reason = crate::control_plane::redact_text(&error.to_string()); if let Some(existing) = error.chain().find_map(|cause| cause.downcast_ref::()) { return existing.with_reason(reason); } @@ -193,7 +193,7 @@ impl VnidropError { } fn from_error(error: anyhow::Error, fallback: impl FnOnce(String) -> Self) -> Self { - let reason = error.to_string(); + let reason = crate::control_plane::redact_text(&error.to_string()); if let Some(existing) = error.chain().find_map(|cause| cause.downcast_ref::()) { existing.with_reason(reason) } else { diff --git a/crates/vnidrop/src/event_hub.rs b/crates/vnidrop/src/event_hub.rs index c04ec88..74921c5 100644 --- a/crates/vnidrop/src/event_hub.rs +++ b/crates/vnidrop/src/event_hub.rs @@ -8,6 +8,7 @@ use tokio::{ use crate::{ api::{CoreEvent, CoreEventSink}, + control_plane::redact_json, repository::Repository, transfer_state::TransferDirection, util::now_ms, @@ -51,6 +52,10 @@ enum EventPhase { Pairing, /// Prototype contact lifecycle notifications. Contacts, + /// Prototype contact offer prompts. + Offer, + /// Saved-device targeted-transfer pre-approval prompts. + TargetedTransfer, } impl EventPhase { @@ -74,6 +79,8 @@ impl EventPhase { "delivery" => Some(Self::Delivery), "pairing" => Some(Self::Pairing), "contacts" => Some(Self::Contacts), + "offer" => Some(Self::Offer), + "targeted_transfer" => Some(Self::TargetedTransfer), _ => None, } } @@ -98,6 +105,8 @@ impl EventPhase { Self::Delivery => "delivery", Self::Pairing => "pairing", Self::Contacts => "contacts", + Self::Offer => "offer", + Self::TargetedTransfer => "targeted_transfer", } } } @@ -245,6 +254,8 @@ impl EventHub { // Compose observes this event synchronously, while SQLite persistence is // serialized through the queue. That keeps the UI responsive without // losing the ability to flush persisted history during shutdown/tests. + // Production diagnostics redact endpoint ids, tickets, grants, and paths; + // typed UniFFI list APIs still expose the values the UI needs. let event = CoreEvent { id, timestamp, @@ -253,7 +264,7 @@ impl EventHub { direction: direction.map(|direction| direction.as_str().to_string()), phase: phase.as_str().to_string(), kind: kind.0, - data_json: data.to_string(), + data_json: redact_json(data).to_string(), }; if let Err(error) = self.tx.try_send(EventCommand::Persist(event.clone())) { tracing::warn!(event_id = %event.id, %error, "event persistence queue dropped event"); diff --git a/crates/vnidrop/src/lib.rs b/crates/vnidrop/src/lib.rs index 7851553..2e2cbc8 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 control_plane; mod device_relationship; mod error; mod event_hub; diff --git a/crates/vnidrop/src/offer_inbox.rs b/crates/vnidrop/src/offer_inbox.rs index ced0948..320b819 100644 --- a/crates/vnidrop/src/offer_inbox.rs +++ b/crates/vnidrop/src/offer_inbox.rs @@ -20,10 +20,6 @@ use crate::{event_hub::EventHub, offer::OfferResponse, util::now_ms}; /// How long the sender waits for the receiving user to decide. const OFFER_WAIT_TIMEOUT: Duration = Duration::from_secs(120); -/// Quiet period after a decline, so a paired device cannot re-prompt on a loop. -/// A contact is not a stranger, but it is not unlimited either. -const DECLINE_COOLDOWN_MS: i64 = 60 * 1_000; - #[derive(Debug, Clone)] pub(crate) struct PendingOffer { pub(crate) offer_id: String, @@ -50,16 +46,22 @@ pub(crate) struct OfferInbox { /// Endpoint → time before which new offers are refused. cooldowns: Arc>>, max_pending: usize, + decline_cooldown_ms: i64, } impl OfferInbox { - pub(crate) fn new(event_hub: Arc, max_pending: usize) -> Self { + pub(crate) fn new( + event_hub: Arc, + max_pending: usize, + identity_cooldown_ms: u64, + ) -> Self { Self { event_hub, pending: Arc::new(Mutex::new(HashMap::new())), waiters: Arc::new(Mutex::new(HashMap::new())), cooldowns: Arc::new(Mutex::new(HashMap::new())), max_pending, + decline_cooldown_ms: identity_cooldown_ms as i64, } } @@ -229,7 +231,7 @@ impl OfferInbox { if !accepted { self.cooldowns.lock().await.insert( offer.from_endpoint_id.clone(), - now_ms() + DECLINE_COOLDOWN_MS, + now_ms() + self.decline_cooldown_ms, ); } if let Some(waiter) = waiter { diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index ee73b01..618ede6 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -90,6 +90,22 @@ impl VnidropCore { event_sink: Arc, store: Arc, network_config: CoreNetworkConfig, + ) -> Result, VnidropError> { + Self::initialize_with_test_secret_store_limits_and_network( + app_data_dir, + event_sink, + store, + CoreLimits::default(), + network_config, + ) + } + + pub(crate) fn initialize_with_test_secret_store_limits_and_network( + app_data_dir: String, + event_sink: Arc, + store: Arc, + limits: CoreLimits, + network_config: CoreNetworkConfig, ) -> Result, VnidropError> { let app_data_path = PathBuf::from(&app_data_dir); std::fs::create_dir_all(&app_data_path).map_err(VnidropError::filesystem)?; @@ -99,7 +115,7 @@ impl VnidropCore { Self::initialize_with_identity_mode( app_data_dir, event_sink, - CoreLimits::default(), + limits, network_config, IdentityMode::Protected { store, diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index 402d56d..4e503bf 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -394,9 +394,21 @@ impl CoreInner { { tracing::warn!(%error, "failed to sweep dead grants"); } - let offers = OfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize); - let targeted_offers = - TargetedOfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize); + let offers = OfferInbox::new( + event_hub.clone(), + limits.max_pending_offers as usize, + limits.identity_cooldown_ms, + ); + let identity_cooldown = crate::control_plane::IdentityCooldown::new( + limits.identity_cooldown_ms, + limits.malformed_strike_limit, + ); + let targeted_offers = TargetedOfferInbox::new( + event_hub.clone(), + limits.max_pending_offers as usize, + identity_cooldown, + limits.offer_timeout_ms, + ); let device_relationships = Arc::new(DeviceRelationshipService::new( repository.sqlite_pool(), secret_custody.clone(), @@ -406,6 +418,8 @@ impl CoreInner { endpoint.clone(), relay_mode, relay_urls.clone(), + limits.max_saved_devices, + limits.pairing_timeout_ms, )); let router = Router::builder(endpoint.clone()) .accept(iroh_blobs::ALPN, blobs) diff --git a/crates/vnidrop/src/runtime/targeted.rs b/crates/vnidrop/src/runtime/targeted.rs index 45c3fd2..f2928d0 100644 --- a/crates/vnidrop/src/runtime/targeted.rs +++ b/crates/vnidrop/src/runtime/targeted.rs @@ -27,9 +27,15 @@ use crate::{ util::{non_empty, now_ms}, }; -const OFFER_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); - impl CoreInner { + fn connection_timeout(&self) -> std::time::Duration { + std::time::Duration::from_millis(self.limits.connection_timeout_ms) + } + + fn offer_wait_timeout(&self) -> std::time::Duration { + std::time::Duration::from_millis(self.limits.offer_timeout_ms) + } + pub(super) fn targeted_store(&self) -> TargetedTransferStore { TargetedTransferStore::new(self.repository.sqlite_pool()) } @@ -131,7 +137,7 @@ impl CoreInner { { let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr); let _ = tokio::time::timeout( - OFFER_CONNECT_TIMEOUT, + self.connection_timeout(), client.cancel_offer(CancelTargetedOffer { transfer_id: id.clone(), }), @@ -287,7 +293,8 @@ impl CoreInner { .await?; let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr); let challenge = - match tokio::time::timeout(OFFER_CONNECT_TIMEOUT, client.request_challenge()).await { + match tokio::time::timeout(self.connection_timeout(), client.request_challenge()).await + { Ok(Ok(challenge)) => challenge, Ok(Err(error)) => { let _ = store @@ -331,7 +338,7 @@ impl CoreInner { .await?; let response = match tokio::time::timeout( - OFFER_CONNECT_TIMEOUT + std::time::Duration::from_secs(120), + self.connection_timeout() + self.offer_wait_timeout(), client.submit_offer(SubmitTargetedOffer { proof, generation, diff --git a/crates/vnidrop/src/targeted_transfer/inbox.rs b/crates/vnidrop/src/targeted_transfer/inbox.rs index ba6b060..3800a4d 100644 --- a/crates/vnidrop/src/targeted_transfer/inbox.rs +++ b/crates/vnidrop/src/targeted_transfer/inbox.rs @@ -10,9 +10,7 @@ use serde_json::json; use tokio::sync::{watch, Mutex}; use uuid::Uuid; -use crate::{api::PendingTargetedOffer, event_hub::EventHub}; - -const OFFER_WAIT_TIMEOUT: Duration = Duration::from_secs(120); +use crate::{api::PendingTargetedOffer, control_plane::IdentityCooldown, event_hub::EventHub}; #[derive(Debug, Clone)] pub(crate) struct PendingTargetedOfferRecord { @@ -40,27 +38,45 @@ pub(crate) struct TargetedOfferInbox { decisions: Arc>>, auths: Arc>>, settled: Arc>>, + cooldown: IdentityCooldown, max_pending: usize, + offer_timeout: Duration, } impl TargetedOfferInbox { - pub(crate) fn new(event_hub: Arc, max_pending: usize) -> Self { + pub(crate) fn new( + event_hub: Arc, + max_pending: usize, + cooldown: IdentityCooldown, + offer_timeout_ms: u64, + ) -> Self { Self { event_hub, pending: Arc::new(Mutex::new(HashMap::new())), decisions: Arc::new(Mutex::new(HashMap::new())), auths: Arc::new(Mutex::new(HashMap::new())), settled: Arc::new(Mutex::new(HashMap::new())), + cooldown, max_pending, + offer_timeout: Duration::from_millis(offer_timeout_ms), } } + pub(crate) fn cooldown(&self) -> &IdentityCooldown { + &self.cooldown + } + /// Surface a validated offer and block until the local user decides. /// /// Replaying the same transfer identity returns the settled result or joins /// the existing pending wait — never a second prompt. pub(crate) async fn submit(&self, offer: PendingTargetedOffer) -> TargetedOfferDecision { let transfer_id = offer.transfer_id.clone(); + if self.cooldown.is_cooling(&offer.sender_endpoint_id) { + return TargetedOfferDecision::Refused { + reason: "identity-cooldown".to_string(), + }; + } if let Some(settled) = self.settled.lock().await.get(&transfer_id).cloned() { return settled_to_decision(settled); } @@ -76,11 +92,7 @@ impl TargetedOfferInbox { reason: "immutable-transfer-mismatch".to_string(), }; } - if pending.len() >= self.max_pending { - return TargetedOfferDecision::Refused { - reason: "too-many-pending-offers".to_string(), - }; - } + // Prefer the more specific per-sender refusal before the global bound. if pending .values() .any(|entry| entry.offer.sender_endpoint_id == offer.sender_endpoint_id) @@ -89,6 +101,11 @@ impl TargetedOfferInbox { reason: "offer-already-pending".to_string(), }; } + if pending.len() >= self.max_pending { + return TargetedOfferDecision::Refused { + reason: "too-many-pending-offers".to_string(), + }; + } } let (decision_tx, _decision_rx) = watch::channel(None); @@ -99,6 +116,19 @@ impl TargetedOfferInbox { drop(pending); return self.wait_existing_decision(&transfer_id).await; } + if pending + .values() + .any(|entry| entry.offer.sender_endpoint_id == offer.sender_endpoint_id) + { + return TargetedOfferDecision::Refused { + reason: "offer-already-pending".to_string(), + }; + } + if pending.len() >= self.max_pending { + return TargetedOfferDecision::Refused { + reason: "too-many-pending-offers".to_string(), + }; + } pending.insert( transfer_id.clone(), PendingTargetedOfferRecord { @@ -112,6 +142,7 @@ impl TargetedOfferInbox { decision: decision_tx, }, ); + self.cooldown.clear_strikes(&offer.sender_endpoint_id); self.event_hub.emit_endpoint( "targeted_transfer", @@ -153,7 +184,7 @@ impl TargetedOfferInbox { } }; - match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait).await { + match tokio::time::timeout(self.offer_timeout, wait).await { Ok(true) => TargetedOfferDecision::Accepted, Ok(false) => { self.discard(transfer_id).await; @@ -206,10 +237,15 @@ impl TargetedOfferInbox { return Ok(Some(auth)); } - let exists = self.pending.lock().await.contains_key(transfer_id); - if !exists { + let sender_endpoint_id = { + let pending = self.pending.lock().await; + pending + .get(transfer_id) + .map(|entry| entry.offer.sender_endpoint_id.clone()) + }; + let Some(sender_endpoint_id) = sender_endpoint_id else { return Err(RespondError::Unknown); - } + }; let waiter = { let decisions = self.decisions.lock().await; decisions @@ -222,6 +258,7 @@ impl TargetedOfferInbox { if !accepted { let _ = decision_tx.send(Some(false)); self.discard(transfer_id).await; + self.cooldown.record_decline(&sender_endpoint_id); self.settled.lock().await.insert( transfer_id.to_string(), SettledOfferResult::Declined { @@ -258,7 +295,7 @@ impl TargetedOfferInbox { } }; - match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait_auth).await { + match tokio::time::timeout(self.offer_timeout, wait_auth).await { Ok(Ok(auth)) => { self.pending.lock().await.remove(transfer_id); self.auths.lock().await.remove(transfer_id); diff --git a/crates/vnidrop/src/targeted_transfer/protocol.rs b/crates/vnidrop/src/targeted_transfer/protocol.rs index b471c4b..3c48038 100644 --- a/crates/vnidrop/src/targeted_transfer/protocol.rs +++ b/crates/vnidrop/src/targeted_transfer/protocol.rs @@ -89,17 +89,25 @@ impl TargetedTransferProtocol { offer: SubmitTargetedOffer, ) -> TargetedOfferResponse { let expected = experimental_saved_device_capabilities().targeted_transfer_protocol_version; + if self.inbox.cooldown().is_cooling(remote_endpoint_id) { + return TargetedOfferResponse::Refused { + reason: "identity-cooldown".to_string(), + }; + } if offer.protocol_version != expected { + self.inbox.cooldown().record_malformed(remote_endpoint_id); return TargetedOfferResponse::Refused { reason: "protocol-incompatible".to_string(), }; } if offer.receiver_endpoint_id != self.local_endpoint_id { + self.inbox.cooldown().record_malformed(remote_endpoint_id); return TargetedOfferResponse::Refused { reason: "receiver-mismatch".to_string(), }; } if offer.sender_endpoint_id != remote_endpoint_id { + self.inbox.cooldown().record_malformed(remote_endpoint_id); return TargetedOfferResponse::Refused { reason: "sender-mismatch".to_string(), }; @@ -112,6 +120,7 @@ impl TargetedTransferProtocol { || offer.manifest_id.is_empty() || offer.content_hash.is_empty() { + self.inbox.cooldown().record_malformed(remote_endpoint_id); return TargetedOfferResponse::Refused { reason: "manifest-limits".to_string(), }; @@ -120,6 +129,7 @@ impl TargetedTransferProtocol { .limits .validate_metadata_text("transfer name", Some(offer.transfer_name.as_str())) { + self.inbox.cooldown().record_malformed(remote_endpoint_id); return TargetedOfferResponse::Refused { reason: error.to_string(), }; @@ -190,7 +200,8 @@ impl TargetedTransferProtocol { ) .await { - tracing::debug!(%error, "targeted offer relationship proof rejected"); + tracing::debug!(error = %error, "targeted offer relationship proof rejected"); + self.inbox.cooldown().record_malformed(remote_endpoint_id); if matches!(error, VnidropError::ProtocolIncompatible { .. }) { return TargetedOfferResponse::Refused { reason: "protocol-incompatible".to_string(), diff --git a/crates/vnidrop/src/tests.rs b/crates/vnidrop/src/tests.rs index 764810f..83e53a2 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/control_plane.rs"] +mod control_plane_tests; #[path = "tests/device_relationship.rs"] mod device_relationship_tests; #[path = "tests/error.rs"] diff --git a/crates/vnidrop/src/tests/control_plane.rs b/crates/vnidrop/src/tests/control_plane.rs new file mode 100644 index 0000000..e51ea4d --- /dev/null +++ b/crates/vnidrop/src/tests/control_plane.rs @@ -0,0 +1,527 @@ +//! Control-plane hardening: offer bounds, cooldowns, saved-device cap, redaction. + +use std::sync::{Arc, Mutex}; + +use crate::{ + api::{CoreEvent, CoreEventSink, CoreLimits, PendingTargetedOffer}, + control_plane::IdentityCooldown, + event_hub::EventHub, + repository::Repository, + secure_secret::FaultInjectingSecretStore, + targeted_transfer::inbox::{TargetedOfferDecision, TargetedOfferInbox}, + CoreNetworkConfig, DeviceRelationshipState, ShareMetadataInput, ShareSource, SourceKind, + TransferAccessMode, VnidropCore, VnidropError, +}; + +struct RecordingSink { + events: Mutex>, +} + +impl CoreEventSink for RecordingSink { + fn on_event(&self, event: CoreEvent) { + self.events.lock().unwrap().push(event); + } +} + +impl RecordingSink { + fn events(&self) -> Vec { + self.events.lock().unwrap().clone() + } + + fn kinds(&self) -> Vec { + self.events().into_iter().map(|event| event.kind).collect() + } +} + +fn sample_offer(transfer_id: &str, sender: &str) -> PendingTargetedOffer { + PendingTargetedOffer { + transfer_id: transfer_id.to_string(), + sender_endpoint_id: sender.to_string(), + receiver_endpoint_id: "receiver".to_string(), + manifest_id: "manifest".to_string(), + content_hash: "hash".to_string(), + transfer_name: "secret-name.pdf".to_string(), + file_count: 1, + total_size: 12, + protocol_version: 1, + received_at: 1, + } +} + +async fn inbox_with_limits( + max_pending: usize, + cooldown_ms: u64, + strikes: u64, +) -> (TargetedOfferInbox, Arc) { + let temp = tempfile::tempdir().unwrap(); + let repository = Repository::open(temp.path()).await.unwrap(); + let sink = Arc::new(RecordingSink { + events: Mutex::new(Vec::new()), + }); + let hub = Arc::new(EventHub::start(repository, sink.clone(), 64, 100)); + let cooldown = IdentityCooldown::new(cooldown_ms, strikes); + let inbox = TargetedOfferInbox::new(hub, max_pending, cooldown, 5_000); + // Keep temp dir alive for the hub's repository by leaking — tests are short-lived. + std::mem::forget(temp); + (inbox, sink) +} + +#[tokio::test] +async fn one_unresolved_offer_per_sender_and_global_queue_bound() { + let (inbox, sink) = inbox_with_limits(1, 60_000, 5).await; + + let first = sample_offer("t1", "sender-a"); + let submit_first = { + let inbox = inbox.clone(); + tokio::spawn(async move { inbox.submit(first).await }) + }; + // Wait until the prompt is live. + let started = std::time::Instant::now(); + loop { + if !inbox.list().await.is_empty() { + break; + } + assert!(started.elapsed() < std::time::Duration::from_secs(2)); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + assert!(sink.kinds().contains(&"offer-received".to_string())); + + let same_sender = inbox.submit(sample_offer("t2", "sender-a")).await; + assert_eq!( + same_sender, + TargetedOfferDecision::Refused { + reason: "offer-already-pending".to_string() + } + ); + + let other_sender = inbox.submit(sample_offer("t3", "sender-b")).await; + assert_eq!( + other_sender, + TargetedOfferDecision::Refused { + reason: "too-many-pending-offers".to_string() + } + ); + assert_eq!(inbox.list().await.len(), 1); + // Excess rejects never emit a second prompt. + assert_eq!( + sink.kinds() + .into_iter() + .filter(|kind| kind == "offer-received") + .count(), + 1 + ); + + inbox.respond("t1", false).await.unwrap(); + let _ = submit_first.await.unwrap(); +} + +#[tokio::test] +async fn decline_cools_sender_without_affecting_unrelated_devices() { + let (inbox, _sink) = inbox_with_limits(8, 60_000, 5).await; + let offer = sample_offer("decline-1", "noisy"); + let wait = { + let inbox = inbox.clone(); + tokio::spawn(async move { inbox.submit(offer).await }) + }; + let started = std::time::Instant::now(); + loop { + if inbox.get_pending("decline-1").await.is_some() { + break; + } + assert!(started.elapsed() < std::time::Duration::from_secs(2)); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + inbox.respond("decline-1", false).await.unwrap(); + let _ = wait.await.unwrap(); + + let cooled = inbox.submit(sample_offer("decline-2", "noisy")).await; + assert_eq!( + cooled, + TargetedOfferDecision::Refused { + reason: "identity-cooldown".to_string() + } + ); + assert!(inbox.list().await.is_empty()); + + let unrelated = { + let inbox = inbox.clone(); + tokio::spawn(async move { inbox.submit(sample_offer("ok", "friend")).await }) + }; + let started = std::time::Instant::now(); + loop { + if inbox.get_pending("ok").await.is_some() { + break; + } + assert!(started.elapsed() < std::time::Duration::from_secs(2)); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + inbox.respond("ok", false).await.unwrap(); + let _ = unrelated.await.unwrap(); +} + +#[tokio::test] +async fn malformed_strikes_trip_cooldown() { + let cooldown = IdentityCooldown::new(60_000, 2); + assert!(!cooldown.record_malformed("attacker")); + assert!(cooldown.record_malformed("attacker")); + assert!(cooldown.is_cooling("attacker")); + assert!(!cooldown.is_cooling("bystander")); +} + +#[tokio::test] +async fn offer_received_events_redact_endpoint_ids_and_names() { + let (inbox, sink) = inbox_with_limits(4, 60_000, 5).await; + let sender = "abc123endpointid000000000000000000000000000000000000000000000000"; + let wait = { + let inbox = inbox.clone(); + let offer = sample_offer("redact-1", sender); + tokio::spawn(async move { inbox.submit(offer).await }) + }; + let started = std::time::Instant::now(); + loop { + if sink + .kinds() + .into_iter() + .any(|kind| kind == "offer-received") + { + break; + } + assert!(started.elapsed() < std::time::Duration::from_secs(2)); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + let event = sink + .events() + .into_iter() + .find(|event| event.kind == "offer-received") + .expect("offer event"); + assert!(!event.data_json.contains(sender)); + assert!(!event.data_json.contains("secret-name")); + assert!(event.data_json.contains("redacted")); + inbox.respond("redact-1", false).await.unwrap(); + let _ = wait.await.unwrap(); +} + +struct ProtectedNode { + _data_dir: tempfile::TempDir, + _secret_store: Arc, + _limits: CoreLimits, + _network_config: CoreNetworkConfig, + sink: Arc, + core: Option>, +} + +impl ProtectedNode { + fn with_limits(limits: CoreLimits) -> Self { + let data_dir = tempfile::tempdir().unwrap(); + let sink = Arc::new(RecordingSink { + events: Mutex::new(Vec::new()), + }); + let store = Arc::new(FaultInjectingSecretStore::default()); + let network_config = CoreNetworkConfig::default(); + let core = VnidropCore::initialize_with_test_secret_store_limits_and_network( + data_dir.path().to_string_lossy().into_owned(), + sink.clone(), + store.clone(), + limits.clone(), + network_config.clone(), + ) + .expect("protected test core"); + Self { + _data_dir: data_dir, + _secret_store: store, + _limits: limits, + _network_config: network_config, + sink, + core: Some(core), + } + } + + fn core(&self) -> Arc { + self.core.as_ref().expect("core alive").clone() + } +} + +impl Drop for ProtectedNode { + fn drop(&mut self) { + if let Some(core) = self.core.take() { + core.shutdown(); + } + } +} + +fn share_path( + core: &VnidropCore, + source: &std::path::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 = std::time::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() < std::time::Duration::from_secs(15), + "timed out waiting for receiver request" + ); + std::thread::sleep(std::time::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 = std::time::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() < std::time::Duration::from_secs(15), + "eligibility never appeared" + ); + std::thread::sleep(std::time::Duration::from_millis(25)); + } +} + +fn establish_saved(alice: &ProtectedNode, bob: &ProtectedNode, transfer_id: u64) { + complete_transfer(alice, bob, transfer_id); + let bob_id = bob.core().status().endpoint_id.clone(); + assert!(alice + .core() + .request_saved_device_pairing(bob_id.clone()) + .unwrap()); + let started = std::time::Instant::now(); + loop { + let pending = bob + .core() + .list_device_relationships() + .unwrap() + .into_iter() + .find(|entry| { + entry.remote_endpoint_id == alice.core().status().endpoint_id + && entry.state == DeviceRelationshipState::PendingIncoming + }); + if pending.is_some() { + break; + } + assert!( + started.elapsed() < std::time::Duration::from_secs(15), + "pairing prompt never arrived" + ); + std::thread::sleep(std::time::Duration::from_millis(25)); + } + assert!(bob + .core() + .respond_to_device_pairing(alice.core().status().endpoint_id.clone(), true) + .unwrap()); + let started = std::time::Instant::now(); + loop { + if alice.core().list_saved_devices().unwrap().len() == 1 + && bob.core().list_saved_devices().unwrap().len() == 1 + { + break; + } + assert!( + started.elapsed() < std::time::Duration::from_secs(15), + "saved relationship never activated" + ); + std::thread::sleep(std::time::Duration::from_millis(25)); + } +} + +#[test] +fn default_limits_include_saved_device_cap_and_control_plane_timeouts() { + let limits = CoreLimits::default(); + limits.validate().unwrap(); + assert_eq!(limits.max_saved_devices, 256); + assert!(limits.identity_cooldown_ms > 0); + assert!(limits.malformed_strike_limit > 0); + assert!(limits.pairing_timeout_ms > 0); + assert!(limits.offer_timeout_ms > 0); + assert!(limits.connection_timeout_ms > 0); + assert!( + limits.max_pending_offers <= 64, + "pending offers stay tightly bounded" + ); +} + +#[test] +fn saved_device_cap_blocks_only_new_relationships() { + let tight = CoreLimits { + max_saved_devices: 1, + ..CoreLimits::default() + }; + let alice = ProtectedNode::with_limits(tight.clone()); + let bob = ProtectedNode::with_limits(tight.clone()); + let carol = ProtectedNode::with_limits(tight); + + establish_saved(&alice, &bob, 13_001); + assert_eq!(alice.core().list_saved_devices().unwrap().len(), 1); + + // Existing relationship remains usable for targeted transfer. + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("payload.txt"); + std::fs::write(&source_path, b"still works").unwrap(); + let bob_id = bob.core().status().endpoint_id.clone(); + let bob_core = bob.core().clone(); + let accept = std::thread::spawn(move || { + let started = std::time::Instant::now(); + let offer = loop { + if let Some(offer) = bob_core.list_pending_targeted_offers().into_iter().next() { + break offer; + } + assert!( + started.elapsed() < std::time::Duration::from_secs(20), + "offer never arrived for existing saved peer" + ); + std::thread::sleep(std::time::Duration::from_millis(25)); + }; + bob_core + .respond_to_targeted_offer(offer.transfer_id, true) + .unwrap() + }); + alice + .core() + .create_targeted_transfer( + bob_id, + vec![ShareSource { + kind: SourceKind::Path, + value: source_path.to_string_lossy().into_owned(), + display_name: Some("payload.txt".to_string()), + is_directory: false, + }], + Some("payload.txt".to_string()), + ) + .unwrap(); + accept.join().unwrap().unwrap(); + + // New relationship is refused while the cap is full. + complete_transfer(&alice, &carol, 13_002); + let carol_id = carol.core().status().endpoint_id.clone(); + assert!( + !alice.core().request_saved_device_pairing(carol_id).unwrap(), + "cap must block only new relationships" + ); + assert!(alice.core().list_saved_devices().unwrap().len() <= 1); + assert!(carol.core().list_saved_devices().unwrap().is_empty()); +} + +#[test] +fn silent_reject_keeps_invalid_offers_off_the_prompt_surface() { + let alice = ProtectedNode::with_limits(CoreLimits::default()); + let bob = ProtectedNode::with_limits(CoreLimits::default()); + // No Saved relationship — create must fail before any receiver prompt. + let bob_id = bob.core().status().endpoint_id.clone(); + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("payload.txt"); + std::fs::write(&source_path, b"nope").unwrap(); + let err = alice + .core() + .create_targeted_transfer( + bob_id, + vec![ShareSource { + kind: SourceKind::Path, + value: source_path.to_string_lossy().into_owned(), + display_name: Some("payload.txt".to_string()), + is_directory: false, + }], + Some("payload.txt".to_string()), + ) + .unwrap_err(); + assert!(matches!(err, VnidropError::Permission { .. })); + assert!(bob.core().list_pending_targeted_offers().is_empty()); + assert!( + !bob.sink + .kinds() + .into_iter() + .any(|kind| kind == "offer-received"), + "silent reject must not emit offer prompts" + ); +} + +#[test] +fn blocked_peer_cannot_create_pairing_prompt() { + let alice = ProtectedNode::with_limits(CoreLimits::default()); + let bob = ProtectedNode::with_limits(CoreLimits::default()); + complete_transfer(&alice, &bob, 13_010); + let alice_id = alice.core().status().endpoint_id.clone(); + bob.core().block_device(alice_id.clone()).unwrap(); + assert!( + !alice + .core() + .request_saved_device_pairing(bob.core().status().endpoint_id.clone()) + .unwrap() + || bob + .core() + .list_device_relationships() + .unwrap() + .iter() + .all(|entry| entry.state != DeviceRelationshipState::PendingIncoming) + ); + // Stronger: after block, bob must not surface an incoming pairing prompt. + std::thread::sleep(std::time::Duration::from_millis(200)); + assert!(bob + .core() + .list_device_relationships() + .unwrap() + .into_iter() + .filter(|entry| entry.remote_endpoint_id == alice_id) + .all(|entry| entry.state != DeviceRelationshipState::PendingIncoming)); +} + +#[test] +fn ticket_errors_redact_raw_ticket_blobs() { + let err = VnidropError::ticket(anyhow::anyhow!( + "bad ticket vnd1:abcDEF1234567890 and endpoint 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + )); + let rendered = err.to_string(); + assert!(!rendered.contains("vnd1:abcDEF")); + assert!(!rendered.contains("0123456789abcdef0123456789abcdef")); + assert!(rendered.contains("[redacted]")); +} diff --git a/crates/vnidrop/src/tests/limits.rs b/crates/vnidrop/src/tests/limits.rs index e462d0c..e227896 100644 --- a/crates/vnidrop/src/tests/limits.rs +++ b/crates/vnidrop/src/tests/limits.rs @@ -11,6 +11,8 @@ fn default_limits_bound_ticket_and_approval_pressure() { assert!(limits.max_ticket_bytes <= 256 * 1024); assert!(limits.max_pending_approvals <= 64); assert!(limits.max_total_bytes <= 256 * 1024 * 1024 * 1024); + assert_eq!(limits.max_saved_devices, 256); + assert!(limits.max_pending_offers <= 16); } #[test] @@ -21,3 +23,12 @@ fn zero_limit_is_rejected() { }; assert!(limits.validate().is_err()); } + +#[test] +fn zero_saved_device_limit_is_rejected() { + let limits = CoreLimits { + max_saved_devices: 0, + ..CoreLimits::default() + }; + assert!(limits.validate().is_err()); +} diff --git a/crates/vnidrop/tests/transfer.rs b/crates/vnidrop/tests/transfer.rs index 64d909a..a707468 100644 --- a/crates/vnidrop/tests/transfer.rs +++ b/crates/vnidrop/tests/transfer.rs @@ -98,9 +98,11 @@ fn transfers_file_between_two_cores() { let completed = wait_for_sender_transfer_event(&sender, share.transfer_id, "completed"); assert!(completed.data_json.contains("\"connection_id\":")); assert!(completed.data_json.contains("\"request_id\":")); - assert!(completed + // Production events redact endpoint ids; typed APIs remain the source of truth. + assert!(!completed .data_json .contains(receiver.core.status().endpoint_id.as_str())); + assert!(completed.data_json.contains("redacted")); receiver.core.delete_receive_history().unwrap(); assert_eq!(receiver.core.list_received_artifacts().unwrap(), artifacts);