diff --git a/src/dht_network_manager.rs b/src/dht_network_manager.rs index 12c42dd1..045aa7fc 100644 --- a/src/dht_network_manager.rs +++ b/src/dht_network_manager.rs @@ -27,6 +27,12 @@ use crate::{ dht::{AdmissionResult, DhtCoreEngine, DhtKey, Key, RoutingTableEvent}, error::{DhtError, IdentityError, NetworkError}, network::{NodeConfig, NodeMode}, + rate_limit::{Engine, SharedEngine}, + reachability::canary::{ + RELAY_CANARY_HANDLER_TIMEOUT, RELAY_CANARY_PROTOCOL, RELAY_CANARY_WIRE_TOPIC, + RelayCanaryProbeResult, RelayCanaryRequest, RelayCanaryResponse, + answer_relay_canary_request, relay_canary_rate_limit_config, validate_relay_canary_request, + }, security::canonicalize_ip, self_address::build_self_address_set, }; @@ -728,6 +734,12 @@ pub struct DhtNetworkManager { /// Self-lookups and foreground/payment/user lookup calls do not use this /// semaphore. bucket_refresh_lookup_semaphore: Arc, + /// Per-source relay canary rate limiter. + /// + /// Answering a canary request triggers a cold relay dial, so each + /// authenticated source is throttled (see [`relay_canary_rate_limit_config`]) + /// to stop a peer using this node as a reflection/amplification dialer. + relay_canary_rate_limiter: SharedEngine, /// Shutdown token for background tasks shutdown: CancellationToken, /// Handle for the network event handler task @@ -1633,6 +1645,7 @@ impl DhtNetworkManager { identity_failure_cache: Arc::new(IdentityFailureCache::new()), pending_peer_dials: Arc::new(DashMap::new()), lookup_failures: Arc::new(LookupFailureCoordinator::new()), + relay_canary_rate_limiter: Arc::new(Engine::new(relay_canary_rate_limit_config())), }) } @@ -4237,6 +4250,116 @@ impl DhtNetworkManager { self.send_dht_request(peer_id, operation, None).await } + /// Ask a peer to cold-dial a freshly acquired relay address. + /// + /// This intentionally uses the generic request/response transport rather + /// than adding a DHT operation: the canary is a reachability proof for the + /// acquisition driver, not routing-table state. + pub(crate) async fn send_relay_canary_request( + &self, + peer_id: &PeerId, + candidates: &[(MultiAddr, AddressType)], + request: RelayCanaryRequest, + timeout: Duration, + ) -> Result { + self.ensure_peer_channel(peer_id, candidates).await?; + + let request_bytes = postcard::to_stdvec(&request) + .map_err(|e| P2PError::Serialization(e.to_string().into()))?; + let response = self + .transport + .send_request(peer_id, RELAY_CANARY_PROTOCOL, request_bytes, timeout) + .await?; + postcard::from_bytes(&response.data) + .map_err(|e| P2PError::Serialization(e.to_string().into())) + } + + async fn handle_relay_canary_message(&self, source_peer: PeerId, data: Vec) -> Result<()> { + if data.len() > MAX_MESSAGE_SIZE { + debug!( + "Ignoring oversized relay canary message from {source_peer}: {} bytes (max: {MAX_MESSAGE_SIZE})", + data.len() + ); + return Ok(()); + } + + let Some((message_id, is_response, payload)) = + crate::transport_handle::TransportHandle::parse_request_envelope(&data) + else { + debug!( + peer = %source_peer.to_hex(), + "Ignoring malformed relay canary request envelope" + ); + return Ok(()); + }; + + if is_response { + trace!( + message_id = %message_id, + peer = %source_peer.to_hex(), + "Ignoring relay canary response in request handler" + ); + return Ok(()); + } + + let request: RelayCanaryRequest = match postcard::from_bytes(&payload) { + Ok(request) => request, + Err(e) => { + debug!( + peer = %source_peer.to_hex(), + error = %e, + "Ignoring malformed relay canary request payload" + ); + return Ok(()); + } + }; + if let Err(reason) = validate_relay_canary_request(&source_peer, &request) { + debug!( + peer = %source_peer.to_hex(), + reason = %reason.summary(), + "Rejecting relay canary request" + ); + return Ok(()); + } + // Answering this request makes us cold-dial `relay_addr`, so throttle + // each authenticated source to stop a peer using this node as a + // reflection/amplification dialer toward an address of its choosing. + if !self.relay_canary_rate_limiter.try_consume_key(&source_peer) { + debug!( + peer = %source_peer.to_hex(), + "Throttling relay canary request from source" + ); + let response = RelayCanaryResponse { + result: RelayCanaryProbeResult::WitnessRateLimited, + }; + return self + .send_relay_canary_response(&source_peer, &message_id, response) + .await; + } + + let response = answer_relay_canary_request(self.transport.as_ref(), request).await; + self.send_relay_canary_response(&source_peer, &message_id, response) + .await + } + + async fn send_relay_canary_response( + &self, + source_peer: &PeerId, + message_id: &str, + response: RelayCanaryResponse, + ) -> Result<()> { + let response_bytes = postcard::to_stdvec(&response) + .map_err(|e| P2PError::Serialization(e.to_string().into()))?; + self.transport + .send_response( + source_peer, + RELAY_CANARY_PROTOCOL, + message_id, + response_bytes, + ) + .await + } + /// Handle DHT response message /// /// Delivers the response via oneshot channel to the waiting request coroutine. @@ -4751,6 +4874,48 @@ impl DhtNetworkManager { } // _permit dropped here, releasing semaphore slot }); + } else if topic == RELAY_CANARY_WIRE_TOPIC { + // Relay canary requests must be authenticated so the + // response can be routed back through request/response. + let Some(source_peer) = source else { + warn!("Ignoring unsigned relay canary request"); + continue; + }; + let manager_clone = Arc::clone(&self_arc); + let semaphore = Arc::clone(&self_arc.message_handler_semaphore); + tokio::spawn(async move { + let _permit = match semaphore.acquire().await { + Ok(permit) => permit, + Err(_) => { + warn!("Message handler semaphore closed"); + return; + } + }; + + match tokio::time::timeout( + RELAY_CANARY_HANDLER_TIMEOUT, + manager_clone + .handle_relay_canary_message(source_peer, data), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => { + warn!( + peer = %source_peer.to_hex(), + error = %e, + "Failed to handle relay canary request" + ); + } + Err(_) => { + warn!( + timeout = ?RELAY_CANARY_HANDLER_TIMEOUT, + peer = %source_peer.to_hex(), + "Relay canary request handler timed out" + ); + } + } + }); } } }, diff --git a/src/network.rs b/src/network.rs index 27597d75..c5135a33 100644 --- a/src/network.rs +++ b/src/network.rs @@ -1200,49 +1200,16 @@ impl P2PNode { info!("client mode — skipping relay acquisition driver"); } - // Spawn background task to forward peer address updates to the DHT. + // Drain peer-advertised ADD_ADDRESS updates from the transport layer. // - // Two event streams are bridged from the transport layer onto DHT - // routing-table mutations: - // - // - **Relay established**: when THIS node sets up a MASQUE relay, - // perform a DHT self-lookup so the transport's re-advertisement - // loop can ADD_ADDRESS the new relay address to the K closest - // peers — propagating it beyond peers we already happen to be - // connected to. - // - **Peer address update**: when a connected peer advertises a new - // reachable address via ADD_ADDRESS (typically its relay), update - // the DHT routing table so future lookups return that address. - // - // Both are handled in a `tokio::select!` against the receiver - // futures so updates propagate immediately. The previous - // implementation polled both queues on a 1-second interval, which - // opened a race window in which a freshly-established relay was - // invisible to outbound DHT queries until the next tick — causing - // the first peers to dial direct (and fail) before learning about - // the relay. - // - // **Slow work isolation**: the relay-propagation path runs an - // iterative DHT lookup (`find_closest_nodes_network`) which can - // take many seconds. Doing it inline in the select loop would - // starve the peer-address-update branch and back up the bounded - // forwarder mpsc into drop territory. Instead, the lookup + - // publish is detached into its own task per relay event, so the - // select loop keeps polling both branches. - // DHT_BRIDGE: forward peer-advertised address updates from the - // transport layer onto DHT routing table mutations. When a connected - // peer's ADD_ADDRESS notification carries a different IP than the - // connection's source (i.e., the peer is behind a relay or has - // migrated), merge the advertised address into the peer's DHT entry. - // - // This node's OWN relay state changes are NOT handled here — the - // relay acquisition driver (see `reachability::driver`) owns them - // directly, so the "relay established" branch no longer belongs to - // the bridge. The driver knows the full typed address set for the - // self-record; the bridge did not. + // Relay-looking ADD_ADDRESS frames are intentionally not merged into + // DHT records. Relay reachability must arrive through the sequenced + // self-record path owned by `reachability::driver`, after the relay + // canary quorum has passed. Accepting transport hints here would let + // an unverified relay allocation leak into routing-table gossip before + // the canary gate has made a verdict. { let transport = Arc::clone(&self.transport); - let dht = self.adaptive_dht.dht_manager().clone(); let shutdown = self.shutdown.clone(); tokio::spawn(async move { loop { @@ -1255,10 +1222,10 @@ impl P2PNode { saorsa_transport::shared::normalize_socket_addr(peer_addr); let normalized_adv = saorsa_transport::shared::normalize_socket_addr(advertised_addr); - // Only update DHT when the advertised IP differs - // from the peer's connection IP. Same-IP updates - // are just different NATted ports (useless for - // symmetric NAT); different-IP means a relay. + // Same-IP updates are just different NATted ports, + // which are useless for symmetric NAT. Different + // IPs are relay-like hints and must not bypass the + // canary-gated DHT publish path. if normalized_peer.ip() == normalized_adv.ip() { debug!( "DHT_BRIDGE: dropping same-IP update peer={} addr={}", @@ -1267,31 +1234,11 @@ impl P2PNode { ); continue; } - info!( - "DHT_BRIDGE: processing relay update peer={} addr={}", + debug!( + "DHT_BRIDGE: ignoring relay ADD_ADDRESS update pending sequenced canary publish peer={} addr={}", normalized_peer, normalized_adv ); - // Look up peer ID by address (tries both IPv4 and - // IPv4-mapped IPv6 forms via dual_stack_alternate). - // For symmetric NAT, this may fail because the - // connection's channel key uses a different NATted port. - if let Some(peer_id) = transport.peer_id_for_addr(&normalized_peer).await { - let multi_addr = MultiAddr::quic(normalized_adv); - info!( - "Updating DHT: peer {} relay address {} (connection was {})", - peer_id, advertised_addr, peer_addr - ); - if !dht - .touch_legacy_relay_hint_if_unsequenced(&peer_id, &multi_addr) - .await - { - debug!( - "DHT_BRIDGE: ignored legacy relay hint for sequenced peer {} addr {}", - peer_id, advertised_addr - ); - } - } } } } diff --git a/src/reachability/acquisition.rs b/src/reachability/acquisition.rs index 7322adca..3c9ed774 100644 --- a/src/reachability/acquisition.rs +++ b/src/reachability/acquisition.rs @@ -15,10 +15,9 @@ //! //! Every non-client node calls [`RelayAcquisition::acquire`] after bootstrap //! to establish a relay from a close-group peer. The walker is unaware of -//! whether the local node is public or private — if a candidate's Direct -//! address is unreachable (private peer), the QUIC dial fails and the walk -//! advances to the next-closest peer. "Is this candidate public?" is -//! inferred ambiently from the dial attempt. +//! whether the accepted relay is externally useful for third parties; it +//! only establishes the MASQUE session. The reachability driver runs relay +//! canaries before publishing the allocated address. //! //! 1. The caller supplies a pre-filtered list of [`RelayCandidate`]s sorted //! by XOR distance (closest first). Filtering — selecting peers whose diff --git a/src/reachability/canary.rs b/src/reachability/canary.rs new file mode 100644 index 00000000..8a443c85 --- /dev/null +++ b/src/reachability/canary.rs @@ -0,0 +1,894 @@ +// Copyright 2024 Saorsa Labs Limited +// +// This software is dual-licensed under: +// - GNU Affero General Public License v3.0 or later (AGPL-3.0-or-later) +// - Commercial License +// +// For AGPL-3.0 license, see LICENSE-AGPL-3.0 +// For commercial licensing, contact: david@saorsalabs.com +// +// Unless required by applicable law or agreed to in writing, software +// distributed under these licenses is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +//! Third-party relay canary probes. +//! +//! A relay acquisition is not publishable just because the local node +//! established a MASQUE session to a candidate relayer. Before the driver +//! writes the relay-allocated address into the DHT, it asks randomized +//! non-close peers to cold-dial that address and confirm that the +//! authenticated identity on the far end is this node. + +use std::collections::HashSet; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::Arc; +use std::time::Duration; + +use futures::stream::{FuturesUnordered, StreamExt}; +use rand::{Rng, seq::SliceRandom}; +use serde::{Deserialize, Serialize}; +use tracing::{debug, info, warn}; + +use crate::address::is_lan_ip; +use crate::dht::AddressType; +use crate::dht_network_manager::{DHTNode, DhtNetworkManager, IDENTITY_EXCHANGE_TIMEOUT}; +use crate::error::P2PError; +use crate::rate_limit::EngineConfig; +use crate::security::canonicalize_ip; +use crate::transport_handle::TransportHandle; +use crate::{MultiAddr, PeerId}; + +/// Request/response protocol name used with `TransportHandle::send_request`. +pub(crate) const RELAY_CANARY_PROTOCOL: &str = "relay-canary-v1"; + +/// Wire topic emitted by the request/response wrapper for canary requests. +pub(crate) const RELAY_CANARY_WIRE_TOPIC: &str = "/rr/relay-canary-v1"; + +/// Number of independent non-close witnesses to ask for a relay proof. +const RELAY_CANARY_WITNESS_TARGET: usize = 3; + +/// Minimum successful witness probes needed before a relay is publishable. +const RELAY_CANARY_REQUIRED_SUCCESSES: usize = 2; + +/// Witness-side handler budget for answering one relay canary request. +/// +/// The connect and identity budgets below fit inside this cap, with a small +/// margin for serialization and sending the response before the requester +/// gives up. +pub(crate) const RELAY_CANARY_HANDLER_TIMEOUT: Duration = Duration::from_secs(11); + +/// End-to-end budget for asking one witness to dial the proposed relay. +/// +/// The witness-side DHT handler has a smaller cap. Keep the requester +/// budget above that so slow-but-valid witness dials are not discarded just +/// before the handler can reply. +const RELAY_CANARY_REQUEST_TIMEOUT: Duration = Duration::from_secs(12); + +/// Cold-dial connection budget spent by a witness when probing a relay. +/// +/// A relay that cannot establish a transport connection within this window is +/// a failed probe, not an ineligible witness. Keeping this below the handler +/// budget leaves room for the identity check and response. +const RELAY_CANARY_CONNECT_TIMEOUT: Duration = Duration::from_secs(4); + +/// Cold-dial identity budget spent by a witness when probing a relay address. +const RELAY_CANARY_DIAL_TIMEOUT: Duration = IDENTITY_EXCHANGE_TIMEOUT; + +/// Sliding window for per-source relay canary rate limiting. +const RELAY_CANARY_RATE_WINDOW: Duration = Duration::from_secs(10); + +/// Maximum canary-triggered cold dials a single source may request per window. +const RELAY_CANARY_RATE_MAX_PER_WINDOW: u32 = 1; + +/// Per-source throttle applied to inbound relay canary requests. +/// +/// Answering a canary request makes this node cold-dial an arbitrary relay +/// address, so each authenticated source is limited to one dial per window to +/// stop a peer using this node as a reflection/amplification dialer. A +/// legitimate source asks any given witness at most once per acquisition cycle +/// (>= the driver backoff), so this never throttles honest traffic. +pub(crate) fn relay_canary_rate_limit_config() -> EngineConfig { + EngineConfig { + window: RELAY_CANARY_RATE_WINDOW, + max_requests: RELAY_CANARY_RATE_MAX_PER_WINDOW, + burst_size: RELAY_CANARY_RATE_MAX_PER_WINDOW, + } +} + +/// Socket port zero is not a routable service endpoint. +const UNSPECIFIED_PORT: u16 = 0; + +/// Request sent to a witness asking it to verify a proposed relay address. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct RelayCanaryRequest { + pub(crate) target_peer_id: PeerId, + pub(crate) relayer_peer_id: PeerId, + pub(crate) relay_addr: SocketAddr, +} + +impl RelayCanaryRequest { + fn new(target_peer_id: PeerId, relayer_peer_id: PeerId, relay_addr: SocketAddr) -> Self { + Self { + target_peer_id, + relayer_peer_id, + relay_addr, + } + } +} + +/// Witness response after attempting the cold relay dial. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct RelayCanaryResponse { + pub(crate) result: RelayCanaryProbeResult, +} + +/// Result of one witness's relay probe. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum RelayCanaryProbeResult { + Success, + DialFailed, + IdentityExchangeFailed, + IdentityMismatch, + WitnessRateLimited, +} + +impl RelayCanaryProbeResult { + fn disposition(&self) -> RelayCanaryProbeDisposition { + match self { + Self::Success => RelayCanaryProbeDisposition::Success, + Self::WitnessRateLimited => RelayCanaryProbeDisposition::Ineligible, + Self::DialFailed | Self::IdentityExchangeFailed | Self::IdentityMismatch => { + RelayCanaryProbeDisposition::Failure + } + } + } + + fn summary(&self) -> String { + match self { + Self::Success => "success".to_string(), + Self::DialFailed => "dial failed".to_string(), + Self::IdentityExchangeFailed => "identity exchange failed".to_string(), + Self::IdentityMismatch => "identity mismatch".to_string(), + Self::WitnessRateLimited => "witness rate-limited source".to_string(), + } + } +} + +/// Reject reason for a malformed or unauthorized canary request. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum RelayCanaryRequestRejection { + SourceMismatch { + source_peer_id: PeerId, + target_peer_id: PeerId, + }, + SelfRelayer { + peer_id: PeerId, + }, + UnspecifiedPort, + UnspecifiedIp, + LocalScopeIp(IpAddr), + MulticastIp(IpAddr), + BroadcastIp(Ipv4Addr), +} + +impl RelayCanaryRequestRejection { + pub(crate) fn summary(&self) -> String { + match self { + Self::SourceMismatch { + source_peer_id, + target_peer_id, + } => format!( + "source {} does not match target {}", + source_peer_id.to_hex(), + target_peer_id.to_hex() + ), + Self::SelfRelayer { peer_id } => { + format!("target {} also claimed to be relayer", peer_id.to_hex()) + } + Self::UnspecifiedPort => "relay address has port 0".to_string(), + Self::UnspecifiedIp => "relay address has unspecified IP".to_string(), + Self::LocalScopeIp(ip) => format!("relay address uses local-scope IP {ip}"), + Self::MulticastIp(ip) => format!("relay address uses multicast IP {ip}"), + Self::BroadcastIp(ip) => format!("relay address uses broadcast IP {ip}"), + } + } +} + +/// Validate a witness can safely act on a canary request. +pub(crate) fn validate_relay_canary_request( + source_peer_id: &PeerId, + request: &RelayCanaryRequest, +) -> std::result::Result<(), RelayCanaryRequestRejection> { + if request.target_peer_id != *source_peer_id { + return Err(RelayCanaryRequestRejection::SourceMismatch { + source_peer_id: *source_peer_id, + target_peer_id: request.target_peer_id, + }); + } + if request.relayer_peer_id == request.target_peer_id { + return Err(RelayCanaryRequestRejection::SelfRelayer { + peer_id: request.target_peer_id, + }); + } + + validate_relay_canary_address(request.relay_addr) +} + +fn validate_relay_canary_address( + relay_addr: SocketAddr, +) -> std::result::Result<(), RelayCanaryRequestRejection> { + if relay_addr.port() == UNSPECIFIED_PORT { + return Err(RelayCanaryRequestRejection::UnspecifiedPort); + } + + let ip = relay_addr.ip(); + if ip.is_unspecified() { + return Err(RelayCanaryRequestRejection::UnspecifiedIp); + } + if is_lan_ip(ip) { + return Err(RelayCanaryRequestRejection::LocalScopeIp(ip)); + } + if ip.is_multicast() { + return Err(RelayCanaryRequestRejection::MulticastIp(ip)); + } + if let IpAddr::V4(ipv4) = ip + && ipv4 == Ipv4Addr::BROADCAST + { + return Err(RelayCanaryRequestRejection::BroadcastIp(ipv4)); + } + + Ok(()) +} + +/// Aggregate decision for a just-acquired relay. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum RelayCanaryVerdict { + Verified { successes: usize, attempts: usize }, + Rejected { successes: usize, attempts: usize }, + InsufficientWitnesses { available: usize, required: usize }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RelayCanaryProbeDisposition { + Success, + Failure, + Ineligible, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RelayCanaryProgressDecision { + Continue, + Verified { successes: usize, attempts: usize }, + Rejected { successes: usize, attempts: usize }, +} + +#[derive(Debug, Clone)] +struct RelayCanaryProgress { + total: usize, + responses: usize, + eligible_attempts: usize, + successes: usize, + ineligible: usize, +} + +impl RelayCanaryProgress { + fn new(total: usize) -> Self { + Self { + total, + responses: 0, + eligible_attempts: 0, + successes: 0, + ineligible: 0, + } + } + + fn record(&mut self, disposition: RelayCanaryProbeDisposition) -> RelayCanaryProgressDecision { + self.responses += 1; + match disposition { + RelayCanaryProbeDisposition::Success => { + self.eligible_attempts += 1; + self.successes += 1; + } + RelayCanaryProbeDisposition::Failure => { + self.eligible_attempts += 1; + } + RelayCanaryProbeDisposition::Ineligible => { + self.ineligible += 1; + } + } + + if self.successes >= RELAY_CANARY_REQUIRED_SUCCESSES { + return RelayCanaryProgressDecision::Verified { + successes: self.successes, + attempts: self.eligible_attempts, + }; + } + + let remaining = self.total.saturating_sub(self.responses); + if self.successes + remaining < RELAY_CANARY_REQUIRED_SUCCESSES + && self.eligible_attempts >= RELAY_CANARY_REQUIRED_SUCCESSES + { + return RelayCanaryProgressDecision::Rejected { + successes: self.successes, + attempts: self.eligible_attempts, + }; + } + + RelayCanaryProgressDecision::Continue + } + + fn final_verdict(&self) -> RelayCanaryVerdict { + if self.eligible_attempts < RELAY_CANARY_REQUIRED_SUCCESSES { + RelayCanaryVerdict::InsufficientWitnesses { + available: self.eligible_attempts, + required: RELAY_CANARY_REQUIRED_SUCCESSES, + } + } else { + RelayCanaryVerdict::Rejected { + successes: self.successes, + attempts: self.eligible_attempts, + } + } + } +} + +#[derive(Debug, Clone)] +struct RelayCanaryWitness { + peer_id: PeerId, + typed_addresses: Vec<(MultiAddr, AddressType)>, +} + +#[derive(Debug, Clone)] +struct RelayCanaryProbeReport { + witness: PeerId, + disposition: RelayCanaryProbeDisposition, + detail: String, +} + +/// Verify that `relay_addr` is externally dialable before publication. +pub(crate) async fn verify_relay_with_canaries( + dht: &Arc, + relayer: PeerId, + relay_addr: SocketAddr, +) -> RelayCanaryVerdict { + let target_peer_id = *dht.peer_id(); + let request = RelayCanaryRequest::new(target_peer_id, relayer, relay_addr); + if let Err(reason) = validate_relay_canary_request(&target_peer_id, &request) { + warn!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + reason = %reason.summary(), + "relay canary: refusing invalid relay address" + ); + return RelayCanaryVerdict::Rejected { + successes: 0, + attempts: 0, + }; + } + + let target_key = *target_peer_id.to_bytes(); + let close_group_ids: HashSet = dht + .find_closest_nodes_local(&target_key, dht.k_value()) + .await + .into_iter() + .map(|node| node.peer_id) + .collect(); + let routing_table = dht.routing_table_peers().await; + let routing_table_size = routing_table.len(); + let witnesses = select_relay_canary_witnesses( + routing_table, + &close_group_ids, + &target_peer_id, + &relayer, + relay_addr.ip(), + RELAY_CANARY_WITNESS_TARGET, + &mut rand::thread_rng(), + ); + + if witnesses.len() < RELAY_CANARY_REQUIRED_SUCCESSES { + warn!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + available = witnesses.len(), + required = RELAY_CANARY_REQUIRED_SUCCESSES, + close_group_excluded = close_group_ids.len(), + routing_table_size, + "relay canary: insufficient random non-close witnesses, refusing to publish relay" + ); + return RelayCanaryVerdict::InsufficientWitnesses { + available: witnesses.len(), + required: RELAY_CANARY_REQUIRED_SUCCESSES, + }; + } + + debug!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + available_witnesses = witnesses.len(), + routing_table_size, + close_group_excluded = close_group_ids.len(), + "relay canary: probing random non-close witnesses" + ); + + let mut progress = RelayCanaryProgress::new(witnesses.len()); + let mut probes = FuturesUnordered::new(); + for witness in witnesses { + let dht = Arc::clone(dht); + let request = RelayCanaryRequest::new(target_peer_id, relayer, relay_addr); + probes.push(async move { request_relay_canary(dht, witness, request).await }); + } + + while let Some(report) = probes.next().await { + let decision = progress.record(report.disposition); + if report.disposition == RelayCanaryProbeDisposition::Success { + debug!( + witness = %report.witness.to_hex(), + successes = progress.successes, + eligible_attempts = progress.eligible_attempts, + responses = progress.responses, + "relay canary: witness confirmed relay" + ); + } else if report.disposition == RelayCanaryProbeDisposition::Ineligible { + debug!( + witness = %report.witness.to_hex(), + detail = %report.detail, + ineligible = progress.ineligible, + eligible_attempts = progress.eligible_attempts, + responses = progress.responses, + "relay canary: witness could not evaluate relay" + ); + } else { + debug!( + witness = %report.witness.to_hex(), + detail = %report.detail, + successes = progress.successes, + eligible_attempts = progress.eligible_attempts, + responses = progress.responses, + "relay canary: witness failed relay probe" + ); + } + + match decision { + RelayCanaryProgressDecision::Verified { + successes, + attempts, + } => { + info!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + successes, + attempts, + responses = progress.responses, + ineligible = progress.ineligible, + available_witnesses = progress.total, + "relay canary: quorum verified relay" + ); + return RelayCanaryVerdict::Verified { + successes, + attempts, + }; + } + RelayCanaryProgressDecision::Rejected { + successes, + attempts, + } => { + warn!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + successes, + attempts, + responses = progress.responses, + total = progress.total, + ineligible = progress.ineligible, + "relay canary: quorum failed relay" + ); + return RelayCanaryVerdict::Rejected { + successes, + attempts, + }; + } + RelayCanaryProgressDecision::Continue => {} + } + } + + progress.final_verdict() +} + +/// Probe `request.relay_addr` from this witness node and return the result. +pub(crate) async fn answer_relay_canary_request( + transport: &TransportHandle, + request: RelayCanaryRequest, +) -> RelayCanaryResponse { + let relay_address = MultiAddr::quic(request.relay_addr); + let dial = tokio::time::timeout( + RELAY_CANARY_CONNECT_TIMEOUT, + // The address type is an informational hint for logging/classification; + // correctness comes from dialing the allocated socket and checking the + // authenticated target identity below. + transport.connect_peer_typed(&relay_address, AddressType::Relay), + ) + .await; + + let result = match dial { + Ok(Ok(channel_id)) => { + let identity = transport + .wait_for_peer_identity(&channel_id, RELAY_CANARY_DIAL_TIMEOUT) + .await; + let result = match identity { + Ok(authenticated) if authenticated == request.target_peer_id => { + RelayCanaryProbeResult::Success + } + Ok(authenticated) => { + debug!( + expected = %request.target_peer_id.to_hex(), + actual = %authenticated.to_hex(), + relay = %request.relay_addr, + "relay canary witness: identity mismatch" + ); + RelayCanaryProbeResult::IdentityMismatch + } + Err(e) => { + debug!( + relay = %request.relay_addr, + error = %e, + "relay canary witness: identity exchange failed" + ); + RelayCanaryProbeResult::IdentityExchangeFailed + } + }; + transport.disconnect_channel(&channel_id).await; + result + } + Ok(Err(e)) => { + debug!( + relay = %request.relay_addr, + error = %e, + "relay canary witness: dial failed" + ); + RelayCanaryProbeResult::DialFailed + } + Err(_) => { + debug!( + relay = %request.relay_addr, + timeout = ?RELAY_CANARY_CONNECT_TIMEOUT, + "relay canary witness: dial timed out" + ); + RelayCanaryProbeResult::DialFailed + } + }; + + RelayCanaryResponse { result } +} + +fn select_relay_canary_witnesses( + mut candidates: Vec, + close_group_ids: &HashSet, + target_peer_id: &PeerId, + relayer: &PeerId, + relay_ip: IpAddr, + count: usize, + rng: &mut R, +) -> Vec { + let mut witnesses = Vec::with_capacity(count); + let mut seen_ips = HashSet::new(); + let relay_ip = canonicalize_ip(relay_ip); + + candidates.shuffle(rng); + for node in candidates { + if node.peer_id == *target_peer_id + || node.peer_id == *relayer + || close_group_ids.contains(&node.peer_id) + { + continue; + } + + let typed_addresses = node.typed_addresses(); + if !typed_addresses + .iter() + .any(|(addr, _)| addr.dialable_socket_addr().is_some()) + { + continue; + } + + let Some(ip) = first_dialable_ip(&typed_addresses) else { + continue; + }; + let ip = canonicalize_ip(ip); + if ip == relay_ip || !seen_ips.insert(ip) { + continue; + } + + witnesses.push(RelayCanaryWitness { + peer_id: node.peer_id, + typed_addresses, + }); + if witnesses.len() == count { + break; + } + } + + witnesses +} + +fn first_dialable_ip(typed_addresses: &[(MultiAddr, AddressType)]) -> Option { + typed_addresses + .iter() + .filter_map(|(addr, _)| addr.dialable_socket_addr().map(|sa| sa.ip())) + .next() +} + +async fn request_relay_canary( + dht: Arc, + witness: RelayCanaryWitness, + request: RelayCanaryRequest, +) -> RelayCanaryProbeReport { + let witness_peer_id = witness.peer_id; + match dht + .send_relay_canary_request( + &witness_peer_id, + &witness.typed_addresses, + request, + RELAY_CANARY_REQUEST_TIMEOUT, + ) + .await + { + Ok(response) => RelayCanaryProbeReport { + witness: witness_peer_id, + disposition: response.result.disposition(), + detail: response.result.summary(), + }, + Err(e) => RelayCanaryProbeReport { + witness: witness_peer_id, + disposition: canary_request_error_disposition(&e), + detail: e.to_string(), + }, + } +} + +fn canary_request_error_disposition(_error: &P2PError) -> RelayCanaryProbeDisposition { + // Request-level failures do not prove the relay is bad: in mixed-version + // networks the witness may simply not implement the relay canary protocol. + // Only typed canary responses count as eligible probe attempts. + RelayCanaryProbeDisposition::Ineligible +} + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + use std::net::{Ipv4Addr, SocketAddr}; + + use rand::SeedableRng; + use rand::rngs::StdRng; + + use super::*; + use crate::error::NetworkError; + use crate::rate_limit::Engine; + + const TARGET_SEED: u8 = 1; + const RELAYER_SEED: u8 = 2; + const CLOSE_GROUP_SEED: u8 = 3; + const FIRST_WITNESS_SEED: u8 = 4; + const SECOND_WITNESS_SEED: u8 = 5; + const THIRD_WITNESS_SEED: u8 = 6; + const DUPLICATE_IP_WITNESS_SEED: u8 = 7; + const RELAY_IP_WITNESS_SEED: u8 = 8; + const TEST_PORT: u16 = 9000; + const TEST_RNG_SEED: u64 = 42; + + fn peer_id(seed: u8) -> PeerId { + PeerId::from_bytes([seed; 32]) + } + + fn node(seed: u8, ip: Ipv4Addr) -> DHTNode { + DHTNode { + peer_id: peer_id(seed), + addresses: vec![MultiAddr::from_ipv4(ip, TEST_PORT + u16::from(seed))], + address_types: vec![AddressType::Direct], + distance: None, + reliability: 1.0, + } + } + + #[test] + fn witness_selection_uses_random_non_close_independent_sources() { + let target = peer_id(TARGET_SEED); + let relayer = peer_id(RELAYER_SEED); + let relay_ip = Ipv4Addr::new(203, 0, 113, 2); + let close_group_ids = HashSet::from([peer_id(CLOSE_GROUP_SEED)]); + let candidates = vec![ + node(TARGET_SEED, Ipv4Addr::new(203, 0, 113, 1)), + node(RELAYER_SEED, relay_ip), + node(CLOSE_GROUP_SEED, Ipv4Addr::new(203, 0, 113, 3)), + node(FIRST_WITNESS_SEED, Ipv4Addr::new(203, 0, 113, 3)), + node(SECOND_WITNESS_SEED, Ipv4Addr::new(203, 0, 113, 4)), + node(THIRD_WITNESS_SEED, Ipv4Addr::new(203, 0, 113, 5)), + node(DUPLICATE_IP_WITNESS_SEED, Ipv4Addr::new(203, 0, 113, 3)), + node(RELAY_IP_WITNESS_SEED, relay_ip), + ]; + let mut rng = StdRng::seed_from_u64(TEST_RNG_SEED); + + let witnesses = select_relay_canary_witnesses( + candidates, + &close_group_ids, + &target, + &relayer, + IpAddr::V4(relay_ip), + RELAY_CANARY_WITNESS_TARGET, + &mut rng, + ); + + let selected: HashSet = witnesses.iter().map(|w| w.peer_id).collect(); + assert_eq!(selected.len(), RELAY_CANARY_WITNESS_TARGET); + assert!(!selected.contains(&target)); + assert!(!selected.contains(&relayer)); + assert!(!selected.contains(&peer_id(CLOSE_GROUP_SEED))); + assert!(!selected.contains(&peer_id(RELAY_IP_WITNESS_SEED))); + assert!(selected.contains(&peer_id(SECOND_WITNESS_SEED))); + assert!(selected.contains(&peer_id(THIRD_WITNESS_SEED))); + + let duplicate_pair_selected = selected.contains(&peer_id(FIRST_WITNESS_SEED)) + && selected.contains(&peer_id(DUPLICATE_IP_WITNESS_SEED)); + assert!(!duplicate_pair_selected); + } + + #[test] + fn witness_rate_limited_is_ineligible_not_relay_failure() { + assert_eq!( + RelayCanaryProbeResult::WitnessRateLimited.disposition(), + RelayCanaryProbeDisposition::Ineligible + ); + } + + #[test] + fn explicit_probe_failures_count_as_relay_failures() { + assert_eq!( + RelayCanaryProbeResult::DialFailed.disposition(), + RelayCanaryProbeDisposition::Failure + ); + assert_eq!( + RelayCanaryProbeResult::IdentityExchangeFailed.disposition(), + RelayCanaryProbeDisposition::Failure + ); + assert_eq!( + RelayCanaryProbeResult::IdentityMismatch.disposition(), + RelayCanaryProbeDisposition::Failure + ); + } + + #[test] + fn rate_limit_throttles_per_source_not_across_sources() { + let limiter = Engine::new(relay_canary_rate_limit_config()); + let source = peer_id(FIRST_WITNESS_SEED); + let other_source = peer_id(SECOND_WITNESS_SEED); + + // First request from a source is admitted, the immediate next is not. + assert!(limiter.try_consume_key(&source)); + assert!(!limiter.try_consume_key(&source)); + // A different source is unaffected by another source's throttle. + assert!(limiter.try_consume_key(&other_source)); + } + + #[test] + fn ineligible_witnesses_produce_insufficient_witnesses() { + let mut progress = RelayCanaryProgress::new(RELAY_CANARY_WITNESS_TARGET); + + assert_eq!( + progress.record(RelayCanaryProbeDisposition::Success), + RelayCanaryProgressDecision::Continue + ); + assert_eq!( + progress.record(RelayCanaryProbeDisposition::Ineligible), + RelayCanaryProgressDecision::Continue + ); + assert_eq!( + progress.record(RelayCanaryProbeDisposition::Ineligible), + RelayCanaryProgressDecision::Continue + ); + assert_eq!( + progress.final_verdict(), + RelayCanaryVerdict::InsufficientWitnesses { + available: 1, + required: RELAY_CANARY_REQUIRED_SUCCESSES + } + ); + } + + #[test] + fn two_eligible_failures_reject_relay() { + let mut progress = RelayCanaryProgress::new(RELAY_CANARY_WITNESS_TARGET); + + assert_eq!( + progress.record(RelayCanaryProbeDisposition::Failure), + RelayCanaryProgressDecision::Continue + ); + assert_eq!( + progress.record(RelayCanaryProbeDisposition::Failure), + RelayCanaryProgressDecision::Rejected { + successes: 0, + attempts: 2 + } + ); + } + + #[test] + fn request_timeout_is_ineligible_for_mixed_version_witness() { + assert_eq!( + canary_request_error_disposition(&P2PError::Timeout(RELAY_CANARY_REQUEST_TIMEOUT)), + RelayCanaryProbeDisposition::Ineligible + ); + assert_eq!( + canary_request_error_disposition(&P2PError::Network(NetworkError::Timeout)), + RelayCanaryProbeDisposition::Ineligible + ); + } + + #[test] + fn witness_contact_failure_is_ineligible() { + assert_eq!( + canary_request_error_disposition(&P2PError::Network(NetworkError::PeerNotFound( + "witness".into() + ))), + RelayCanaryProbeDisposition::Ineligible + ); + } + + #[test] + fn canary_request_rejects_source_mismatch() { + let relay_addr = SocketAddr::from((Ipv4Addr::new(203, 0, 113, 7), TEST_PORT)); + let request = + RelayCanaryRequest::new(peer_id(TARGET_SEED), peer_id(RELAYER_SEED), relay_addr); + + let err = validate_relay_canary_request(&peer_id(FIRST_WITNESS_SEED), &request) + .expect_err("source mismatch must be rejected"); + + assert!(matches!( + err, + RelayCanaryRequestRejection::SourceMismatch { .. } + )); + } + + #[test] + fn canary_request_rejects_local_scope_relay_address() { + let request = RelayCanaryRequest::new( + peer_id(TARGET_SEED), + peer_id(RELAYER_SEED), + SocketAddr::from((Ipv4Addr::new(192, 168, 1, 10), TEST_PORT)), + ); + + let err = validate_relay_canary_request(&peer_id(TARGET_SEED), &request) + .expect_err("private relay address must be rejected"); + + assert!(matches!(err, RelayCanaryRequestRejection::LocalScopeIp(_))); + } + + #[test] + fn canary_request_rejects_unspecified_port() { + let request = RelayCanaryRequest::new( + peer_id(TARGET_SEED), + peer_id(RELAYER_SEED), + SocketAddr::from((Ipv4Addr::new(203, 0, 113, 8), UNSPECIFIED_PORT)), + ); + + let err = validate_relay_canary_request(&peer_id(TARGET_SEED), &request) + .expect_err("port zero must be rejected"); + + assert_eq!(err, RelayCanaryRequestRejection::UnspecifiedPort); + } + + #[test] + fn canary_request_rejects_self_as_relayer() { + let request = RelayCanaryRequest::new( + peer_id(TARGET_SEED), + peer_id(TARGET_SEED), + SocketAddr::from((Ipv4Addr::new(203, 0, 113, 9), TEST_PORT)), + ); + + let err = validate_relay_canary_request(&peer_id(TARGET_SEED), &request) + .expect_err("target must not be its own relayer"); + + assert!(matches!( + err, + RelayCanaryRequestRejection::SelfRelayer { .. } + )); + } +} diff --git a/src/reachability/driver.rs b/src/reachability/driver.rs index 52c541d9..77323e48 100644 --- a/src/reachability/driver.rs +++ b/src/reachability/driver.rs @@ -23,14 +23,16 @@ //! //! The driver runs as a single tokio task and cycles through three states: //! -//! 1. **Acquiring**: call [`run_relay_acquisition`]. On success, publish -//! the full typed self-record (relay-allocated address tagged -//! [`AddressType::Relay`] first, then one best non-relay address per -//! IP family) to K-closest peers, store the relayer peer ID, and enter -//! the **Holding** state. Relay serving stays permanently enabled. On -//! failure, publish the direct-only address set so the node remains as -//! reachable as possible, arm the exponential backoff timer, and enter -//! the **Backoff** state. +//! 1. **Acquiring**: call [`run_relay_acquisition`]. On success, run +//! third-party relay canaries before publishing. Only a canary-verified +//! relay is written to the full typed self-record (relay-allocated +//! address tagged [`AddressType::Relay`] first, then one best non-relay +//! address per IP family), stored as the current relayer, and held. A +//! canary-rejected relayer is excluded from subsequent acquisition attempts +//! until a relay verifies or non-close witness coverage drops below +//! quorum. On failure, publish the direct-only address set so the node +//! remains as reachable as possible, arm the exponential backoff timer, +//! and enter the **Backoff** state. //! 2. **Holding**: subscribe to `KClosestPeersChanged` events, republish //! when a pinned external address is promoted to //! [`AddressType::Direct`], and poll @@ -52,6 +54,7 @@ //! not spawn the driver at all — they are outbound-only and do not need //! a relay. +use std::collections::HashSet; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; @@ -63,6 +66,7 @@ use tracing::{debug, info, trace, warn}; use crate::dht::AddressType; use crate::dht_network_manager::{DhtNetworkEvent, DhtNetworkManager}; +use crate::reachability::canary::{RelayCanaryVerdict, verify_relay_with_canaries}; use crate::reachability::session::{RelayAcquisitionOutcome, run_relay_acquisition}; use crate::self_address::build_self_address_set; use crate::transport_handle::TransportHandle; @@ -111,6 +115,7 @@ pub(crate) fn spawn_acquisition_driver( shutdown, current_backoff: BACKOFF_INITIAL, last_published_typed_set: None, + canary_rejected_relayers: HashSet::new(), }; driver.run().await; }); @@ -127,6 +132,7 @@ struct AcquisitionDriver { shutdown: CancellationToken, current_backoff: Duration, last_published_typed_set: Option, + canary_rejected_relayers: HashSet, } #[derive(Clone, Debug, PartialEq)] @@ -135,6 +141,43 @@ struct PublishedTypedSet { peers: Vec, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CanaryRejectionEvent { + Verified, + Rejected(PeerId), + InsufficientWitnesses, + AcquisitionFailed, +} + +/// Update the per-acquisition canary exclusion set in response to an outcome. +/// +/// Exclusions accumulate only across a contiguous run of canary `Rejected` +/// verdicts, so the next acquisition walk skips a relay that just failed its +/// proof and advances to the next candidate. Every other outcome resets the +/// set: +/// - `Verified`: a relay was published; prior rejections are no longer relevant. +/// - `InsufficientWitnesses`: the relay was never disproven, only unverifiable. +/// - `AcquisitionFailed`: no candidate could be acquired at all. Preserving the +/// set here would be a trap — if the only close Direct candidate is the +/// excluded relayer, acquisition fails every round and the node stays +/// permanently relay-less. Clearing lets it retry; backoff rate-limits the +/// retries and a still-unreachable relay is simply re-excluded next round. +fn apply_canary_rejection_event( + rejected_relayers: &mut HashSet, + event: CanaryRejectionEvent, +) { + match event { + CanaryRejectionEvent::Verified + | CanaryRejectionEvent::InsufficientWitnesses + | CanaryRejectionEvent::AcquisitionFailed => { + rejected_relayers.clear(); + } + CanaryRejectionEvent::Rejected(relayer) => { + rejected_relayers.insert(relayer); + } + } +} + impl AcquisitionDriver { async fn run(&mut self) { info!("relay acquisition driver starting"); @@ -144,34 +187,110 @@ impl AcquisitionDriver { return; } - let outcome = run_relay_acquisition(self.dht.as_ref(), &self.transport).await; + let outcome = run_relay_acquisition( + self.dht.as_ref(), + &self.transport, + &self.canary_rejected_relayers, + ) + .await; match outcome { RelayAcquisitionOutcome::Acquired(relay) => { - self.current_backoff = BACKOFF_INITIAL; - *self.relayer_peer_id.write().await = Some(relay.relayer); - *self.relay_address.write().await = Some(relay.allocated_public_addr); - self.transport - .set_relay_address(relay.allocated_public_addr); - self.force_publish_typed_set(Some(relay.allocated_public_addr)) - .await; - info!( - relayer = ?relay.relayer, - allocated = %relay.allocated_public_addr, - "driver: relay acquired and published" - ); - // Hold the relay until an eviction or tunnel-death - // event forces us back into the acquisition loop. - if self.hold_until_lost().await { - // shutdown - return; + match verify_relay_with_canaries( + &self.dht, + relay.relayer, + relay.allocated_public_addr, + ) + .await + { + RelayCanaryVerdict::Verified { + successes, + attempts, + } => { + apply_canary_rejection_event( + &mut self.canary_rejected_relayers, + CanaryRejectionEvent::Verified, + ); + self.current_backoff = BACKOFF_INITIAL; + *self.relayer_peer_id.write().await = Some(relay.relayer); + *self.relay_address.write().await = Some(relay.allocated_public_addr); + self.transport + .set_relay_address(relay.allocated_public_addr); + self.force_publish_typed_set(Some(relay.allocated_public_addr)) + .await; + info!( + relayer = ?relay.relayer, + allocated = %relay.allocated_public_addr, + successes, + attempts, + "driver: relay canary verified and published" + ); + // Hold the relay until an eviction or tunnel-death + // event forces us back into the acquisition loop. + if self.hold_until_lost().await { + // shutdown + return; + } + // Fall through: hold_until_lost() returned false, the + // relay is considered lost, we need to republish + // direct-only BEFORE re-trying acquisition. + self.lose_relay_and_republish().await; + } + RelayCanaryVerdict::Rejected { + successes, + attempts, + } => { + warn!( + relayer = ?relay.relayer, + allocated = %relay.allocated_public_addr, + successes, + attempts, + "driver: relay failed canary quorum, entering backoff before trying next candidate" + ); + apply_canary_rejection_event( + &mut self.canary_rejected_relayers, + CanaryRejectionEvent::Rejected(relay.relayer), + ); + self.clear_unpublished_relay_state().await; + self.publish_typed_set(None).await; + if self.wait_backoff_or_event().await { + return; // shutdown + } + self.advance_backoff(); + } + RelayCanaryVerdict::InsufficientWitnesses { + available, + required, + } => { + warn!( + relayer = ?relay.relayer, + allocated = %relay.allocated_public_addr, + available, + required, + "driver: insufficient relay canary witnesses, entering backoff without publishing relay" + ); + apply_canary_rejection_event( + &mut self.canary_rejected_relayers, + CanaryRejectionEvent::InsufficientWitnesses, + ); + self.clear_unpublished_relay_state().await; + self.publish_typed_set(None).await; + if self.wait_backoff_or_event().await { + return; // shutdown + } + self.advance_backoff(); + } } - // Fall through: hold_until_lost() returned false, the - // relay is considered lost, we need to republish - // direct-only BEFORE re-trying acquisition. - self.lose_relay_and_republish().await; } RelayAcquisitionOutcome::Failed(reason) => { - warn!(reason, "driver: acquisition failed, entering backoff"); + warn!( + reason, + rejected_relayers = self.canary_rejected_relayers.len(), + "driver: acquisition failed, clearing canary exclusions and entering backoff" + ); + apply_canary_rejection_event( + &mut self.canary_rejected_relayers, + CanaryRejectionEvent::AcquisitionFailed, + ); *self.relayer_peer_id.write().await = None; *self.relay_address.write().await = None; self.transport.clear_relay_address(); @@ -218,6 +337,20 @@ impl AcquisitionDriver { self.publish_typed_set_with_policy(relay, true).await; } + /// Clear core advertisement state for a relay that never passed canary. + /// + /// This prevents DHT publication. Core also ignores peer ADD_ADDRESS relay + /// hints, so unverified relays cannot enter DHT gossip through the legacy + /// transport bridge. A future saorsa-transport API still needs to defer + /// outbound ADD_ADDRESS and tear down rejected MASQUE sessions for a fully + /// airtight transport-layer gate. Tracked in + /// . + async fn clear_unpublished_relay_state(&mut self) { + *self.relayer_peer_id.write().await = None; + *self.relay_address.write().await = None; + self.transport.clear_relay_address(); + } + async fn publish_typed_set_with_policy(&mut self, relay: Option, force: bool) { let listen = self.transport.listen_addrs().await; let observed = self.transport.non_relay_external_addresses(); @@ -468,3 +601,68 @@ impl AcquisitionDriver { self.current_backoff = next.min(BACKOFF_MAX); } } + +#[cfg(test)] +mod tests { + use super::*; + + const REJECTED_RELAYER_SEED: u8 = 7; + const SECOND_RELAYER_SEED: u8 = 8; + const PEER_ID_BYTES: usize = 32; + + fn peer_id(seed: u8) -> PeerId { + PeerId::from_bytes([seed; PEER_ID_BYTES]) + } + + #[test] + fn acquisition_failure_clears_canary_rejected_relayers() { + // A failed acquisition must reset exclusions: if the only close Direct + // candidate is the excluded relayer, preserving the set would fail + // acquisition every round and leave the node permanently relay-less. + let mut rejected_relayers = + HashSet::from([peer_id(REJECTED_RELAYER_SEED), peer_id(SECOND_RELAYER_SEED)]); + + apply_canary_rejection_event( + &mut rejected_relayers, + CanaryRejectionEvent::AcquisitionFailed, + ); + + assert!(rejected_relayers.is_empty()); + } + + #[test] + fn verified_relay_clears_canary_rejected_relayers() { + let mut rejected_relayers = + HashSet::from([peer_id(REJECTED_RELAYER_SEED), peer_id(SECOND_RELAYER_SEED)]); + + apply_canary_rejection_event(&mut rejected_relayers, CanaryRejectionEvent::Verified); + + assert!(rejected_relayers.is_empty()); + } + + #[test] + fn insufficient_witnesses_clear_canary_rejected_relayers() { + let mut rejected_relayers = + HashSet::from([peer_id(REJECTED_RELAYER_SEED), peer_id(SECOND_RELAYER_SEED)]); + + apply_canary_rejection_event( + &mut rejected_relayers, + CanaryRejectionEvent::InsufficientWitnesses, + ); + + assert!(rejected_relayers.is_empty()); + } + + #[test] + fn canary_rejection_adds_relayer_to_exclusion_set() { + let relayer = peer_id(REJECTED_RELAYER_SEED); + let mut rejected_relayers = HashSet::new(); + + apply_canary_rejection_event( + &mut rejected_relayers, + CanaryRejectionEvent::Rejected(relayer), + ); + + assert!(rejected_relayers.contains(&relayer)); + } +} diff --git a/src/reachability/mod.rs b/src/reachability/mod.rs index 70f728d6..1b34899c 100644 --- a/src/reachability/mod.rs +++ b/src/reachability/mod.rs @@ -11,20 +11,21 @@ // distributed under these licenses is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -//! # Unconditional relay acquisition +//! # Canary-gated relay acquisition //! //! Every non-client node tries to acquire a MASQUE relay from an XOR-closest -//! peer after bootstrap. There is no dial-back probe, no `Public`/`Private` -//! classification, and no `AssumePrivate` flag: the "is this candidate -//! public?" question is inferred ambiently from the dial attempt itself. -//! A private candidate's Direct address is unreachable from outside its NAT, -//! so the QUIC dial fails and the walker advances to the next close peer. +//! peer after bootstrap. Once a candidate accepts, the driver asks +//! independent randomized non-close witnesses to cold-dial the relay-allocated +//! address and confirm this node's authenticated identity before the address +//! is published to the DHT. //! //! ## Module layout //! //! - [`acquisition`]: the reusable XOR-closest [`RelayAcquisition`] //! coordinator. Pure logic — wraps a [`RelaySessionEstablisher`] trait so //! the walk can be unit-tested with mock establishers. +//! - [`canary`]: internal request/response protocol and quorum check used +//! to verify a freshly acquired relay from third-party vantage points. //! - [`session`]: the [`run_relay_acquisition`] entry point. Builds the //! filtered candidate list from the routing table and hands it to the //! coordinator. @@ -34,6 +35,7 @@ //! the republish-then-reacquire sequence on loss. pub(crate) mod acquisition; +pub(crate) mod canary; pub(crate) mod driver; pub(crate) mod session; diff --git a/src/reachability/session.rs b/src/reachability/session.rs index b0fb2482..7d434f36 100644 --- a/src/reachability/session.rs +++ b/src/reachability/session.rs @@ -11,27 +11,30 @@ // distributed under these licenses is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -//! Unconditional MASQUE relay acquisition. +//! MASQUE relay acquisition candidate walk. //! //! Every non-client node tries to acquire a MASQUE relay from an XOR-closest //! peer after bootstrap. Candidates on the same WAN as this node are filtered -//! out because they do not provide a distinct public route. There is no -//! dial-back probe and no broader public/private classification: the "is this -//! candidate public?" question is answered ambiently by the dial attempt -//! itself. A candidate whose Direct address is unreachable will simply fail to -//! accept the CONNECT-UDP request, and the walker moves to the next-closest -//! peer. +//! out because they do not provide a distinct public route. A candidate whose +//! Direct address is unreachable will simply fail to accept the CONNECT-UDP +//! request, and the walker moves to the next-closest peer. +//! +//! The acquisition walk only proves that the local node can reserve relay +//! service from a candidate. The driver must still run third-party canaries +//! before publishing the relay-allocated address. //! //! The acquisition walk is a thin wrapper around the reusable //! [`RelayAcquisition`] coordinator: build a filtered candidate list from //! the routing table, hand it off, and return the outcome. +use std::collections::HashSet; use std::sync::Arc; use std::time::Duration; use rand::Rng; use tracing::{debug, info, warn}; +use crate::PeerId; use crate::dht_network_manager::DhtNetworkManager; use crate::reachability::acquisition::{AcquiredRelay, RelayAcquisition, RelayCandidate}; use crate::transport_handle::TransportHandle; @@ -88,6 +91,7 @@ pub(crate) enum RelayAcquisitionOutcome { pub(crate) async fn run_relay_acquisition( dht: &DhtNetworkManager, transport: &Arc, + excluded_relayers: &HashSet, ) -> RelayAcquisitionOutcome { let jitter_ms = rand::thread_rng().gen_range(0..STARTUP_JITTER_UPPER_MS); if jitter_ms > 0 { @@ -100,11 +104,20 @@ pub(crate) async fn run_relay_acquisition( debug!( closest_count = closest.len(), + excluded_relayers = excluded_relayers.len(), "relay acquisition: evaluating closest peers for Direct relay candidates" ); let mut candidates: Vec = Vec::new(); for node in &closest { + if excluded_relayers.contains(&node.peer_id) { + debug!( + peer = %node.peer_id.to_hex(), + "relay acquisition: skipping relayer rejected by canary in this round" + ); + continue; + } + let typed = node.typed_addresses(); let direct = DhtNetworkManager::first_direct_dialable_for_relay(node, &local_address_context); diff --git a/src/transport_handle.rs b/src/transport_handle.rs index b914e3c5..ad51f8d6 100644 --- a/src/transport_handle.rs +++ b/src/transport_handle.rs @@ -380,6 +380,26 @@ pub struct TransportHandle { proof_eligible_peers: Arc>, } +struct ActiveRequestGuard { + active_requests: Arc>, + message_id: String, +} + +impl ActiveRequestGuard { + fn new(active_requests: Arc>, message_id: String) -> Self { + Self { + active_requests, + message_id, + } + } +} + +impl Drop for ActiveRequestGuard { + fn drop(&mut self) { + self.active_requests.remove(&self.message_id); + } +} + // ============================================================================ // Construction // ============================================================================ @@ -1848,6 +1868,8 @@ impl TransportHandle { expected_peer: *peer_id, }, ); + let _active_request_guard = + ActiveRequestGuard::new(Arc::clone(&self.active_requests), message_id.clone()); let envelope = RequestResponseEnvelope { message_id: message_id.clone(), @@ -1857,7 +1879,6 @@ impl TransportHandle { let envelope_bytes = match postcard::to_allocvec(&envelope) { Ok(bytes) => bytes, Err(e) => { - self.active_requests.remove(&message_id); return Err(P2PError::Serialization( format!("Failed to serialize request envelope: {e}").into(), )); @@ -1865,15 +1886,10 @@ impl TransportHandle { }; let wire_protocol = format!("/rr/{}", protocol); - if let Err(e) = self - .send_message(peer_id, &wire_protocol, envelope_bytes) - .await - { - self.active_requests.remove(&message_id); - return Err(e); - } + self.send_message(peer_id, &wire_protocol, envelope_bytes) + .await?; - let result = match tokio::time::timeout(timeout, rx).await { + match tokio::time::timeout(timeout, rx).await { Ok(Ok(response_bytes)) => { let latency = started_at.elapsed(); Ok(PeerResponse { @@ -1885,19 +1901,8 @@ impl TransportHandle { Ok(Err(_)) => Err(P2PError::Network(NetworkError::ConnectionClosed { peer_id: peer_id.to_hex().into(), })), - Err(_) => Err(P2PError::Transport( - crate::error::TransportError::StreamError( - format!( - "Request to {} on {} timed out after {:?}", - peer_id, protocol, timeout - ) - .into(), - ), - )), - }; - - self.active_requests.remove(&message_id); - result + Err(_) => Err(P2PError::Timeout(timeout)), + } } /// Send a response to a previously received request.