diff --git a/Cargo.lock b/Cargo.lock index 41210043..aea5563b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2584,9 +2584,8 @@ dependencies = [ [[package]] name = "saorsa-transport" -version = "0.35.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3284026c300f642077315b782462b22558e24d621265a8deeae23287b1c5542" +version = "0.35.2" +source = "git+https://github.com/WithAutonomi/saorsa-transport.git?branch=fix%2Fpr136-provisional-relay#4b1ed67973763e4636694a9d9c0601ff41b9e933" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index bf406d6b..e7714a35 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -64,7 +64,7 @@ once_cell = "1.21" dashmap = "6" # Networking -saorsa-transport = "0.35.3" +saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport.git", branch = "fix/pr136-provisional-relay" } # Core-specific dependencies dirs = "6.0" diff --git a/docs/ROUTING_TABLE_DESIGN.md b/docs/ROUTING_TABLE_DESIGN.md index 5e387e17..e2fdc1c8 100644 --- a/docs/ROUTING_TABLE_DESIGN.md +++ b/docs/ROUTING_TABLE_DESIGN.md @@ -396,10 +396,10 @@ The routing table MUST emit events on membership changes to allow consumers to r |---|---| | `PeerAdded(PeerId)` | New peer inserted into routing table | | `PeerRemoved(PeerId)` | Peer evicted, blocked, or departed | -| `KClosestPeersChanged { old, new }` | Composition of the `K_BUCKET_SIZE`-closest peers to self changed | +| `KClosestPeersChanged { old, new, added, removed }` | Composition of the `K_BUCKET_SIZE`-closest peers to self changed | | `BootstrapComplete { num_peers }` | Bootstrap process finished (routing table stabilized or timeout reached) | -`KClosestPeersChanged` is emitted when a routing table admission attempt causes the set of `K_BUCKET_SIZE` nearest peers to self to differ from the pre-admission set. The routing table snapshots the K-closest set before each admission attempt and compares after; the event carries both the old and new sets. This fires at most once per admission attempt — the entire admission (including sub-mutations like swaps and stale evictions) is treated as one logical operation. +`KClosestPeersChanged` is emitted when a routing table admission attempt causes the set of `K_BUCKET_SIZE` nearest peers to self to differ from the pre-admission set. The routing table snapshots the K-closest set before each admission attempt and compares after; the network event carries the old and new sets plus their precomputed `added` and `removed` differences. This fires at most once per admission attempt — the entire admission (including sub-mutations like swaps and stale evictions) is treated as one logical operation. `BootstrapComplete` is emitted once per bootstrap cycle — both at initial startup and on each auto re-bootstrap (Section 10.3). It fires when the bootstrap lookups for that cycle complete — specifically, after the self-lookup and bucket refresh operations (Section 11) have all terminated. The event carries the total number of peers in the routing table at the time of emission. Consumers (e.g., replication, application-layer services) SHOULD wait for this event before initiating operations that depend on a populated routing table. diff --git a/docs/adr/ADR-014-proactive-relay-first-nat-traversal.md b/docs/adr/ADR-014-proactive-relay-first-nat-traversal.md index d81da9a9..9a35a971 100644 --- a/docs/adr/ADR-014-proactive-relay-first-nat-traversal.md +++ b/docs/adr/ADR-014-proactive-relay-first-nat-traversal.md @@ -2,7 +2,7 @@ ## Status -Proposed +Superseded by [ADR-016](./ADR-016-canary-gated-proactive-relays.md) ## Context diff --git a/docs/adr/ADR-016-canary-gated-proactive-relays.md b/docs/adr/ADR-016-canary-gated-proactive-relays.md new file mode 100644 index 00000000..b0a4ed79 --- /dev/null +++ b/docs/adr/ADR-016-canary-gated-proactive-relays.md @@ -0,0 +1,166 @@ +# ADR-016: Canary-Gated Proactive Relays + +## Status + +Accepted + +Supersedes [ADR-014](./ADR-014-proactive-relay-first-nat-traversal.md). + +## Context + +ADR-014 described an earlier relay-first design. The implementation evolved in +several important ways: + +- a relay allocation must not be published merely because the requesting node + can establish it; +- the canary protocol is necessarily a bounded public dial service, so its + abuse controls must not depend on a requester-supplied proof that the + requester can mint for itself; +- a canary dial must not reuse or disconnect a live application connection; +- relay state changes and network teardown must not block one another behind a + lifecycle mutex; and +- close-group churn is a replication concern, not evidence that a healthy + relay should be replaced. + +This ADR records the implemented model and replaces the contradictory +thresholds, capacity limits, and maintenance behavior in ADR-014. + +## Decision + +### Relay acquisition and publication + +Every non-client node may walk suitable routing-table peers and prepare one +proactive MASQUE allocation. Preparation creates a dedicated relay control +connection and a separate Quinn endpoint, but the allocation remains +provisional and absent from the node's published address set. + +The target asks three randomized, non-close witnesses to probe the provisional +address using the unreleased `relay-canary-v1` request/response protocol. The +request contains the target peer ID, public relay socket, and an hourly witness +eligibility epoch. Its ordinary signed transport envelope must authenticate as +the same target peer ID, so a node can request a probe only for its own +identity. + +Witness eligibility is deterministic and independent of the requested +address. A domain-separated BLAKE3 hash of the target peer ID, witness peer ID, +and eligibility epoch must have its first two bits clear. This assigns roughly +one quarter of witnesses to a target for an hour and prevents a requester from +recruiting the whole routing table for one identity. A witness accepts the +current or immediately previous epoch to tolerate an hour boundary; requesters +use the current epoch and filter candidates before selecting three randomized, +non-close witnesses. + +After validating the request, an eligible witness opens a fresh one-shot +authenticated QUIC connection which never enters ordinary peer, address, or +dial-deduplication maps. The witness closes only that owned probe connection. +The wire response is deliberately coarse: success, failure, or rate limited. +Detailed dial and identity failures remain local debug information rather than +turning the protocol into a richer port-scanning oracle. + +Admission requires three positive witness results. One explicit +canary-capable failure rejects the provisional allocation. + +During the mixed-version rollout, a request that was successfully sent to a +selected witness but receives no canary-protocol response before the response +deadline counts as an assumed positive result. This preserves the pre-canary +behavior until that witness upgrades. This compatibility rule is deliberately +limited to the response stage: failure to connect to a selected witness and an +explicit rate-limit response remain ineligible; neither is promoted to +success. An assumed result is logged separately from a confirmed probe. + +The implementation still requires three selectable non-close witnesses and +intentionally has no sparse-network threshold or replacement sampling. + +Canary work has its own four-permit concurrency semaphore and hourly limits. +Before starting a dial, each witness consumes all of these budgets: + +- at most 4 probes per authenticated target peer ID; +- at most 20 probes per transport source IPv4 address or IPv6 `/64` prefix; +- at most 4 probes per destination socket; +- at most 20 probes per destination IP address; and +- at most 60 probes in total on that witness. + +The limits are intentionally redundant. Ephemeral identities cannot bypass the +source-network or witness-wide limits, while rotating destination ports cannot +bypass the destination-IP limit. The source IP is taken from the authenticated +transport connection, never from request data. Validation and budgets happen +before any canary-triggered network acquisition. Canary work does not consume +the general DHT handler budget and does not retry a failed cold dial. Replayed +requests consume the same hourly budgets as new requests. + +### Established-relay maintenance + +The node polls local tunnel health every five seconds and repeats independent +third-party canary verification every two hours, with deterministic initial +jitter spread across a full interval. The slower external cadence is +intentional: admission already proved reachability, tunnel loss is detected by +the cheap local health path, and every canary round creates three witness +requests plus three fresh PQC relay handshakes. The two-hour interval avoids +continuous fleet-wide dial pressure and remains comfortably inside the hourly +witness budgets. + +Maintenance accepts two positive witness results, including temporary +assumed-positive legacy results, and rejects on two explicit canary-capable +failures. An inconclusive maintenance round retains the relay and waits for the +ordinary two-hour interval; immediately retrying unavailable witnesses would +amplify a partial outage. A rejected round withdraws the relay immediately; it +is not confirmed by a second round. + +Tunnel death, explicit canary rejection, or an explicit trust/quality decision +may replace a relay. A healthy established relay remains in place when the +K-closest set changes. Close-group changes only publish the current +authoritative address set to peers newly entering the replication set. + +### Publication and teardown ordering + +On relay loss, local published-relay state is cleared first. DHT withdrawal and +transport teardown then run concurrently, so neither waits for the other. +Relay allocation resources are owned by a small lifecycle actor. The actor +serializes short state transitions; relay acquisition and teardown awaits run +outside it. Generation numbers prevent a late acquisition or canary verdict +from acting on a superseding allocation. Every owned allocation carries a +synchronous cleanup guard: if a lifecycle reply or graceful teardown future is +cancelled, dropping the owner closes the endpoint, aborts the tunnel tasks, and +removes the matching relay session. + +Candidate `ADD_ADDRESS` advertisements are allowed while an allocation is +absent or provisional and suppressed only after the relay reaches the +`Published` state. Relay publication itself is owned by the authenticated, +sequenced DHT address-set path. Saorsa-core therefore does not forward or drain +transport `PeerAddressUpdated` events. + +### Capacity and address-family ownership + +Public relay servers accept at most four active relay clients. A prepared +allocation must preserve the address family of the selected relay path. A +mismatch is aborted through the same transport-stack owner that created it and +is returned as an error; later publication and teardown never redispatch an +allocation to a different stack. + +### Packaging + +Dependency versioning and release packaging are managed separately by the +release process and are not decided here. + +## Consequences + +- Published relay addresses have independent external reachability evidence + when selected witnesses support canaries; during mixed-version rollout an + unsupported selected witness temporarily contributes assumed-positive + compatibility credit. +- Canary traffic cannot tear down shared application/DHT connections. +- A malicious node can ask eligible witnesses to attempt a connection to an + unrelated public address, but the authenticated-self rule, deterministic + witness assignment, hourly peer/source/destination/global limits, and + isolated concurrency budget strictly bound that service. Canary work cannot + exhaust the general handler pool. +- Healthy relay sessions avoid churn when routing-table responsibility moves. +- DHT withdrawal begins without waiting for local transport shutdown. +- Mixed-version witnesses do not block admission merely because they lack the + canary protocol; their missing protocol response is temporarily counted as + positive. +- Routing tables with fewer than three selectable non-close witnesses can + still produce inconclusive admission. +- Canary requests carry no allocation receipt. This removes untrusted + self-signed proof material and several kilobytes of redundant ML-DSA key and + signature data from every request. diff --git a/docs/adr/README.md b/docs/adr/README.md index 39ae42d4..f0759bf6 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -49,6 +49,13 @@ An Architecture Decision Record (ADR) is a document that captures an important a |-----|-------|--------|---------| | [ADR-013](./ADR-013-no-offline-delivery-v1.md) | No Offline Message Delivery (v1) | Accepted | 1-hour TTL limit without extended offline delivery (future reconsideration) | +### Reachability + +| ADR | Title | Status | Summary | +|-----|-------|--------|---------| +| [ADR-014](./ADR-014-proactive-relay-first-nat-traversal.md) | Proactive Relay-First NAT Traversal | Superseded | Initial proactive-relay design replaced by canary-gated publication | +| [ADR-016](./ADR-016-canary-gated-proactive-relays.md) | Canary-Gated Proactive Relays | Accepted | Signed allocation receipts, isolated witnesses, stable relay lifecycle, and sequenced publication | + ### Operations | ADR | Title | Status | Summary | diff --git a/src/adaptive/dht.rs b/src/adaptive/dht.rs index b0697b49..09f47937 100644 --- a/src/adaptive/dht.rs +++ b/src/adaptive/dht.rs @@ -277,6 +277,21 @@ impl AdaptiveDHT { .peer_addresses_for_dial_typed(peer_id) .await } + + /// Ensure the shared DHT dial coordinator has an authenticated channel. + /// + /// Keeping application reconnects on the same path as iterative lookups + /// means both callers share address-failure suppression and never create + /// independent retry loops against a known-bad relay. + pub(crate) async fn ensure_peer_channel( + &self, + peer_id: &PeerId, + candidates: &[(MultiAddr, AddressType)], + ) -> Result<()> { + self.dht_manager + .ensure_peer_channel(peer_id, candidates) + .await + } } #[cfg(test)] diff --git a/src/dht_network_manager.rs b/src/dht_network_manager.rs index eafd54c4..a68cba74 100644 --- a/src/dht_network_manager.rs +++ b/src/dht_network_manager.rs @@ -27,6 +27,15 @@ 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, RelayCanaryRequestOutcome, RelayCanaryResponse, + answer_relay_canary_request, relay_canary_destination_ip_rate_limit_config, + relay_canary_destination_rate_limit_config, relay_canary_global_rate_limit_config, + relay_canary_rate_limit_config, relay_canary_source_network, + relay_canary_source_network_rate_limit_config, validate_relay_canary_request, + }, security::canonicalize_ip, self_address::build_self_address_set, }; @@ -62,6 +71,7 @@ const MAX_MESSAGE_SIZE: usize = 64 * 1024; /// Prevents long-running handlers from starving the semaphore permit pool /// SEC-001: DoS mitigation via timeout enforcement on concurrent operations const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const MAX_CONCURRENT_RELAY_CANARY_PROBES: usize = 4; /// Reliability score assigned to the local node in K-closest results. /// The local node is always considered fully reliable for its own lookups. @@ -742,6 +752,8 @@ pub struct DhtNetworkManager { stats: Arc>, /// Semaphore for limiting concurrent message handlers (backpressure) message_handler_semaphore: Arc, + /// Isolated concurrency budget for canary-triggered network dials. + relay_canary_semaphore: Arc, /// Global semaphore limiting concurrent stale revalidation passes. /// Prevents a flood of revalidation attempts from consuming excessive /// resources when many buckets have stale peers simultaneously. @@ -756,6 +768,20 @@ 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, + /// Per-source-network canary limiter that survives peer-ID rotation. + relay_canary_source_network_rate_limiter: SharedEngine, + /// Node-wide canary work limiter that cannot be bypassed with new IDs. + relay_canary_global_rate_limiter: SharedEngine<&'static str>, + /// Repeated-work limiter for one destination socket. + relay_canary_destination_rate_limiter: SharedEngine, + /// Repeated-work limiter for one destination IP across rotated ports. + relay_canary_destination_ip_rate_limiter: SharedEngine, /// Shutdown token for background tasks shutdown: CancellationToken, /// Handle for the network event handler task @@ -1087,18 +1113,19 @@ impl DialFailureCache { } /// ADR-011 self-heal: when a newer authoritative `PublishAddressSet` from -/// `publisher` is applied, clear any stale dial-failure suppression for its -/// freshly published socket addresses. A recovered address — for example a relay -/// that now hands a reconnecting peer back its previous stable port — is -/// otherwise kept suppressed for up to [`DIAL_FAILURE_CACHE_TTL`] even though the -/// owner has just re-attested it. The exemption it grants is keyed by -/// `(publisher, socket)` so only a dial *to that publisher* benefits. Returns the -/// number of addresses cleared; a no-op when the publish was not applied (stale -/// or duplicate sequence). +/// `publisher` is applied, clear stale dial-failure suppression only for socket +/// addresses absent from `previous_addresses`. +/// +/// This immediately retries a genuinely withdrawn then recovered address +/// without treating a sequence-only refresh of an unchanged bad relay as proof +/// of recovery. The exemption is keyed by `(publisher, socket)` so only a dial +/// *to that publisher* benefits. Returns the number of addresses cleared; a +/// no-op when the publish was not applied (stale or duplicate sequence). fn clear_dial_failures_for_published( cache: &DialFailureCache, publisher: &PeerId, applied: bool, + previous_addresses: &[(crate::MultiAddr, AddressType)], addresses: &[(crate::MultiAddr, AddressType)], ) -> usize { if !applied { @@ -1106,6 +1133,12 @@ fn clear_dial_failures_for_published( } let mut cleared = 0; for (addr, _ty) in addresses { + if previous_addresses + .iter() + .any(|(previous, _)| previous == addr) + { + continue; + } if let Some(socket_addr) = addr.dialable_socket_addr() { // Clears this address's own failure and grants (publisher, socket) a // short exemption from IP suppression — but never lifts IP-level relay @@ -1262,6 +1295,10 @@ pub enum DhtNetworkEvent { old: Vec, /// K-closest peer IDs after the mutation. new: Vec, + /// Peers newly entering the K-closest set. + added: Vec, + /// Peers leaving the K-closest set. + removed: Vec, }, /// New peer added to the routing table. PeerAdded { peer_id: PeerId }, @@ -1752,6 +1789,7 @@ impl DhtNetworkManager { event_tx, stats: Arc::new(RwLock::new(DhtNetworkStats::default())), message_handler_semaphore, + relay_canary_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_RELAY_CANARY_PROBES)), revalidation_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_REVALIDATIONS)), bucket_revalidation_active: Arc::new(parking_lot::Mutex::new(HashSet::new())), bucket_refresh_lookup_semaphore: Arc::new(Semaphore::new( @@ -1766,6 +1804,19 @@ 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())), + relay_canary_source_network_rate_limiter: Arc::new(Engine::new( + relay_canary_source_network_rate_limit_config(), + )), + relay_canary_global_rate_limiter: Arc::new(Engine::new( + relay_canary_global_rate_limit_config(), + )), + relay_canary_destination_rate_limiter: Arc::new(Engine::new( + relay_canary_destination_rate_limit_config(), + )), + relay_canary_destination_ip_rate_limiter: Arc::new(Engine::new( + relay_canary_destination_ip_rate_limit_config(), + )), }) } @@ -3561,7 +3612,7 @@ impl DhtNetworkManager { /// broadcast and translate the shared outcome back into a /// caller-facing [`P2PError`] — they do not duplicate the /// owner's side effects. - async fn ensure_peer_channel( + pub(crate) async fn ensure_peer_channel( &self, peer_id: &PeerId, candidates: &[(MultiAddr, AddressType)], @@ -4455,13 +4506,15 @@ impl DhtNetworkManager { ); } let dht = self.dht.read().await; + let previous_addresses = dht.get_node_addresses_typed(authenticated_sender).await; let applied = dht .replace_node_addresses(authenticated_sender, filtered_addresses.clone(), *seq) .await; - // ADR-011: a newer authoritative address set just landed — lift - // stale dial-failure suppression for its addresses so a - // self-healed (e.g. reclaimed stable relay) address is retried - // immediately instead of staying suppressed for the cache TTL. + // ADR-011: a newer authoritative address set just landed. Lift + // stale dial-failure suppression only for addresses that were + // absent from the previous record and have genuinely + // reappeared (for example a reclaimed stable relay). Refreshing + // an unchanged failed address must not create a redial loop. // // Race guard: only exempt addresses that are STILL the peer's // current record after the apply (read back under the same lock @@ -4480,6 +4533,7 @@ impl DhtNetworkManager { self.dial_failure_cache.as_ref(), authenticated_sender, true, + &previous_addresses, &still_current, ) } else { @@ -4511,6 +4565,188 @@ 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 = match self + .transport + .send_request(peer_id, RELAY_CANARY_PROTOCOL, request_bytes, timeout) + .await + { + Ok(response) => response, + // The peer channel was established and the request send completed, + // but no canary response arrived. This is how a selected legacy + // witness presents during a mixed-version rollout. + Err(P2PError::Timeout(_)) => { + return Ok(RelayCanaryRequestOutcome::NoProtocolResponse); + } + Err(error) => return Err(error), + }; + postcard::from_bytes(&response.data) + .map(RelayCanaryRequestOutcome::Response) + .map_err(|e| P2PError::Serialization(e.to_string().into())) + } + + async fn handle_relay_canary_message( + &self, + source_peer: PeerId, + source_addr: Option, + 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, self.peer_id(), &request, SystemTime::now()) + { + debug!( + peer = %source_peer.to_hex(), + reason = %reason.summary(), + "Rejecting relay canary request" + ); + let response = RelayCanaryResponse { + result: RelayCanaryProbeResult::WitnessRateLimited, + }; + return self + .send_relay_canary_response(&source_peer, &message_id, response) + .await; + } + let Some(source_addr) = source_addr else { + debug!( + peer = %source_peer.to_hex(), + "Throttling relay canary request without transport provenance" + ); + let response = RelayCanaryResponse { + result: RelayCanaryProbeResult::WitnessRateLimited, + }; + return self + .send_relay_canary_response(&source_peer, &message_id, response) + .await; + }; + + // Consume every independent abuse budget before any network + // acquisition. Evaluate them separately so a denial in one dimension + // cannot be used to avoid consuming the others. + let destination = saorsa_transport::shared::normalize_socket_addr(request.relay_addr); + let source_network = relay_canary_source_network(source_addr.ip()); + let destination_ip = canonicalize_ip(destination.ip()); + let peer_allowed = self.relay_canary_rate_limiter.try_consume_key(&source_peer); + let source_network_allowed = self + .relay_canary_source_network_rate_limiter + .try_consume_key(&source_network); + let global_allowed = self + .relay_canary_global_rate_limiter + .try_consume_key(&"global"); + let destination_allowed = self + .relay_canary_destination_rate_limiter + .try_consume_key(&destination); + let destination_ip_allowed = self + .relay_canary_destination_ip_rate_limiter + .try_consume_key(&destination_ip); + if !(peer_allowed + && source_network_allowed + && global_allowed + && destination_allowed + && destination_ip_allowed) + { + debug!( + peer = %source_peer.to_hex(), + source_network = %source_network, + relay = %destination, + "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 Ok(_permit) = self.relay_canary_semaphore.try_acquire() else { + debug!( + peer = %source_peer.to_hex(), + "Throttling relay canary request: isolated probe budget exhausted" + ); + 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. @@ -5025,6 +5261,57 @@ 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 source_addr = transport_source + .as_ref() + .and_then(MultiAddr::socket_addr); + let manager_clone = Arc::clone(&self_arc); + tokio::spawn(async move { + match tokio::time::timeout( + RELAY_CANARY_HANDLER_TIMEOUT, + manager_clone + .handle_relay_canary_message( + source_peer, + source_addr, + data, + ), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) if e.is_stale_channel_send_failure() => { + // The requester may disconnect while this + // witness is completing its isolated probe. + // That is a normal request-cancellation race, + // not a canary-handler failure. + debug!( + peer = %source_peer.to_hex(), + error = %e, + "Relay canary response recipient disconnected" + ); + } + 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" + ); + } + } + }); } } }, @@ -5269,9 +5556,21 @@ impl DhtNetworkManager { .send(DhtNetworkEvent::PeerRemoved { peer_id: *id }); } RoutingTableEvent::KClosestPeersChanged { old, new } => { + let old_set: HashSet<_> = old.iter().copied().collect(); + let new_set: HashSet<_> = new.iter().copied().collect(); let _ = self.event_tx.send(DhtNetworkEvent::KClosestPeersChanged { old: old.clone(), new: new.clone(), + added: new + .iter() + .filter(|peer| !old_set.contains(peer)) + .copied() + .collect(), + removed: old + .iter() + .filter(|peer| !new_set.contains(peer)) + .copied() + .collect(), }); } } @@ -5499,12 +5798,14 @@ impl DhtNetworkManager { &self, typed_addresses: Vec<(crate::MultiAddr, AddressType)>, peers: &[DHTNode], - ) { + ) -> Vec { let seq = Self::next_publish_seq(); let op = DhtNetworkOperation::PublishAddressSet { seq, addresses: typed_addresses.clone(), }; + let mut confirmed = Vec::new(); + let mut publishes = FuturesUnordered::new(); for peer in peers { if peer.peer_id == self.config.peer_id { continue; // Skip self @@ -5512,28 +5813,49 @@ impl DhtNetworkManager { // Pass the peer's typed addresses through directly so // send_dht_request avoids a redundant routing-table read for // a peer we already have in hand. + let peer_id = peer.peer_id; let peer_typed = peer.typed_addresses(); - match self - .send_dht_request(&peer.peer_id, op.clone(), Some(&peer_typed)) - .await - { - Ok(_) => { + let op = op.clone(); + publishes.push(async move { + ( + peer_id, + self.send_dht_request(&peer_id, op, Some(&peer_typed)).await, + ) + }); + } + + // A withdrawal must not spend one full request timeout per unavailable + // peer while the rest of the network continues dialing the old relay. + // Fan the full replacement out concurrently and retain the exact + // acknowledgers so the driver can retry only missing replicas. + while let Some((peer_id, result)) = publishes.next().await { + match result { + Ok(DhtNetworkResult::PublishAddressAck) => { + confirmed.push(peer_id); debug!( - peer = %peer.peer_id.to_hex(), + peer = %peer_id.to_hex(), addrs = typed_addresses.len(), seq, "published address set to peer", ); } + Ok(other) => { + debug!( + peer = %peer_id.to_hex(), + result = ?other, + "Peer returned an unexpected address publication response" + ); + } Err(e) => { debug!( "Failed to publish address set to peer {}: {}", - peer.peer_id.to_hex(), + peer_id.to_hex(), e ); } } } + confirmed } /// Generate the next monotonic publish sequence number. @@ -7092,7 +7414,7 @@ mod tests { // A stale/duplicate publish (not applied) must NOT lift suppression. assert_eq!( - clear_dial_failures_for_published(&cache, &peer, false, &addrs), + clear_dial_failures_for_published(&cache, &peer, false, &[], &addrs), 0 ); assert!(cache.is_failed(&relay_sa, AddressType::Relay)); @@ -7101,13 +7423,53 @@ mod tests { // A newer applied publish clears the per-address failures (neither IP is // over the suppression threshold here, so both become dialable). assert_eq!( - clear_dial_failures_for_published(&cache, &peer, true, &addrs), + clear_dial_failures_for_published(&cache, &peer, true, &[], &addrs), 2 ); assert!(!cache.is_failed(&relay_sa, AddressType::Relay)); assert!(!cache.is_failed(&direct_sa, AddressType::Direct)); } + #[test] + fn repeated_unchanged_publish_does_not_clear_fresh_dial_failure() { + let cache = DialFailureCache::new(); + let peer = PeerId::from_bytes([1; 32]); + let relay = crate::MultiAddr::quic(sock("203.0.113.7:9000")); + let relay_sa = relay.dialable_socket_addr().expect("dialable"); + let addrs = vec![(relay, AddressType::Relay)]; + + cache.record_failure(relay_sa, AddressType::Relay); + assert_eq!( + clear_dial_failures_for_published(&cache, &peer, true, &addrs, &addrs), + 0 + ); + assert!( + cache.is_failed_for_dial(&peer, &relay_sa, AddressType::Relay), + "refreshing an unchanged bad relay must not create a retry loop" + ); + } + + #[test] + fn relay_reappearing_after_withdrawal_clears_old_dial_failure() { + let cache = DialFailureCache::new(); + let peer = PeerId::from_bytes([1; 32]); + let relay = crate::MultiAddr::quic(sock("203.0.113.7:9000")); + let direct = crate::MultiAddr::quic(sock("203.0.113.8:9000")); + let relay_sa = relay.dialable_socket_addr().expect("dialable"); + let previous = vec![(direct.clone(), AddressType::Direct)]; + let current = vec![(relay, AddressType::Relay), (direct, AddressType::Direct)]; + + cache.record_failure(relay_sa, AddressType::Relay); + assert_eq!( + clear_dial_failures_for_published(&cache, &peer, true, &previous, ¤t), + 1 + ); + assert!( + !cache.is_failed_for_dial(&peer, &relay_sa, AddressType::Relay), + "a genuinely withdrawn then re-acquired relay should be retried" + ); + } + #[test] fn publish_self_heal_keeps_ip_tier_suppression_for_others() { let cache = DialFailureCache::new(); @@ -7128,7 +7490,7 @@ mod tests { // The owner re-attests one of those addresses via an applied publish. let addrs = vec![(crate::MultiAddr::quic(socks[0]), AddressType::Relay)]; assert_eq!( - clear_dial_failures_for_published(&cache, &peer, true, &addrs), + clear_dial_failures_for_published(&cache, &peer, true, &[], &addrs), 1 ); @@ -7169,7 +7531,7 @@ mod tests { // Its owner reclaims and re-attests it via an applied publish. let addrs = vec![(crate::MultiAddr::quic(socks[0]), AddressType::Relay)]; assert_eq!( - clear_dial_failures_for_published(&cache, &owner, true, &addrs), + clear_dial_failures_for_published(&cache, &owner, true, &[], &addrs), 1 ); @@ -7574,14 +7936,28 @@ mod tests { let event = DhtNetworkEvent::KClosestPeersChanged { old: old.clone(), new: new.clone(), + added: new + .iter() + .filter(|peer| !old.contains(peer)) + .copied() + .collect(), + removed: old + .iter() + .filter(|peer| !new.contains(peer)) + .copied() + .collect(), }; match event { DhtNetworkEvent::KClosestPeersChanged { old: got_old, new: got_new, + added, + removed, } => { assert_eq!(got_old, old); assert_eq!(got_new, new); + assert_eq!(added, new); + assert_eq!(removed, old); } _ => panic!("expected KClosestPeersChanged"), } diff --git a/src/identity/node_identity.rs b/src/identity/node_identity.rs index 4cb69f0b..2c296129 100644 --- a/src/identity/node_identity.rs +++ b/src/identity/node_identity.rs @@ -72,6 +72,24 @@ pub fn peer_id_from_public_key_bytes(bytes: &[u8]) -> Result { Ok(peer_id_from_public_key(&public_key)) } +/// Create a [`PeerId`] from an authenticated ML-DSA-65 TLS SPKI. +/// +/// `saorsa-transport` exposes the exact peer certificate identity from a +/// completed QUIC/TLS handshake as DER-encoded SubjectPublicKeyInfo. Validate +/// the DER shape, algorithm identifier, absent ML-DSA parameters, and +/// byte-aligned key before deriving the overlay identity from the raw key. +pub(crate) fn peer_id_from_public_key_spki(spki_bytes: &[u8]) -> Result { + let public_key = + saorsa_transport::crypto::raw_public_keys::pqc::extract_public_key_from_spki(spki_bytes) + .map_err(|e| { + P2PError::Identity(IdentityError::InvalidFormat( + format!("Invalid ML-DSA SubjectPublicKeyInfo: {e}").into(), + )) + })?; + + peer_id_from_public_key_bytes(public_key.as_bytes()) +} + /// Public node identity information (without secret keys) - safe to clone #[derive(Clone)] pub struct PublicNodeIdentity { @@ -331,6 +349,37 @@ impl NodeIdentity { mod tests { use super::*; + fn ml_dsa_65_spki(public_key: &[u8]) -> Vec { + const OID: [u8; 9] = [0x60, 0x86, 0x48, 0x01, 0x65, 0x03, 0x04, 0x03, 0x12]; + let bit_string_len = public_key.len() + 1; + let algorithm_len = 2 + OID.len(); + let algorithm_total_len = 2 + algorithm_len; + let bit_string_total_len = 4 + bit_string_len; + let outer_len = algorithm_total_len + bit_string_total_len; + + let mut encoded = Vec::with_capacity(4 + outer_len); + encoded.extend_from_slice(&[ + 0x30, + 0x82, + (outer_len >> 8) as u8, + outer_len as u8, + 0x30, + algorithm_len as u8, + 0x06, + OID.len() as u8, + ]); + encoded.extend_from_slice(&OID); + encoded.extend_from_slice(&[ + 0x03, + 0x82, + (bit_string_len >> 8) as u8, + bit_string_len as u8, + 0x00, + ]); + encoded.extend_from_slice(public_key); + encoded + } + #[test] fn test_peer_id_generation() { let (public_key, _secret_key) = crate::quantum_crypto::generate_ml_dsa_keypair() @@ -345,6 +394,37 @@ mod tests { assert_eq!(peer_id, peer_id2); } + #[test] + fn transport_spki_derives_same_peer_id_as_raw_key() { + let (public_key, _secret_key) = crate::quantum_crypto::generate_ml_dsa_keypair() + .expect("ML-DSA key generation should succeed"); + let spki = ml_dsa_65_spki(public_key.as_bytes()); + + let from_spki = + peer_id_from_public_key_spki(&spki).expect("valid ML-DSA SPKI should parse"); + + assert_eq!(from_spki, peer_id_from_public_key(&public_key)); + } + + #[test] + fn transport_spki_rejects_wrong_algorithm() { + let (public_key, _secret_key) = crate::quantum_crypto::generate_ml_dsa_keypair() + .expect("ML-DSA key generation should succeed"); + let mut spki = ml_dsa_65_spki(public_key.as_bytes()); + let oid_last_byte = 16; + spki[oid_last_byte] = 0x11; + + assert!(peer_id_from_public_key_spki(&spki).is_err()); + } + + #[test] + fn transport_spki_rejects_raw_public_key_bytes() { + let (public_key, _secret_key) = crate::quantum_crypto::generate_ml_dsa_keypair() + .expect("ML-DSA key generation should succeed"); + + assert!(peer_id_from_public_key_spki(public_key.as_bytes()).is_err()); + } + #[test] fn test_xor_distance() { let id1 = PeerId([0u8; 32]); diff --git a/src/network.rs b/src/network.rs index 6370faeb..6392dd5e 100644 --- a/src/network.rs +++ b/src/network.rs @@ -21,10 +21,8 @@ use crate::adaptive::trust::{TrustRecord, TrustSnapshot}; use crate::adaptive::{AdaptiveDHT, AdaptiveDhtConfig, TrustEngine, TrustEvent}; use crate::bootstrap::cache::{CachedCloseGroupPeer, CloseGroupCache}; use crate::dht::core_engine::AddressType; -use crate::dht_network_manager::{ - DhtNetworkConfig, DhtNetworkEvent, DhtNetworkManager, IDENTITY_EXCHANGE_TIMEOUT, -}; -use crate::error::{IdentityError, NetworkError, P2PError, P2pResult as Result}; +use crate::dht_network_manager::{DhtNetworkConfig, DhtNetworkEvent, DhtNetworkManager}; +use crate::error::{NetworkError, P2PError, P2pResult as Result}; use crate::reachability::spawn_acquisition_driver; use crate::MultiAddr; @@ -138,7 +136,8 @@ const DEFAULT_CONNECTION_TIMEOUT_SECS: u64 = 25; /// Timeout in seconds for waiting on a bootstrap peer's identity exchange. /// -/// Tighter than the post-bootstrap budget (`IDENTITY_EXCHANGE_TIMEOUT`, +/// Tighter than the post-bootstrap budget +/// ([`crate::dht_network_manager::IDENTITY_EXCHANGE_TIMEOUT`], /// 5 s) on purpose: bootstrap candidates are unverified and a stuck one /// must not be allowed to head-of-line block convergence. 3 s covers /// loopback (<100 ms) and direct WAN paths (~1–2 s with one handshake @@ -1065,7 +1064,7 @@ impl P2PNode { /// /// `timeout` bounds only the response wait inside the transport; the dial /// is independently bounded by `connect_peer_typed` plus - /// [`IDENTITY_EXCHANGE_TIMEOUT`]. + /// [`crate::dht_network_manager::IDENTITY_EXCHANGE_TIMEOUT`]. async fn send_request_reconnecting( &self, peer_id: &PeerId, @@ -1306,104 +1305,6 @@ impl P2PNode { info!("client mode — skipping relay acquisition driver"); } - // Spawn background task to forward peer address updates to the DHT. - // - // 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. - { - let transport = Arc::clone(&self.transport); - let dht = self.adaptive_dht.dht_manager().clone(); - let shutdown = self.shutdown.clone(); - tokio::spawn(async move { - loop { - tokio::select! { - biased; - _ = shutdown.cancelled() => break, - update = transport.recv_peer_address_update() => { - let Some((peer_addr, advertised_addr)) = update else { break }; - let normalized_peer = - 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. - if normalized_peer.ip() == normalized_adv.ip() { - debug!( - "DHT_BRIDGE: dropping same-IP update peer={} addr={}", - normalized_peer, - normalized_adv - ); - continue; - } - info!( - "DHT_BRIDGE: processing relay update 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 - ); - } - } - } - } - } - }); - } - self.is_started .store(true, std::sync::atomic::Ordering::Release); @@ -1700,14 +1601,6 @@ impl P2PNode { saved_addrs: &[MultiAddr], stale_channels: &[String], ) -> Result<()> { - // Resolve a dial address: caller-provided > saved > DHT. - let (address, kind) = self - .resolve_dial_address(peer_id, addrs, saved_addrs) - .await - .ok_or_else(|| { - P2PError::Network(NetworkError::PeerNotFound(peer_id.to_hex().into())) - })?; - // Tear down stale QUIC connections using their actual channel IDs. // transport.send_message only removes bookkeeping (peer_to_channel, // peers, active_connections) — it does NOT close the underlying QUIC @@ -1720,31 +1613,17 @@ impl P2PNode { tokio::time::sleep(QUIC_TEARDOWN_GRACE).await; } - // Dial and wait for identity exchange. - let channel_id = self.transport.connect_peer_typed(&address, kind).await?; - let authenticated = match self - .transport - .wait_for_peer_identity(&channel_id, IDENTITY_EXCHANGE_TIMEOUT) - .await - { - Ok(peer) => peer, - Err(e) => { - // Close the freshly-dialed QUIC connection so it doesn't - // linger as a zombie until idle timeout. - self.transport.disconnect_channel(&channel_id).await; - return Err(e); - } - }; - - if &authenticated != peer_id { - self.transport.disconnect_channel(&channel_id).await; - return Err(P2PError::Identity(IdentityError::IdentityMismatch { - expected: peer_id.to_hex().into(), - actual: authenticated.to_hex().into(), - })); + let candidates = self + .resolve_dial_candidates(peer_id, addrs, saved_addrs) + .await; + if candidates.is_empty() { + return Err(P2PError::Network(NetworkError::PeerNotFound( + peer_id.to_hex().into(), + ))); } - - Ok(()) + self.adaptive_dht + .ensure_peer_channel(peer_id, &candidates) + .await } /// Tear down stale channels, reconnect to a peer, and send a message. @@ -1763,51 +1642,35 @@ impl P2PNode { self.transport.send_message(peer_id, protocol, data).await } - /// Resolve a dial address for `peer_id`, preferring caller-provided + /// Resolve typed dial candidates for `peer_id`, preferring caller-provided /// addresses over cached/DHT sources. /// - /// Returns the first dialable (QUIC, non-unspecified) address found, - /// paired with the [`AddressType`] the DHT routing table believes - /// for that address. Caller-provided / saved addresses that don't - /// appear in the routing table fall back to + /// Returns every dialable (QUIC, non-unspecified) address from the first + /// non-empty source. Caller-provided / saved addresses inherit the + /// [`AddressType`] from the DHT when possible and otherwise fall back to /// [`AddressType::Unverified`] — the same default the routing table /// applies to legacy peers that never asserted reachability. - /// Returns `None` when no dialable address is available. - async fn resolve_dial_address( + async fn resolve_dial_candidates( &self, peer_id: &PeerId, caller_addrs: &[MultiAddr], saved_addrs: &[MultiAddr], - ) -> Option<(MultiAddr, AddressType)> { - // Caller- and saved-supplied addresses skip the routing-table read. - // The kind is only consumed as a log tag by `connect_peer_typed`, so - // defaulting to Unverified — the same fallback the routing table - // applies to legacy peers — saves an async lookup on the hot - // reconnect path. Only consult the DHT when both upstream sources - // are exhausted. - if let Some(addr) = Self::first_dialable(caller_addrs) { - return Some((addr, AddressType::Unverified)); - } - if let Some(addr) = Self::first_dialable(saved_addrs) { - return Some((addr, AddressType::Unverified)); - } - - self.adaptive_dht + ) -> Vec<(MultiAddr, AddressType)> { + let dht_candidates = self + .adaptive_dht .peer_addresses_for_dial_typed(peer_id) - .await - .into_iter() - .find(|(a, _)| { - a.dialable_socket_addr() - .is_some_and(|sa| !sa.ip().is_unspecified()) - }) - } + .await; + let preferred = if !caller_addrs.is_empty() { + caller_addrs + } else if !saved_addrs.is_empty() { + saved_addrs + } else { + return dht_candidates; + }; - /// Return the first dialable QUIC address from a slice, skipping - /// non-QUIC and unspecified (`0.0.0.0` / `::`) addresses. - fn first_dialable(addrs: &[MultiAddr]) -> Option { - addrs + preferred .iter() - .find(|a| { + .filter(|a| { let dialable = a .dialable_socket_addr() .is_some_and(|sa| !sa.ip().is_unspecified()); @@ -1816,7 +1679,14 @@ impl P2PNode { } dialable }) - .cloned() + .map(|addr| { + let kind = dht_candidates + .iter() + .find_map(|(candidate, kind)| (candidate == addr).then_some(*kind)) + .unwrap_or(AddressType::Unverified); + (addr.clone(), kind) + }) + .collect() } /// Get or create a per-peer reconnect lock. diff --git a/src/reachability/acquisition.rs b/src/reachability/acquisition.rs index 7322adca..9d2a25c2 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 @@ -45,6 +44,7 @@ use std::net::SocketAddr; use async_trait::async_trait; +use saorsa_transport::nat_traversal_api::PreparedRelay; use thiserror::Error; use tracing::{debug, info, warn}; @@ -82,15 +82,16 @@ impl RelayCandidate { /// /// 1. Remember `relayer` so the monitor can rebind if that peer drops out of /// the K closest set. -/// 2. Publish `allocated_public_addr` (tagged `AddressType::Relay`) as its +/// 2. Publish `allocation.public_addr()` (tagged `AddressType::Relay`) as its /// contact address in the DHT self-record. #[derive(Debug, Clone, PartialEq, Eq)] pub struct AcquiredRelay { /// Peer ID of the node running the MASQUE relay we are now using. pub relayer: PeerId, - /// The public socket address the relay allocated for inbound traffic to - /// us. This is what the private peer publishes as its contact address. - pub allocated_public_addr: SocketAddr, + /// Opaque transport allocation. Its public address is what the private + /// peer publishes after the canary gate accepts it; its identity ensures a + /// late verdict cannot act on a replacement allocation at the same address. + pub allocation: PreparedRelay, } /// Per-candidate establishment error returned by a [`RelaySessionEstablisher`]. @@ -101,7 +102,7 @@ pub struct AcquiredRelay { pub enum RelaySessionEstablishError { /// The relay refused because its relay-client slots are full. See the /// saorsa-transport `NatTraversalError::RelayAtCapacity` variant and - /// ADR-014's 2-client-per-public-peer cap. + /// ADR-016's four-client-per-public-peer cap. #[error("relay at client capacity: {0}")] AtCapacity(String), /// The relay could not be reached at all (timeout, refused, protocol @@ -131,26 +132,26 @@ pub enum RelayAcquisitionError { /// Establishes a proactive MASQUE relay session against a candidate relay. /// /// A production implementation of this trait wraps saorsa-transport's -/// `NatTraversalEndpoint::setup_proactive_relay()`, which establishes the -/// MASQUE `CONNECT-UDP` session, rebinds the local Quinn endpoint onto the -/// tunnel, and returns the allocated public address. Test implementations -/// return canned results to exercise the coordinator's walk logic. +/// `NatTraversalEndpoint::prepare_proactive_relay()`, which establishes the +/// MASQUE `CONNECT-UDP` session and a provisional Quinn endpoint without +/// advertising the allocated address. The reachability driver publishes or +/// aborts that allocation after its canary verdict. Test implementations return +/// canned results to exercise the coordinator's walk logic. #[async_trait] pub trait RelaySessionEstablisher: Send + Sync + 'static { /// Attempt to establish a proactive relay session with the peer reachable /// at `relay_addr`. /// - /// - Returns `Ok(allocated_public_addr)` when the MASQUE session is - /// established and the local endpoint has been rebound onto the tunnel. - /// The returned socket address is the relay-allocated public address - /// the caller should publish. + /// - Returns `Ok(allocation)` when the provisional MASQUE + /// session and relay endpoint are ready for an inbound canary. The + /// returned socket address is not advertised until the caller commits it. /// - Returns `Err(AtCapacity(_))` when the relay refused because its /// client slots are full. /// - Returns `Err(Unreachable(_))` for any network-level failure. async fn establish( &self, relay_addr: SocketAddr, - ) -> Result; + ) -> Result; } /// XOR-closest relay acquisition coordinator. @@ -207,13 +208,13 @@ impl RelayAcquisition { Ok(allocated) => { info!( relayer = ?candidate.peer_id, - allocated = %allocated, + allocated = %allocated.public_addr(), index = index, "acquired proactive relay session" ); return Ok(AcquiredRelay { relayer: candidate.peer_id, - allocated_public_addr: allocated, + allocation: allocated, }); } Err(RelaySessionEstablishError::AtCapacity(reason)) => { @@ -267,12 +268,12 @@ mod tests { /// `calls` atomic tracks the number of invocations so tests can verify /// the coordinator walked exactly as far as expected and no further. struct ScriptedEstablisher { - outcomes: std::sync::Mutex>>, + outcomes: std::sync::Mutex>>, calls: Arc, } impl ScriptedEstablisher { - fn new(outcomes: Vec>) -> Self { + fn new(outcomes: Vec>) -> Self { Self { outcomes: std::sync::Mutex::new(outcomes), calls: Arc::new(AtomicUsize::new(0)), @@ -285,7 +286,7 @@ mod tests { async fn establish( &self, _relay_addr: SocketAddr, - ) -> Result { + ) -> Result { self.calls.fetch_add(1, Ordering::SeqCst); let mut guard = self.outcomes.lock().expect("mutex poisoned in test"); if guard.is_empty() { @@ -295,8 +296,11 @@ mod tests { } } - fn allocated(port: u16) -> SocketAddr { - SocketAddr::new(std::net::IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)), port) + fn allocated(port: u16) -> PreparedRelay { + PreparedRelay::detached(SocketAddr::new( + std::net::IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)), + port, + )) } #[tokio::test] @@ -318,7 +322,7 @@ mod tests { .await .expect("should succeed"); assert_eq!(result.relayer, peer_id(1)); - assert_eq!(result.allocated_public_addr, allocated(9000)); + assert_eq!(result.allocation.public_addr().port(), 9000); assert_eq!( calls.load(Ordering::SeqCst), 1, @@ -340,7 +344,7 @@ mod tests { .await .expect("should succeed"); assert_eq!(result.relayer, peer_id(2)); - assert_eq!(result.allocated_public_addr, allocated(9001)); + assert_eq!(result.allocation.public_addr().port(), 9001); assert_eq!(calls.load(Ordering::SeqCst), 2); } diff --git a/src/reachability/canary.rs b/src/reachability/canary.rs new file mode 100644 index 00000000..a2ab1363 --- /dev/null +++ b/src/reachability/canary.rs @@ -0,0 +1,1249 @@ +// 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, SystemTime, UNIX_EPOCH}; + +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}; +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; + +/// Positive witness results needed before a relay is publishable. +/// +/// Publication is unanimous across the selected independent witnesses. During +/// the mixed-version rollout, a request that was delivered but received no +/// canary-protocol response counts as a positive result so legacy nodes do not +/// make relay availability worse than before the canary gate. +const RELAY_CANARY_ADMISSION_SUCCESSES: usize = RELAY_CANARY_WITNESS_TARGET; + +/// One explicit canary-capable dial failure is enough to reject a provisional +/// relay. +const RELAY_CANARY_ADMISSION_FAILURES: usize = 1; + +/// Established relays use majority evidence so one witness-specific network +/// failure cannot withdraw a relay that two other witnesses just reached. +const RELAY_CANARY_MAINTENANCE_SUCCESSES: usize = 2; +const RELAY_CANARY_MAINTENANCE_FAILURES: 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(8); + +const RELAY_CANARY_RATE_WINDOW: Duration = Duration::from_secs(60 * 60); +const RELAY_CANARY_PEER_RATE_MAX_PER_WINDOW: u32 = 4; +const RELAY_CANARY_SOURCE_NETWORK_RATE_MAX_PER_WINDOW: u32 = 20; +const RELAY_CANARY_DESTINATION_RATE_MAX_PER_WINDOW: u32 = 4; +const RELAY_CANARY_DESTINATION_IP_RATE_MAX_PER_WINDOW: u32 = 20; +const RELAY_CANARY_GLOBAL_RATE_MAX_PER_WINDOW: u32 = 60; + +/// An eligibility assignment is stable for one hour. +const RELAY_CANARY_ELIGIBILITY_EPOCH_SECS: u64 = 60 * 60; + +/// Two clear high bits select approximately one quarter of witnesses. +const RELAY_CANARY_ELIGIBILITY_MASK: u8 = 0b1100_0000; + +/// Per-authenticated-peer throttle applied to inbound relay canary requests. +pub(crate) fn relay_canary_rate_limit_config() -> EngineConfig { + EngineConfig { + window: RELAY_CANARY_RATE_WINDOW, + max_requests: RELAY_CANARY_PEER_RATE_MAX_PER_WINDOW, + burst_size: RELAY_CANARY_PEER_RATE_MAX_PER_WINDOW, + } +} + +/// Source-network throttle that cannot be bypassed by rotating peer IDs. +pub(crate) fn relay_canary_source_network_rate_limit_config() -> EngineConfig { + EngineConfig { + window: RELAY_CANARY_RATE_WINDOW, + max_requests: RELAY_CANARY_SOURCE_NETWORK_RATE_MAX_PER_WINDOW, + burst_size: RELAY_CANARY_SOURCE_NETWORK_RATE_MAX_PER_WINDOW, + } +} + +/// Node-wide cap on accepted canary work, independent of requester identity. +pub(crate) fn relay_canary_global_rate_limit_config() -> EngineConfig { + EngineConfig { + window: RELAY_CANARY_RATE_WINDOW, + max_requests: RELAY_CANARY_GLOBAL_RATE_MAX_PER_WINDOW, + burst_size: RELAY_CANARY_GLOBAL_RATE_MAX_PER_WINDOW, + } +} + +/// Cap repeated canary work aimed at the same destination socket. +pub(crate) fn relay_canary_destination_rate_limit_config() -> EngineConfig { + EngineConfig { + window: RELAY_CANARY_RATE_WINDOW, + max_requests: RELAY_CANARY_DESTINATION_RATE_MAX_PER_WINDOW, + burst_size: RELAY_CANARY_DESTINATION_RATE_MAX_PER_WINDOW, + } +} + +/// Cap repeated canary work aimed at one IP even when the port is rotated. +pub(crate) fn relay_canary_destination_ip_rate_limit_config() -> EngineConfig { + EngineConfig { + window: RELAY_CANARY_RATE_WINDOW, + max_requests: RELAY_CANARY_DESTINATION_IP_RATE_MAX_PER_WINDOW, + burst_size: RELAY_CANARY_DESTINATION_IP_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) relay_addr: SocketAddr, + pub(crate) eligibility_epoch: u64, +} + +impl RelayCanaryRequest { + fn new(target_peer_id: PeerId, relay_addr: SocketAddr, eligibility_epoch: u64) -> Self { + Self { + target_peer_id, + relay_addr, + eligibility_epoch, + } + } +} + +/// Witness response after attempting the cold relay dial. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct RelayCanaryResponse { + pub(crate) result: RelayCanaryProbeResult, +} + +/// Request-level outcome after the requester has contacted a selected witness. +#[derive(Debug)] +pub(crate) enum RelayCanaryRequestOutcome { + /// The witness supports the canary protocol and returned a typed result. + Response(RelayCanaryResponse), + /// The request was sent, but no canary-protocol response arrived. + NoProtocolResponse, +} + +/// Result of one witness's relay probe. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum RelayCanaryProbeResult { + Success, + Failure, + WitnessRateLimited, +} + +impl RelayCanaryProbeResult { + fn disposition(&self) -> RelayCanaryProbeDisposition { + match self { + Self::Success => RelayCanaryProbeDisposition::Success, + Self::WitnessRateLimited => RelayCanaryProbeDisposition::Ineligible, + Self::Failure => RelayCanaryProbeDisposition::Failure, + } + } + + fn summary(&self) -> String { + match self { + Self::Success => "success".to_string(), + Self::Failure => "probe failed".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, + }, + InvalidClock, + StaleEligibilityEpoch { + requested: u64, + current: u64, + }, + IneligibleWitness { + witness_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::InvalidClock => "system clock is before the Unix epoch".to_string(), + Self::StaleEligibilityEpoch { requested, current } => { + format!("eligibility epoch {requested} is neither current ({current}) nor previous") + } + Self::IneligibleWitness { witness_peer_id } => format!( + "witness {} is not assigned to this target in the requested epoch", + witness_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, + witness_peer_id: &PeerId, + request: &RelayCanaryRequest, + now: SystemTime, +) -> 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, + }); + } + validate_relay_canary_address(request.relay_addr)?; + + let current = relay_canary_eligibility_epoch(now)?; + if request.eligibility_epoch != current + && Some(request.eligibility_epoch) != current.checked_sub(1) + { + return Err(RelayCanaryRequestRejection::StaleEligibilityEpoch { + requested: request.eligibility_epoch, + current, + }); + } + if !relay_canary_witness_is_eligible( + &request.target_peer_id, + witness_peer_id, + request.eligibility_epoch, + ) { + return Err(RelayCanaryRequestRejection::IneligibleWitness { + witness_peer_id: *witness_peer_id, + }); + } + + Ok(()) +} + +fn relay_canary_eligibility_epoch( + now: SystemTime, +) -> std::result::Result { + now.duration_since(UNIX_EPOCH) + .map(|elapsed| elapsed.as_secs() / RELAY_CANARY_ELIGIBILITY_EPOCH_SECS) + .map_err(|_| RelayCanaryRequestRejection::InvalidClock) +} + +fn relay_canary_witness_is_eligible( + target_peer_id: &PeerId, + witness_peer_id: &PeerId, + epoch: u64, +) -> bool { + let mut hasher = blake3::Hasher::new(); + hasher.update(b"saorsa-relay-canary-witness-v1\0"); + hasher.update(target_peer_id.to_bytes()); + hasher.update(witness_peer_id.to_bytes()); + hasher.update(&epoch.to_le_bytes()); + hasher.finalize().as_bytes()[0] & RELAY_CANARY_ELIGIBILITY_MASK == 0 +} + +/// Bucket IPv4 sources by address and IPv6 sources by `/64` prefix. +pub(crate) fn relay_canary_source_network(ip: IpAddr) -> IpAddr { + match canonicalize_ip(ip) { + IpAddr::V4(ipv4) => IpAddr::V4(ipv4), + IpAddr::V6(ipv6) => { + let bits = u128::from(ipv6) & (!0_u128 << 64); + IpAddr::V6(bits.into()) + } + } +} + +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, + }, + Inconclusive { + successes: usize, + failures: usize, + unavailable: usize, + }, +} + +/// Evidence policy for a relay canary round. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RelayCanaryPolicy { + /// A provisional relay needs a positive result from every selected witness. + Admission, + /// An established relay is retained or rejected by a completed majority. + Maintenance, +} + +impl RelayCanaryPolicy { + fn required_successes(self) -> usize { + match self { + Self::Admission => RELAY_CANARY_ADMISSION_SUCCESSES, + Self::Maintenance => RELAY_CANARY_MAINTENANCE_SUCCESSES, + } + } + + fn required_failures(self) -> usize { + match self { + Self::Admission => RELAY_CANARY_ADMISSION_FAILURES, + Self::Maintenance => RELAY_CANARY_MAINTENANCE_FAILURES, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RelayCanaryProbeDisposition { + Success, + AssumedSuccess, + Failure, + Ineligible, +} + +#[derive(Debug, Clone)] +struct RelayCanarySummary { + total: usize, + responses: usize, + eligible_attempts: usize, + successes: usize, + assumed_successes: usize, + ineligible: usize, +} + +impl RelayCanarySummary { + fn new(total: usize) -> Self { + Self { + total, + responses: 0, + eligible_attempts: 0, + successes: 0, + assumed_successes: 0, + ineligible: 0, + } + } + + fn record(&mut self, disposition: RelayCanaryProbeDisposition) { + self.responses += 1; + match disposition { + RelayCanaryProbeDisposition::Success => { + self.eligible_attempts += 1; + self.successes += 1; + } + RelayCanaryProbeDisposition::AssumedSuccess => { + self.eligible_attempts += 1; + self.successes += 1; + self.assumed_successes += 1; + } + RelayCanaryProbeDisposition::Failure => { + self.eligible_attempts += 1; + } + RelayCanaryProbeDisposition::Ineligible => { + self.ineligible += 1; + } + } + } + + fn verdict(&self, policy: RelayCanaryPolicy) -> RelayCanaryVerdict { + let failures = self.eligible_attempts.saturating_sub(self.successes); + if self.successes >= policy.required_successes() { + RelayCanaryVerdict::Verified { + successes: self.successes, + attempts: self.eligible_attempts, + } + } else if failures >= policy.required_failures() { + RelayCanaryVerdict::Rejected { + successes: self.successes, + attempts: self.eligible_attempts, + } + } else { + RelayCanaryVerdict::Inconclusive { + successes: self.successes, + failures, + unavailable: self.ineligible + + RELAY_CANARY_WITNESS_TARGET.saturating_sub(self.total), + } + } + } +} + +#[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, + policy: RelayCanaryPolicy, +) -> RelayCanaryVerdict { + let target_peer_id = *dht.peer_id(); + if let Err(reason) = validate_relay_canary_address(relay_addr) { + 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 eligibility_epoch = match relay_canary_eligibility_epoch(SystemTime::now()) { + Ok(epoch) => epoch, + Err(reason) => { + warn!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + reason = %reason.summary(), + "relay canary: refusing request with invalid system clock" + ); + return RelayCanaryVerdict::Inconclusive { + successes: 0, + failures: 0, + unavailable: RELAY_CANARY_WITNESS_TARGET, + }; + } + }; + + 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(), + eligibility_epoch, + RELAY_CANARY_WITNESS_TARGET, + &mut rand::thread_rng(), + ); + + if witnesses.len() < policy.required_successes() { + warn!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + available = witnesses.len(), + required = policy.required_successes(), + ?policy, + close_group_excluded = close_group_ids.len(), + routing_table_size, + "relay canary: insufficient random non-close witnesses, refusing to publish relay" + ); + return RelayCanaryVerdict::Inconclusive { + successes: 0, + failures: 0, + unavailable: RELAY_CANARY_WITNESS_TARGET.saturating_sub(witnesses.len()), + }; + } + + 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 summary = RelayCanarySummary::new(witnesses.len()); + let mut probes = FuturesUnordered::new(); + for witness in witnesses { + let dht = Arc::clone(dht); + let request = RelayCanaryRequest::new(target_peer_id, relay_addr, eligibility_epoch); + probes.push(async move { request_relay_canary(dht, witness, request).await }); + } + + while let Some(report) = probes.next().await { + summary.record(report.disposition); + match report.disposition { + RelayCanaryProbeDisposition::Success => { + debug!( + witness = %report.witness.to_hex(), + successes = summary.successes, + eligible_attempts = summary.eligible_attempts, + responses = summary.responses, + "relay canary: witness confirmed relay" + ); + } + RelayCanaryProbeDisposition::AssumedSuccess => { + debug!( + witness = %report.witness.to_hex(), + detail = %report.detail, + successes = summary.successes, + assumed_successes = summary.assumed_successes, + eligible_attempts = summary.eligible_attempts, + responses = summary.responses, + "relay canary: assuming positive result from legacy witness" + ); + } + RelayCanaryProbeDisposition::Ineligible => { + debug!( + witness = %report.witness.to_hex(), + detail = %report.detail, + ineligible = summary.ineligible, + eligible_attempts = summary.eligible_attempts, + responses = summary.responses, + "relay canary: witness could not evaluate relay" + ); + } + RelayCanaryProbeDisposition::Failure => { + debug!( + witness = %report.witness.to_hex(), + detail = %report.detail, + successes = summary.successes, + eligible_attempts = summary.eligible_attempts, + responses = summary.responses, + "relay canary: witness failed relay probe" + ); + } + } + } + + let verdict = summary.verdict(policy); + match &verdict { + RelayCanaryVerdict::Verified { + successes, + attempts, + } => info!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + successes, + attempts, + responses = summary.responses, + assumed_successes = summary.assumed_successes, + ineligible = summary.ineligible, + available_witnesses = summary.total, + ?policy, + "relay canary: completed witness round verified relay" + ), + RelayCanaryVerdict::Rejected { + successes, + attempts, + } => warn!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + successes, + attempts, + responses = summary.responses, + assumed_successes = summary.assumed_successes, + ineligible = summary.ineligible, + available_witnesses = summary.total, + ?policy, + "relay canary: completed witness round rejected relay" + ), + RelayCanaryVerdict::Inconclusive { + successes, + failures, + unavailable, + } => match policy { + RelayCanaryPolicy::Admission => warn!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + successes, + failures, + unavailable, + responses = summary.responses, + assumed_successes = summary.assumed_successes, + available_witnesses = summary.total, + ?policy, + "relay canary: completed witness round was inconclusive" + ), + RelayCanaryPolicy::Maintenance => info!( + relayer = %relayer.to_hex(), + relay = %relay_addr, + successes, + failures, + unavailable, + responses = summary.responses, + assumed_successes = summary.assumed_successes, + available_witnesses = summary.total, + ?policy, + "relay canary: completed maintenance round was inconclusive" + ), + }, + } + 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, + // Keep prospective canary probes separately identifiable in structured + // logs. Correctness comes from dialing the allocated socket and checking + // the authenticated target identity below. + transport.probe_relay_canary_authenticated(&relay_address), + ) + .await; + + let result = match dial { + Ok(Ok(authenticated_peer)) => { + if authenticated_peer == request.target_peer_id { + RelayCanaryProbeResult::Success + } else { + debug!( + expected = %request.target_peer_id.to_hex(), + actual = %authenticated_peer.to_hex(), + relay = %request.relay_addr, + "relay canary witness: identity mismatch" + ); + RelayCanaryProbeResult::Failure + } + } + Ok(Err(e)) => { + debug!( + relay = %request.relay_addr, + error = %e, + "relay canary witness: dial failed" + ); + RelayCanaryProbeResult::Failure + } + Err(_) => { + debug!( + relay = %request.relay_addr, + timeout = ?RELAY_CANARY_CONNECT_TIMEOUT, + "relay canary witness: dial timed out" + ); + RelayCanaryProbeResult::Failure + } + }; + + RelayCanaryResponse { result } +} + +fn select_relay_canary_witnesses( + mut candidates: Vec, + close_group_ids: &HashSet, + target_peer_id: &PeerId, + relayer: &PeerId, + relay_ip: IpAddr, + eligibility_epoch: u64, + 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) + || !relay_canary_witness_is_eligible(target_peer_id, &node.peer_id, eligibility_epoch) + { + 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; + let outcome = dht + .send_relay_canary_request( + &witness_peer_id, + &witness.typed_addresses, + request, + RELAY_CANARY_REQUEST_TIMEOUT, + ) + .await; + relay_canary_probe_report(witness_peer_id, outcome) +} + +fn relay_canary_probe_report( + witness: PeerId, + outcome: std::result::Result, +) -> RelayCanaryProbeReport { + match outcome { + Ok(RelayCanaryRequestOutcome::Response(response)) => RelayCanaryProbeReport { + witness, + disposition: response.result.disposition(), + detail: response.result.summary(), + }, + Ok(RelayCanaryRequestOutcome::NoProtocolResponse) => RelayCanaryProbeReport { + witness, + disposition: RelayCanaryProbeDisposition::AssumedSuccess, + detail: "no canary-protocol response; treating selected witness as legacy".to_string(), + }, + Err(error) => RelayCanaryProbeReport { + witness, + disposition: RelayCanaryProbeDisposition::Ineligible, + detail: error.to_string(), + }, + } +} + +#[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 TEST_PORT: u16 = 9000; + const TEST_RNG_SEED: u64 = 42; + const TEST_EPOCH: u64 = 1_234; + + fn peer_id(seed: u8) -> PeerId { + PeerId::from_bytes([seed; 32]) + } + + fn time_for_epoch(epoch: u64) -> SystemTime { + UNIX_EPOCH + Duration::from_secs(epoch * RELAY_CANARY_ELIGIBILITY_EPOCH_SECS + 1) + } + + fn eligible_witness(target: &PeerId, epoch: u64, start: u8) -> PeerId { + (start..=u8::MAX) + .map(peer_id) + .find(|candidate| relay_canary_witness_is_eligible(target, candidate, epoch)) + .expect("an eligible witness in the test search range") + } + + 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 eligible: Vec = (FIRST_WITNESS_SEED..=u8::MAX) + .filter(|seed| relay_canary_witness_is_eligible(&target, &peer_id(*seed), TEST_EPOCH)) + .take(5) + .collect(); + assert_eq!(eligible.len(), 5); + 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(eligible[0], Ipv4Addr::new(203, 0, 113, 3)), + node(eligible[1], Ipv4Addr::new(203, 0, 113, 4)), + node(eligible[2], Ipv4Addr::new(203, 0, 113, 5)), + node(eligible[3], Ipv4Addr::new(203, 0, 113, 3)), + node(eligible[4], 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), + TEST_EPOCH, + 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(eligible[4]))); + assert!(selected.contains(&peer_id(eligible[1]))); + assert!(selected.contains(&peer_id(eligible[2]))); + assert!( + selected + .iter() + .all(|peer| relay_canary_witness_is_eligible(&target, peer, TEST_EPOCH)) + ); + + let duplicate_pair_selected = + selected.contains(&peer_id(eligible[0])) && selected.contains(&peer_id(eligible[3])); + 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::Failure.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); + + // Four requests from a source are admitted, the immediate fifth is not. + for _ in 0..RELAY_CANARY_PEER_RATE_MAX_PER_WINDOW { + 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_inconclusive_verdict() { + let mut summary = RelayCanarySummary::new(RELAY_CANARY_WITNESS_TARGET); + + summary.record(RelayCanaryProbeDisposition::Success); + summary.record(RelayCanaryProbeDisposition::Ineligible); + summary.record(RelayCanaryProbeDisposition::Ineligible); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Admission), + RelayCanaryVerdict::Inconclusive { + successes: 1, + failures: 0, + unavailable: 2 + } + ); + } + + #[test] + fn one_failure_rejects_admission_but_not_healthy_maintenance_majority() { + let mut summary = RelayCanarySummary::new(RELAY_CANARY_WITNESS_TARGET); + + summary.record(RelayCanaryProbeDisposition::Failure); + summary.record(RelayCanaryProbeDisposition::Success); + summary.record(RelayCanaryProbeDisposition::Success); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Admission), + RelayCanaryVerdict::Rejected { + successes: 2, + attempts: 3 + } + ); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Maintenance), + RelayCanaryVerdict::Verified { + successes: 2, + attempts: 3 + } + ); + } + + #[test] + fn two_failures_reject_maintenance_round() { + let mut summary = RelayCanarySummary::new(RELAY_CANARY_WITNESS_TARGET); + + summary.record(RelayCanaryProbeDisposition::Failure); + summary.record(RelayCanaryProbeDisposition::Success); + summary.record(RelayCanaryProbeDisposition::Failure); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Maintenance), + RelayCanaryVerdict::Rejected { + successes: 1, + attempts: 3 + } + ); + } + + #[test] + fn split_maintenance_evidence_is_inconclusive() { + let mut summary = RelayCanarySummary::new(RELAY_CANARY_WITNESS_TARGET); + + summary.record(RelayCanaryProbeDisposition::Success); + summary.record(RelayCanaryProbeDisposition::Failure); + summary.record(RelayCanaryProbeDisposition::Ineligible); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Maintenance), + RelayCanaryVerdict::Inconclusive { + successes: 1, + failures: 1, + unavailable: 1 + } + ); + } + + #[test] + fn two_successes_and_one_ineligible_verify_only_maintenance() { + let mut summary = RelayCanarySummary::new(RELAY_CANARY_WITNESS_TARGET); + + summary.record(RelayCanaryProbeDisposition::Success); + summary.record(RelayCanaryProbeDisposition::Success); + summary.record(RelayCanaryProbeDisposition::Ineligible); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Admission), + RelayCanaryVerdict::Inconclusive { + successes: 2, + failures: 0, + unavailable: 1 + } + ); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Maintenance), + RelayCanaryVerdict::Verified { + successes: 2, + attempts: 2 + } + ); + } + + #[test] + fn missing_canary_response_counts_as_legacy_success() { + assert_eq!( + relay_canary_probe_report( + peer_id(FIRST_WITNESS_SEED), + Ok(RelayCanaryRequestOutcome::NoProtocolResponse) + ) + .disposition, + RelayCanaryProbeDisposition::AssumedSuccess + ); + } + + #[test] + fn witness_network_timeout_is_ineligible() { + assert_eq!( + relay_canary_probe_report( + peer_id(FIRST_WITNESS_SEED), + Err(P2PError::Network(NetworkError::Timeout)) + ) + .disposition, + RelayCanaryProbeDisposition::Ineligible + ); + } + + #[test] + fn assumed_legacy_successes_satisfy_admission_threshold() { + let mut summary = RelayCanarySummary::new(RELAY_CANARY_WITNESS_TARGET); + + summary.record(RelayCanaryProbeDisposition::Success); + summary.record(RelayCanaryProbeDisposition::AssumedSuccess); + summary.record(RelayCanaryProbeDisposition::AssumedSuccess); + + assert_eq!(summary.assumed_successes, 2); + assert_eq!( + summary.verdict(RelayCanaryPolicy::Admission), + RelayCanaryVerdict::Verified { + successes: 3, + attempts: 3 + } + ); + } + + #[test] + fn witness_contact_failure_is_ineligible() { + assert_eq!( + relay_canary_probe_report( + peer_id(FIRST_WITNESS_SEED), + Err(P2PError::Network(NetworkError::PeerNotFound( + "witness".into() + ))) + ) + .disposition, + RelayCanaryProbeDisposition::Ineligible + ); + } + + #[test] + fn canary_request_rejects_source_mismatch() { + let relay_addr = SocketAddr::from((Ipv4Addr::new(203, 0, 113, 7), TEST_PORT)); + let target = peer_id(TARGET_SEED); + let witness = eligible_witness(&target, TEST_EPOCH, FIRST_WITNESS_SEED); + let request = RelayCanaryRequest::new(target, relay_addr, TEST_EPOCH); + + let err = validate_relay_canary_request( + &peer_id(SECOND_WITNESS_SEED), + &witness, + &request, + time_for_epoch(TEST_EPOCH), + ) + .expect_err("source mismatch must be rejected"); + + assert!(matches!( + err, + RelayCanaryRequestRejection::SourceMismatch { .. } + )); + } + + #[test] + fn canary_request_accepts_current_and_previous_eligibility_epochs() { + let target = peer_id(TARGET_SEED); + let relay_addr = SocketAddr::from((Ipv4Addr::new(203, 0, 113, 7), TEST_PORT)); + let current_witness = eligible_witness(&target, TEST_EPOCH, FIRST_WITNESS_SEED); + let current = RelayCanaryRequest::new(target, relay_addr, TEST_EPOCH); + assert!( + validate_relay_canary_request( + &target, + ¤t_witness, + ¤t, + time_for_epoch(TEST_EPOCH), + ) + .is_ok() + ); + + let previous_epoch = TEST_EPOCH - 1; + let previous_witness = eligible_witness(&target, previous_epoch, FIRST_WITNESS_SEED); + let previous = RelayCanaryRequest::new(target, relay_addr, previous_epoch); + assert!( + validate_relay_canary_request( + &target, + &previous_witness, + &previous, + time_for_epoch(TEST_EPOCH), + ) + .is_ok() + ); + } + + #[test] + fn canary_request_rejects_stale_eligibility_epoch() { + let target = peer_id(TARGET_SEED); + let relay_addr = SocketAddr::from((Ipv4Addr::new(203, 0, 113, 8), TEST_PORT)); + let stale_epoch = TEST_EPOCH - 2; + let witness = eligible_witness(&target, stale_epoch, FIRST_WITNESS_SEED); + let request = RelayCanaryRequest::new(target, relay_addr, stale_epoch); + + let error = + validate_relay_canary_request(&target, &witness, &request, time_for_epoch(TEST_EPOCH)) + .expect_err("stale witness assignment must be rejected"); + + assert!(matches!( + error, + RelayCanaryRequestRejection::StaleEligibilityEpoch { .. } + )); + } + + #[test] + fn canary_request_rejects_unassigned_witness() { + let target = peer_id(TARGET_SEED); + let relay_addr = SocketAddr::from((Ipv4Addr::new(203, 0, 113, 8), TEST_PORT)); + let witness = (FIRST_WITNESS_SEED..=u8::MAX) + .map(peer_id) + .find(|candidate| !relay_canary_witness_is_eligible(&target, candidate, TEST_EPOCH)) + .expect("an ineligible witness in the test search range"); + let request = RelayCanaryRequest::new(target, relay_addr, TEST_EPOCH); + + assert!(matches!( + validate_relay_canary_request(&target, &witness, &request, time_for_epoch(TEST_EPOCH),), + Err(RelayCanaryRequestRejection::IneligibleWitness { .. }) + )); + } + + #[test] + fn canary_request_rejects_local_scope_relay_address() { + let target = peer_id(TARGET_SEED); + let relay_addr = SocketAddr::from((Ipv4Addr::new(192, 168, 1, 10), TEST_PORT)); + let witness = eligible_witness(&target, TEST_EPOCH, FIRST_WITNESS_SEED); + let request = RelayCanaryRequest::new(target, relay_addr, TEST_EPOCH); + + let err = + validate_relay_canary_request(&target, &witness, &request, time_for_epoch(TEST_EPOCH)) + .expect_err("private relay address must be rejected"); + + assert!(matches!(err, RelayCanaryRequestRejection::LocalScopeIp(_))); + } + + #[test] + fn canary_request_rejects_unspecified_port() { + let target = peer_id(TARGET_SEED); + let relay_addr = SocketAddr::from((Ipv4Addr::new(203, 0, 113, 8), UNSPECIFIED_PORT)); + let witness = eligible_witness(&target, TEST_EPOCH, FIRST_WITNESS_SEED); + let request = RelayCanaryRequest::new(target, relay_addr, TEST_EPOCH); + + let err = + validate_relay_canary_request(&target, &witness, &request, time_for_epoch(TEST_EPOCH)) + .expect_err("port zero must be rejected"); + + assert_eq!(err, RelayCanaryRequestRejection::UnspecifiedPort); + } + + #[test] + fn source_network_buckets_ipv6_by_prefix_and_ipv4_by_address() { + let first_v6: IpAddr = "2001:db8:1234:5678::1".parse().expect("IPv6 address"); + let second_v6: IpAddr = "2001:db8:1234:5678::ffff".parse().expect("IPv6 address"); + let other_v6: IpAddr = "2001:db8:1234:5679::1".parse().expect("IPv6 address"); + assert_eq!( + relay_canary_source_network(first_v6), + relay_canary_source_network(second_v6) + ); + assert_ne!( + relay_canary_source_network(first_v6), + relay_canary_source_network(other_v6) + ); + + let ipv4: IpAddr = "203.0.113.9".parse().expect("IPv4 address"); + assert_eq!(relay_canary_source_network(ipv4), ipv4); + } +} diff --git a/src/reachability/driver.rs b/src/reachability/driver.rs index 52c541d9..f844c609 100644 --- a/src/reachability/driver.rs +++ b/src/reachability/driver.rs @@ -16,33 +16,43 @@ //! Owns every state transition for this node's MASQUE relay: the initial //! acquisition at startup, the backoff retry when no candidate accepts, //! the republish-then-reacquire sequence when an existing relay is lost, -//! and the K-closest-eviction watcher that forces a rebind when the -//! chosen relayer drops out of the close group. +//! and the health checks that retain a verified relay until there is positive +//! evidence that it is no longer usable. //! //! ## State machine //! -//! The driver runs as a single tokio task and cycles through three states: +//! The driver runs as a single tokio task and cycles through five 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. -//! 2. **Holding**: subscribe to `KClosestPeersChanged` events, republish -//! when a pinned external address is promoted to +//! 1. **Starting**: publish a newer relay-free address set before the first +//! acquisition walk. This withdraws any relay allocation left in DHT +//! replicas by a previous process incarnation before peers can dial it. +//! 2. **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. +//! 3. **Holding**: republish when a pinned external address is promoted to //! [`AddressType::Direct`], and poll //! [`TransportHandle::is_relay_healthy`] every -//! [`HEALTH_POLL_INTERVAL`]. On relayer-evicted or unhealthy-tunnel, -//! transition to **Lost**; on shutdown, exit the driver. -//! 3. **Lost**: run the `republish-direct-only → reacquire` sequence. +//! [`HEALTH_POLL_INTERVAL`]. The driver also repeats the independent +//! third-party canary quorum every [`RELAY_REVALIDATION_INTERVAL`]. The +//! maintenance cadence is deliberately much slower than local health +//! polling: admission already proved external reachability, while each +//! maintenance round creates six network operations (three witness requests +//! and three fresh relay dials). K-closest churn does not invalidate an +//! already verified relay. On an unhealthy tunnel or failed revalidation, +//! transition to **Lost**; on shutdown, exit. +//! 4. **Lost**: run the `republish-direct-only → reacquire` sequence. //! The republish MUST happen **before** the acquisition walk starts, //! so the network stops dialing the dead relay address during the //! 1–10 s acquisition window. After republishing, loop back to //! **Acquiring**. -//! 4. **Backoff**: wait for the current backoff window or a +//! 5. **Backoff**: wait for the current backoff window or a //! `KClosestPeersChanged` event (whichever comes first), republishing //! if a pinned external is promoted to [`AddressType::Direct`] while //! waiting, then loop back to **Acquiring**. Successful acquisition @@ -52,6 +62,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,18 +74,39 @@ use tracing::{debug, info, trace, warn}; use crate::dht::AddressType; use crate::dht_network_manager::{DhtNetworkEvent, DhtNetworkManager}; +use crate::reachability::canary::{ + RelayCanaryPolicy, 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; use crate::{MultiAddr, PeerId}; +use saorsa_transport::nat_traversal_api::PreparedRelay; /// How often to poll the transport for tunnel health while holding a relay. /// /// 5 seconds fits inside the 10–30 s failover-window budget and keeps the -/// wake rate low (the poll is non-blocking and only reads an atomic -/// counter inside saorsa-transport). +/// wake rate low. const HEALTH_POLL_INTERVAL: Duration = Duration::from_secs(5); +/// How often an established relay must again pass a third-party canary quorum. +/// +/// Local task health only proves that this process still has a tunnel. A +/// relay-server forwarding regression or expired public allocation can leave +/// that tunnel locally alive but externally unreachable, so retain a periodic +/// external check. Two hours avoids turning a large fleet into a continuous +/// source of witness dials and one-shot PQC handshakes. +const RELAY_REVALIDATION_INTERVAL: Duration = Duration::from_secs(2 * 60 * 60); + +/// Maximum deterministic per-peer offset added before the first maintenance +/// canary. Spreading the first round across a full interval avoids a fleet-wide +/// probe burst after a rolling deployment. +const RELAY_REVALIDATION_JITTER_MAX: Duration = RELAY_REVALIDATION_INTERVAL; + +/// Retry interval for authoritative address publications that were not +/// acknowledged by every current close peer. +const PUBLISH_RETRY_INTERVAL: Duration = Duration::from_secs(15); + /// Initial delay before the first retry after a failed acquisition walk. const BACKOFF_INITIAL: Duration = Duration::from_secs(30); @@ -111,6 +143,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,51 +160,222 @@ struct AcquisitionDriver { shutdown: CancellationToken, current_backoff: Duration, last_published_typed_set: Option, + canary_rejected_relayers: HashSet, } #[derive(Clone, Debug, PartialEq)] struct PublishedTypedSet { typed_addresses: Vec<(MultiAddr, AddressType)>, - peers: Vec, + target_peers: HashSet, + pending_peers: HashSet, +} + +fn pending_publication_targets( + previous: Option<&PublishedTypedSet>, + typed_addresses: &[(MultiAddr, AddressType)], + target_peers: &HashSet, + force: bool, +) -> HashSet { + let Some(previous) = previous + .filter(|previous| !force && previous.typed_addresses.as_slice() == typed_addresses) + else { + return target_peers.clone(); + }; + + let mut pending: HashSet<_> = previous + .pending_peers + .intersection(target_peers) + .copied() + .collect(); + pending.extend(target_peers.difference(&previous.target_peers).copied()); + pending +} + +#[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"); + + // A process restart invalidates every relay allocation owned by the + // previous process, while DHT replicas can still retain its last + // sequenced self-record. Publish a newer relay-free full replacement + // before attempting to acquire another relay. Without this startup + // tombstone, peers can keep dialing an allocation whose tunnel died + // with the old process. + self.transport.clear_relay_address(); + self.force_publish_typed_set(None).await; + loop { if self.shutdown.is_cancelled() { debug!("relay acquisition driver: shutdown, exiting"); 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; + let relay_addr = relay.allocation.public_addr(); + match verify_relay_with_canaries( + &self.dht, + relay.relayer, + relay_addr, + RelayCanaryPolicy::Admission, + ) + .await + { + RelayCanaryVerdict::Verified { + successes, + attempts, + } => { + if let Err(error) = self + .transport + .publish_proactive_relay_session(relay.allocation) + .await + { + warn!( + relayer = ?relay.relayer, + allocated = %relay_addr, + %error, + "driver: failed to commit canary-verified relay" + ); + apply_canary_rejection_event( + &mut self.canary_rejected_relayers, + CanaryRejectionEvent::AcquisitionFailed, + ); + self.clear_unpublished_relay_state(relay.allocation).await; + self.publish_typed_set(None).await; + if self.wait_backoff_or_event().await { + return; + } + self.advance_backoff(); + continue; + } + 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_addr); + self.transport.set_relay_address(relay_addr); + self.force_publish_typed_set(Some(relay_addr)).await; + info!( + relayer = ?relay.relayer, + allocated = %relay_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(relay.allocation).await; + } + RelayCanaryVerdict::Rejected { + successes, + attempts, + } => { + warn!( + relayer = ?relay.relayer, + allocated = %relay_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(relay.allocation).await; + self.publish_typed_set(None).await; + if self.wait_backoff_or_event().await { + return; // shutdown + } + self.advance_backoff(); + } + RelayCanaryVerdict::Inconclusive { + successes, + failures, + unavailable, + } => { + warn!( + relayer = ?relay.relayer, + allocated = %relay_addr, + successes, + failures, + unavailable, + "driver: relay canary evidence inconclusive, entering backoff without publishing relay" + ); + apply_canary_rejection_event( + &mut self.canary_rejected_relayers, + CanaryRejectionEvent::InsufficientWitnesses, + ); + self.clear_unpublished_relay_state(relay.allocation).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 +422,25 @@ impl AcquisitionDriver { self.publish_typed_set_with_policy(relay, true).await; } + /// Tear down and clear a relay allocation that never passed canary. + async fn clear_unpublished_relay_state(&mut self, allocation: PreparedRelay) { + let relay_public_addr = allocation.public_addr(); + if let Err(error) = self + .transport + .abort_proactive_relay_session(allocation) + .await + { + warn!( + relay_addr = %relay_public_addr, + %error, + "driver: failed to abort unpublished relay" + ); + } + *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(); @@ -233,59 +456,101 @@ impl AcquisitionDriver { self.transport.is_external_proven(sa) }); - if self_addresses.is_empty() { + if self_addresses.is_empty() && !force { debug!("driver: publish skipped, no dialable self addresses"); return; } let typed = self_addresses.into_typed_vec(); + if typed.is_empty() { + info!( + "driver: publishing empty authoritative address set to withdraw stale relay state" + ); + } let own_key = *self.dht.peer_id().to_bytes(); let all_peers = self .dht .find_closest_nodes_local(&own_key, self.dht.k_value()) .await; - let peers = all_peers.iter().map(|node| node.peer_id).collect(); - let publish_snapshot = PublishedTypedSet { - typed_addresses: typed.clone(), - peers, - }; - if !force && self.last_published_typed_set.as_ref() == Some(&publish_snapshot) { + let target_peers: HashSet = all_peers + .iter() + .map(|node| node.peer_id) + .filter(|peer| peer != self.dht.peer_id()) + .collect(); + let mut pending_peers = pending_publication_targets( + self.last_published_typed_set.as_ref(), + &typed, + &target_peers, + force, + ); + if pending_peers.is_empty() { debug!( peers = all_peers.len(), typed_addresses = ?typed, relay = ?relay, "driver: publish skipped, typed self address set unchanged" ); + self.last_published_typed_set = Some(PublishedTypedSet { + typed_addresses: typed, + target_peers, + pending_peers, + }); return; } + let peers_to_publish: Vec<_> = all_peers + .into_iter() + .filter(|peer| pending_peers.contains(&peer.peer_id)) + .collect(); + debug!( - peers = all_peers.len(), + peers = peers_to_publish.len(), typed_addresses = ?typed, relay = ?relay, "driver: publishing typed self address set" ); trace!( - peers = all_peers.len(), + peers = peers_to_publish.len(), addrs = typed.len(), relay = ?relay, "driver: publishing typed address set to all routing table peers" ); - self.dht - .publish_address_set_to_peers(typed, &all_peers) + let confirmed = self + .dht + .publish_address_set_to_peers(typed.clone(), &peers_to_publish) .await; - self.last_published_typed_set = Some(publish_snapshot); + for peer in confirmed { + pending_peers.remove(&peer); + } + let missing = pending_peers.len(); + if missing > 0 { + debug!( + missing, + targets = target_peers.len(), + relay = ?relay, + "driver: address publication incomplete; unacknowledged peers will be retried" + ); + } + self.last_published_typed_set = Some(PublishedTypedSet { + typed_addresses: typed, + target_peers, + pending_peers, + }); } - /// Hold the acquired relay until an eviction or death event forces a - /// rebind. Returns `true` on shutdown (caller should exit), `false` - /// when the relay is considered lost and a republish+reacquire is - /// needed. + /// Hold the acquired relay until positive failure evidence forces a rebind. + /// + /// Returns `true` on shutdown (caller should exit), `false` when the relay + /// is considered lost and a republish+reacquire is needed. async fn hold_until_lost(&mut self) -> bool { let mut events = self.dht.subscribe_events(); let mut health = tokio::time::interval(HEALTH_POLL_INTERVAL); health.tick().await; // drop the immediate first tick + let first_revalidation = + tokio::time::Instant::now() + relay_revalidation_initial_delay(self.dht.peer_id()); + let revalidation = tokio::time::sleep_until(first_revalidation); + tokio::pin!(revalidation); loop { tokio::select! { @@ -353,11 +618,18 @@ impl AcquisitionDriver { } event = events.recv() => { match event { - Ok(DhtNetworkEvent::KClosestPeersChanged { ref new, .. }) => { - if self.relayer_evicted_from_k_closest(new).await { - info!("driver: relayer evicted from K-closest, rebinding"); - return false; - } + Ok(DhtNetworkEvent::KClosestPeersChanged { + added, + removed, + .. + }) => { + let relay = *self.relay_address.read().await; + self.publish_typed_set(relay).await; + debug!( + added = added.len(), + removed = removed.len(), + "driver: K-closest changed; published current relay state only to new targets" + ); } Ok(_) => continue, // `RecvError::Lagged` is recoverable — the broadcast @@ -366,39 +638,119 @@ impl AcquisitionDriver { // terminal (the DHT manager is dropping); treat it // the same as shutdown. Err(RecvError::Closed) => return true, - Err(_) => continue, + Err(RecvError::Lagged(skipped)) => { + self.last_published_typed_set = None; + let relay = *self.relay_address.read().await; + self.publish_typed_set(relay).await; + debug!( + skipped, + "driver: refreshed publication after lagging DHT events" + ); + } } } _ = health.tick() => { - if !self.transport.is_relay_healthy() { + if !self.transport.is_relay_healthy().await { info!("driver: relay tunnel unhealthy, rebinding"); return false; } + // Also retries any peers that did not acknowledge the + // latest full address-set publication. + let relay = *self.relay_address.read().await; + self.publish_typed_set(relay).await; + } + _ = &mut revalidation => { + let relayer = *self.relayer_peer_id.read().await; + let relay = *self.relay_address.read().await; + let (Some(relayer), Some(relay)) = (relayer, relay) else { + warn!("driver: relay state disappeared before revalidation"); + return false; + }; + let verdict = verify_relay_with_canaries( + &self.dht, + relayer, + relay, + RelayCanaryPolicy::Maintenance, + ) + .await; + let retry_delay = match verdict { + RelayCanaryVerdict::Verified { successes, attempts } => { + info!( + relayer = %relayer.to_hex(), + relay = %relay, + successes, + attempts, + "driver: established relay passed periodic canary revalidation" + ); + RELAY_REVALIDATION_INTERVAL + } + RelayCanaryVerdict::Rejected { successes, attempts } => { + warn!( + relayer = %relayer.to_hex(), + relay = %relay, + successes, + attempts, + "driver: established relay failed periodic canary revalidation; withdrawing" + ); + apply_canary_rejection_event( + &mut self.canary_rejected_relayers, + CanaryRejectionEvent::Rejected(relayer), + ); + return false; + } + RelayCanaryVerdict::Inconclusive { + successes, + failures, + unavailable, + } => { + info!( + relayer = %relayer.to_hex(), + relay = %relay, + successes, + failures, + unavailable, + "driver: established relay canary evidence inconclusive; retaining relay until the next scheduled check" + ); + // Missing witnesses and transient request failures + // are not evidence that the established relay is + // bad. Retain it and wait for the ordinary cadence; + // retrying a three-witness round after 15 seconds + // amplified partial outages into sustained dial + // storms. + RELAY_REVALIDATION_INTERVAL + } + }; + revalidation + .as_mut() + .reset(tokio::time::Instant::now() + retry_delay); } } } } - /// Returns `true` if the currently-chosen relayer is no longer in the - /// new K-closest set. - async fn relayer_evicted_from_k_closest(&self, new_k_closest: &[PeerId]) -> bool { - let guard = self.relayer_peer_id.read().await; - let Some(relayer) = guard.as_ref() else { - return false; - }; - !new_k_closest.contains(relayer) - } - /// Transition out of the Holding state: republish direct-only and /// clear relayer state, BEFORE the acquisition walk retries. The /// pre-retry publish is critical — without it, other peers would /// continue dialing the dead relay address during the 1–10 s /// acquisition walk. - async fn lose_relay_and_republish(&mut self) { + async fn lose_relay_and_republish(&mut self, allocation: PreparedRelay) { + let relay_public_addr = self.relay_address.write().await.take(); *self.relayer_peer_id.write().await = None; - *self.relay_address.write().await = None; self.transport.clear_relay_address(); - self.force_publish_typed_set(None).await; + + // Withdrawal and transport teardown start together. Peers are told to + // stop using the allocation without waiting for local QUIC/MASQUE + // shutdown, while teardown does not wait on DHT acknowledgements. + let transport = Arc::clone(&self.transport); + let teardown = async move { transport.abort_proactive_relay_session(allocation).await }; + let (teardown_result, ()) = tokio::join!(teardown, self.force_publish_typed_set(None)); + if let Err(error) = teardown_result { + warn!( + relay_addr = ?relay_public_addr, + %error, + "driver: failed to tear down lost or evicted relay" + ); + } } /// Wait out the current backoff window, or short-circuit on a @@ -408,6 +760,8 @@ impl AcquisitionDriver { let mut events = self.dht.subscribe_events(); let sleep = tokio::time::sleep(self.current_backoff); tokio::pin!(sleep); + let mut publish_retry = tokio::time::interval(PUBLISH_RETRY_INTERVAL); + publish_retry.tick().await; loop { tokio::select! { @@ -417,6 +771,9 @@ impl AcquisitionDriver { trace!(window = ?self.current_backoff, "driver: backoff window expired"); return false; } + _ = publish_retry.tick() => { + self.publish_typed_set(None).await; + } promoted = self.transport.recv_direct_address_promoted() => { match promoted { Some(addr) => { @@ -450,12 +807,20 @@ impl AcquisitionDriver { event = events.recv() => { match event { Ok(DhtNetworkEvent::KClosestPeersChanged { .. }) => { + self.last_published_typed_set = None; debug!("driver: K-closest changed, retrying early"); return false; } Ok(_) => continue, Err(RecvError::Closed) => return true, - Err(_) => continue, + Err(RecvError::Lagged(skipped)) => { + self.last_published_typed_set = None; + self.publish_typed_set(None).await; + debug!( + skipped, + "driver: refreshed publication after lagging DHT events during backoff" + ); + } } } } @@ -468,3 +833,140 @@ impl AcquisitionDriver { self.current_backoff = next.min(BACKOFF_MAX); } } + +fn relay_revalidation_initial_delay(peer_id: &PeerId) -> Duration { + let mut prefix = [0u8; std::mem::size_of::()]; + prefix.copy_from_slice(&peer_id.to_bytes()[..std::mem::size_of::()]); + let jitter_bound = RELAY_REVALIDATION_JITTER_MAX.as_secs().saturating_add(1); + let jitter = u64::from_be_bytes(prefix) % jitter_bound; + RELAY_REVALIDATION_INTERVAL.saturating_add(Duration::from_secs(jitter)) +} + +#[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 publication_targets_only_retry_pending_and_new_peers() { + let departed = peer_id(1); + let retained = peer_id(2); + let joined = peer_id(3); + let previous = PublishedTypedSet { + typed_addresses: Vec::new(), + target_peers: HashSet::from([departed, retained]), + pending_peers: HashSet::from([departed]), + }; + let current = HashSet::from([retained, joined]); + + assert_eq!( + pending_publication_targets(Some(&previous), &[], ¤t, false), + HashSet::from([joined]) + ); + } + + #[test] + fn invalidated_publication_retries_rejoined_peer_id() { + let peer = peer_id(1); + let current = HashSet::from([peer]); + + assert_eq!( + pending_publication_targets(None, &[], ¤t, false), + current + ); + } + + #[test] + fn changed_or_forced_publication_targets_every_current_peer() { + let first = peer_id(1); + let second = peer_id(2); + let previous = PublishedTypedSet { + typed_addresses: Vec::new(), + target_peers: HashSet::from([first, second]), + pending_peers: HashSet::new(), + }; + let current = HashSet::from([first, second]); + let changed = [( + MultiAddr::from_ipv4(std::net::Ipv4Addr::new(203, 0, 113, 7), 9000), + AddressType::Direct, + )]; + + assert_eq!( + pending_publication_targets(Some(&previous), &changed, ¤t, false), + current + ); + assert_eq!( + pending_publication_targets(Some(&previous), &[], ¤t, true), + current + ); + } + + #[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)); + } + + #[test] + fn relay_revalidation_delay_is_bounded_and_peer_stable() { + let peer = peer_id(REJECTED_RELAYER_SEED); + let delay = relay_revalidation_initial_delay(&peer); + + assert_eq!(delay, relay_revalidation_initial_delay(&peer)); + assert!(delay >= RELAY_REVALIDATION_INTERVAL); + assert!(delay <= RELAY_REVALIDATION_INTERVAL.saturating_add(RELAY_REVALIDATION_JITTER_MAX)); + } +} 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..a2c3ee0f 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); @@ -136,7 +149,7 @@ pub(crate) async fn run_relay_acquisition( Ok(relay) => { info!( relayer = ?relay.relayer, - allocated = %relay.allocated_public_addr, + allocated = %relay.allocation.public_addr(), "relay acquisition: session established" ); RelayAcquisitionOutcome::Acquired(relay) diff --git a/src/transport/saorsa_transport_adapter.rs b/src/transport/saorsa_transport_adapter.rs index 7dc53ee6..2e937f1a 100644 --- a/src/transport/saorsa_transport_adapter.rs +++ b/src/transport/saorsa_transport_adapter.rs @@ -54,6 +54,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, info, trace}; // Import saorsa-transport types using the new LinkTransport API (0.14+) +use saorsa_transport::nat_traversal_api::PreparedRelay; use saorsa_transport::{ LinkConn, LinkEvent, LinkTransport, NatConfig, P2pConfig, P2pLinkTransport, ProtocolId, Side, StrategyConfig, @@ -97,6 +98,16 @@ pub enum ConnectionEvent { }, } +/// Result of an outbound dial, including the identity authenticated by the +/// exact QUIC/TLS connection. +#[derive(Debug, Clone)] +pub(crate) struct DialedPeer { + /// Normalized remote socket address. + pub(crate) remote_addr: SocketAddr, + /// Authenticated ML-DSA-65 SubjectPublicKeyInfo from the TLS handshake. + pub(crate) peer_public_key_spki: Option>, +} + /// Native saorsa-transport network node using LinkTransport abstraction /// /// This provides a clean interface to saorsa-transport's peer-to-peer networking @@ -169,7 +180,7 @@ const DIRECT_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(4); /// forwarder fails to push into a bounded channel. /// /// Used by the forwarder loop in -/// [`DualStackNetworkNode::spawn_peer_address_update_forwarder`] when the +/// [`DualStackNetworkNode::spawn_address_event_forwarder`] when the /// downstream consumer is too slow to drain. Drops are coalesced to one /// warning per [`ADDRESS_EVENT_DROP_LOG_INTERVAL`] events to avoid log /// floods under sustained backpressure; the very first drop in any burst @@ -622,8 +633,14 @@ impl P2PNetworkNode { /// and abort the per-connection reader task, then removes the peer from /// the local registry. pub async fn disconnect_peer_quic(&self, addr: &SocketAddr) { - if let Err(e) = self.transport.endpoint().disconnect(addr).await { - tracing::warn!("QUIC disconnect for peer {}: {}", addr, e); + match self.transport.endpoint().disconnect(addr).await { + Ok(()) => {} + Err(saorsa_transport::p2p_endpoint::EndpointError::PeerNotFound(_)) => { + // Connection-loss handling and explicit cleanup race by + // design. If the loss path won, disconnect is already done. + tracing::debug!("QUIC peer {} was already disconnected", addr); + } + Err(e) => tracing::warn!("QUIC disconnect for peer {}: {}", addr, e), } // Also clean up from generic adapter state P2PNetworkNode::::disconnect_peer_inner( @@ -837,6 +854,16 @@ impl P2PNetworkNode { /// Connect to a peer by address pub async fn connect_to_peer(&self, peer_addr: SocketAddr) -> Result { + self.connect_to_peer_authenticated(peer_addr) + .await + .map(|peer| peer.remote_addr) + } + + /// Connect to a peer and retain the identity authenticated by this dial. + pub(crate) async fn connect_to_peer_authenticated( + &self, + peer_addr: SocketAddr, + ) -> Result { // saorsa-core publishes typed addresses (Direct or Relay-allocated) // and is responsible for picking the right one before reaching the // transport. Direct addresses are self-asserted by the publisher, @@ -855,6 +882,7 @@ impl P2PNetworkNode { .map_err(|e| anyhow::anyhow!("Failed to connect to peer {}: {}", peer_addr, e))?; let remote_addr = conn.remote_addr(); + let peer_public_key_spki = conn.peer_public_key(); // Register the peer with geographic validation self.add_peer(remote_addr).await; @@ -863,7 +891,10 @@ impl P2PNetworkNode { // to avoid duplicate events info!("Connected to peer at {}", remote_addr); - Ok(remote_addr) + Ok(DialedPeer { + remote_addr, + peer_public_key_spki, + }) } /// Try to accept one incoming connection. @@ -1375,10 +1406,10 @@ impl DualStackNetworkNode { /// Returns `true` if no relay was established or the relay is healthy. /// Returns `false` if a relay was established but the QUIC connection /// has closed — the relayer monitor should trigger rebinding. - pub fn is_relay_healthy(&self) -> bool { + pub async fn is_relay_healthy(&self) -> bool { // If ANY stack reports an unhealthy relay, the relay is dead. for node in [&self.v6, &self.v4].into_iter().flatten() { - if !node.transport.endpoint().is_relay_healthy() { + if !node.transport.endpoint().is_relay_healthy().await { return false; } } @@ -1417,6 +1448,20 @@ impl DualStackNetworkNode { &self, relay_addr: SocketAddr, ) -> std::result::Result { + let allocated = self.prepare_proactive_relay(relay_addr).await?; + if let Err(error) = self.publish_proactive_relay(allocated).await { + let _ = self.abort_proactive_relay(allocated).await; + return Err(error); + } + Ok(allocated.public_addr()) + } + + /// Prepare a proactive MASQUE relay without advertising its allocated + /// address. The relay endpoint is live for inbound canary probes. + pub async fn prepare_proactive_relay( + &self, + relay_addr: SocketAddr, + ) -> std::result::Result { let node = if relay_addr.is_ipv4() { self.v4.as_ref().or(self.v6.as_ref()) } else { @@ -1429,9 +1474,93 @@ impl DualStackNetworkNode { )) })?; + let allocation = node + .transport + .endpoint() + .prepare_proactive_relay(relay_addr) + .await?; + if allocation.public_addr().is_ipv4() != relay_addr.is_ipv4() { + let allocated_addr = allocation.public_addr(); + let _ = node + .transport + .endpoint() + .abort_proactive_relay(allocation) + .await; + return Err(saorsa_transport::p2p_endpoint::EndpointError::Config( + format!( + "relay allocation address family mismatch: requested {relay_addr}, allocated {allocated_addr}" + ), + )); + } + Ok(allocation) + } + + /// Publish a previously prepared proactive relay. + pub async fn publish_proactive_relay( + &self, + allocation: PreparedRelay, + ) -> std::result::Result<(), saorsa_transport::p2p_endpoint::EndpointError> { + let relay_public_addr = allocation.public_addr(); + let node = if relay_public_addr.is_ipv4() { + self.v4.as_ref().or(self.v6.as_ref()) + } else { + self.v6.as_ref().or(self.v4.as_ref()) + } + .ok_or_else(|| { + saorsa_transport::p2p_endpoint::EndpointError::Config(format!( + "no transport stack available for relay address family {}", + relay_public_addr + )) + })?; + + node.transport + .endpoint() + .publish_proactive_relay(allocation) + .await + } + + /// Open one isolated authenticated QUIC probe on the matching stack. + pub async fn probe_fresh_authenticated( + &self, + target: SocketAddr, + ) -> std::result::Result, saorsa_transport::p2p_endpoint::EndpointError> { + let node = if target.is_ipv4() { + self.v4.as_ref().or(self.v6.as_ref()) + } else { + self.v6.as_ref().or(self.v4.as_ref()) + } + .ok_or_else(|| { + saorsa_transport::p2p_endpoint::EndpointError::Config(format!( + "no transport stack available for probe address family {target}" + )) + })?; + node.transport + .endpoint() + .probe_fresh_authenticated(target) + .await + } + + /// Abort a proactive relay allocation and release its transport resources. + pub async fn abort_proactive_relay( + &self, + allocation: PreparedRelay, + ) -> std::result::Result<(), saorsa_transport::p2p_endpoint::EndpointError> { + let relay_public_addr = allocation.public_addr(); + let node = if relay_public_addr.is_ipv4() { + self.v4.as_ref().or(self.v6.as_ref()) + } else { + self.v6.as_ref().or(self.v4.as_ref()) + } + .ok_or_else(|| { + saorsa_transport::p2p_endpoint::EndpointError::Config(format!( + "no transport stack available for relay address family {}", + relay_public_addr + )) + })?; + node.transport .endpoint() - .setup_proactive_relay(relay_addr) + .abort_proactive_relay(allocation) .await } @@ -1453,21 +1582,17 @@ impl DualStackNetworkNode { /// Spawn background tasks that forward address-related `P2pEvent`s from /// each stack's `P2pEndpoint` to the upper layers. /// - /// Four transport event flavours are bridged, and direct-address + /// Three transport event flavours are bridged, and direct-address /// promotion notifications are emitted when the classifier state crosses /// its proof threshold: /// - /// - **`PeerAddressUpdated`**: a connected peer advertised a new - /// reachable address via an ADD_ADDRESS frame (typically a relay). - /// Returned via the first mpsc receiver as - /// `(peer_connection_addr, advertised_addr)`. /// - **`RelayEstablished`**: this node set up a MASQUE relay and now /// needs to publish the relay address to the K closest peers. - /// Returned via the second mpsc receiver. + /// Returned via the first mpsc receiver. /// - **`RelayLost`**: a previously-advertised MASQUE relay address is /// no longer reachable. The reachability driver republishes the /// address set without the relay entry on receipt. Returned via - /// the third mpsc receiver. + /// the second mpsc receiver. /// - **`ExternalAddressDiscovered`**: saorsa-transport's observed /// address quorum cleared. The address is pinned into the supplied /// [`ExternalAddresses`] store and a self-address update is emitted @@ -1479,7 +1604,7 @@ impl DualStackNetworkNode { /// /// Other `P2pEvent` variants are not consumed by saorsa-core and are /// silently ignored. - pub fn spawn_peer_address_update_forwarder( + pub fn spawn_address_event_forwarder( &self, external_addresses: Arc>, peer_observations: Arc>>, @@ -1488,18 +1613,15 @@ impl DualStackNetworkNode { direct_promoted_events: AddressEventPublisher, self_address_updated_events: AddressEventPublisher, ) -> ( - tokio::sync::mpsc::Receiver<(SocketAddr, SocketAddr)>, tokio::sync::mpsc::Receiver, tokio::sync::mpsc::Receiver, ) { - let (tx, rx) = tokio::sync::mpsc::channel(ADDRESS_EVENT_CHANNEL_CAPACITY); let (relay_tx, relay_rx) = tokio::sync::mpsc::channel(ADDRESS_EVENT_CHANNEL_CAPACITY); let (relay_lost_tx, relay_lost_rx) = tokio::sync::mpsc::channel(ADDRESS_EVENT_CHANNEL_CAPACITY); let drop_counter = Arc::new(AtomicU64::new(0)); for node in [&self.v6, &self.v4].into_iter().flatten() { let mut p2p_rx = node.transport.endpoint().subscribe(); - let tx_clone = tx.clone(); let relay_tx_clone = relay_tx.clone(); let relay_lost_tx_clone = relay_lost_tx.clone(); let ext_clone = Arc::clone(&external_addresses); @@ -1517,23 +1639,6 @@ impl DualStackNetworkNode { ); loop { match p2p_rx.recv().await { - Ok(saorsa_transport::P2pEvent::PeerAddressUpdated { - peer_addr, - advertised_addr, - }) => { - tracing::debug!( - "ADDR_FWD: received PeerAddressUpdated peer={} addr={}", - peer_addr, - advertised_addr - ); - let payload = ( - saorsa_transport::shared::normalize_socket_addr(peer_addr), - saorsa_transport::shared::normalize_socket_addr(advertised_addr), - ); - if let Err(err) = tx_clone.try_send(payload) { - handle_address_event_drop(&drops, "PeerAddressUpdated", &err); - } - } Ok(saorsa_transport::P2pEvent::RelayEstablished { relay_addr }) => { tracing::info!( "ADDR_FWD: received RelayEstablished relay_addr={}", @@ -1619,7 +1724,7 @@ impl DualStackNetworkNode { } }); } - (rx, relay_rx, relay_lost_rx) + (relay_rx, relay_lost_rx) } /// Spawn one background task per bound stack (v4, v6) to classify @@ -2105,13 +2210,26 @@ impl DualStackNetworkNode { /// /// The returned address is always normalised (plain IPv4). pub async fn connect_happy_eyeballs(&self, targets: &[SocketAddr]) -> Result { + self.connect_happy_eyeballs_authenticated(targets) + .await + .map(|peer| peer.remote_addr) + } + + /// Happy Eyeballs connect retaining the winning connection's TLS identity. + pub(crate) async fn connect_happy_eyeballs_authenticated( + &self, + targets: &[SocketAddr], + ) -> Result { if self.is_dual_stack { let dial_list = to_dual_stack_dial_list(targets); if dial_list.is_empty() { return Err(anyhow::anyhow!("No suitable transport available")); } - let addr = self.connect_sequential(&self.v6, &dial_list).await?; - return Ok(self.normalize(addr)); + let mut peer = self + .connect_sequential_authenticated(&self.v6, &dial_list) + .await?; + peer.remote_addr = self.normalize(peer.remote_addr); + return Ok(peer); } let (v6_targets, v4_targets) = bucket_targets(targets); @@ -2119,12 +2237,18 @@ impl DualStackNetworkNode { let (v6_node, v4_node) = match (&self.v6, &self.v4) { (Some(v6), Some(v4)) if !v6_targets.is_empty() && !v4_targets.is_empty() => (v6, v4), (Some(_), _) if !v6_targets.is_empty() => { - let addr = self.connect_sequential(&self.v6, &v6_targets).await?; - return Ok(self.normalize(addr)); + let mut peer = self + .connect_sequential_authenticated(&self.v6, &v6_targets) + .await?; + peer.remote_addr = self.normalize(peer.remote_addr); + return Ok(peer); } (_, Some(_)) if !v4_targets.is_empty() => { - let addr = self.connect_sequential(&self.v4, &v4_targets).await?; - return Ok(self.normalize(addr)); + let mut peer = self + .connect_sequential_authenticated(&self.v4, &v4_targets) + .await?; + peer.remote_addr = self.normalize(peer.remote_addr); + return Ok(peer); } _ => return Err(anyhow::anyhow!("No suitable transport available")), }; @@ -2134,8 +2258,8 @@ impl DualStackNetworkNode { let v6_fut = async { for addr in v6_targets_clone { - if let Ok(connected_addr) = v6_node.connect_to_peer(addr).await { - return Ok(connected_addr); + if let Ok(peer) = v6_node.connect_to_peer_authenticated(addr).await { + return Ok(peer); } } Err(anyhow::anyhow!("IPv6 connect attempts failed")) @@ -2144,8 +2268,8 @@ impl DualStackNetworkNode { let v4_fut = async { sleep(HAPPY_EYEBALLS_V4_STAGGER).await; for addr in v4_targets_clone { - if let Ok(connected_addr) = v4_node.connect_to_peer(addr).await { - return Ok(connected_addr); + if let Ok(peer) = v4_node.connect_to_peer_authenticated(addr).await { + return Ok(peer); } } Err(anyhow::anyhow!("IPv4 connect attempts failed")) @@ -2153,22 +2277,22 @@ impl DualStackNetworkNode { tokio::select! { res6 = v6_fut => match res6 { - Ok(connected_addr) => Ok(connected_addr), + Ok(peer) => Ok(peer), Err(_) => { for addr in v4_targets { - if let Ok(connected_addr) = v4_node.connect_to_peer(addr).await { - return Ok(connected_addr); + if let Ok(peer) = v4_node.connect_to_peer_authenticated(addr).await { + return Ok(peer); } } Err(anyhow::anyhow!("All connect attempts failed")) } }, res4 = v4_fut => match res4 { - Ok(connected_addr) => Ok(connected_addr), + Ok(peer) => Ok(peer), Err(_) => { for addr in v6_targets { - if let Ok(connected_addr) = v6_node.connect_to_peer(addr).await { - return Ok(connected_addr); + if let Ok(peer) = v6_node.connect_to_peer_authenticated(addr).await { + return Ok(peer); } } Err(anyhow::anyhow!("All connect attempts failed")) @@ -2193,6 +2317,22 @@ impl DualStackNetworkNode { Err(anyhow::anyhow!("All connect attempts failed")) } + async fn connect_sequential_authenticated( + &self, + node: &Option>, + targets: &[SocketAddr], + ) -> Result { + let node = node + .as_ref() + .ok_or_else(|| anyhow::anyhow!("node not available"))?; + for &addr in targets { + if let Ok(peer) = node.connect_to_peer_authenticated(addr).await { + return Ok(peer); + } + } + Err(anyhow::anyhow!("All connect attempts failed")) + } + /// Return all local listening addresses pub async fn local_addrs(&self) -> Result> { let mut out = Vec::new(); diff --git a/src/transport_handle.rs b/src/transport_handle.rs index ee076aa2..6c7a19b9 100644 --- a/src/transport_handle.rs +++ b/src/transport_handle.rs @@ -22,7 +22,7 @@ use crate::PeerId; use crate::bgp_geo_provider::BgpGeoProvider; use crate::dht::core_engine::AddressType; use crate::error::{NetworkError, P2PError, P2pResult as Result, SendFailureKind, TransportError}; -use crate::identity::node_identity::NodeIdentity; +use crate::identity::node_identity::{NodeIdentity, peer_id_from_public_key_spki}; use crate::network::{ ConnectionStatus, MAX_ACTIVE_REQUESTS, MAX_REQUEST_TIMEOUT, MESSAGE_RECV_CHANNEL_CAPACITY, NetworkSender, P2PEvent, ParsedMessage, PeerInfo, PeerResponse, PendingRequest, @@ -38,6 +38,7 @@ use crate::validation::{RateLimitConfig, RateLimiter}; use dashmap::mapref::entry::Entry as DashEntry; use dashmap::{DashMap, DashSet}; +use saorsa_transport::nat_traversal_api::PreparedRelay; use std::collections::HashSet; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; @@ -282,18 +283,7 @@ pub struct TransportHandle { #[allow(dead_code)] geo_provider: Arc, shutdown: CancellationToken, - /// Peer address updates from ADD_ADDRESS frames (relay address advertisement). - /// - /// Bounded mpsc — see - /// [`crate::transport::saorsa_transport_adapter::ADDRESS_EVENT_CHANNEL_CAPACITY`]. - /// The producer (`spawn_peer_address_update_forwarder`) drops events - /// rather than blocking when the consumer is slow. - peer_address_update_rx: - tokio::sync::Mutex>, /// Relay established events — received when this node sets up a MASQUE relay. - /// - /// Bounded mpsc with the same drop semantics as - /// `peer_address_update_rx`. relay_established_rx: tokio::sync::Mutex>, /// Relay lost events — received when a previously-advertised MASQUE /// relay address is no longer reachable (tunnel died, health check @@ -301,18 +291,12 @@ pub struct TransportHandle { /// to trigger an immediate DHT republish with the stale relay /// address removed — without this, peers keep dialing the dead /// relay for the full health-poll cycle (5 s) or longer. - /// - /// Bounded mpsc with the same drop semantics as - /// `peer_address_update_rx`. relay_lost_rx: tokio::sync::Mutex>, /// Direct address promotion events — received when the passive /// reachability classifier proves one of this node's pinned external /// addresses is cold-dialable. Drained by the reachability driver while /// holding a relay so it can republish `Relay + Direct` instead of /// leaving peers with the older `Relay + Unverified` self-record. - /// - /// Bounded mpsc with the same drop semantics as - /// `peer_address_update_rx`. direct_address_promoted_rx: tokio::sync::Mutex>, /// Latest direct-address promotion for observers/tests. This is /// separate from the driver's single-consumer mpsc receiver so callers @@ -325,9 +309,6 @@ pub struct TransportHandle { /// Direct without crossing the Direct proof threshold. Drained by the /// reachability driver so a relay-only self-record can be corrected as /// soon as a non-relay fallback appears. - /// - /// Bounded mpsc with the same drop semantics as - /// `peer_address_update_rx`. self_address_updated_rx: tokio::sync::Mutex>, /// Latest self-address update for observers/tests. Separate from the /// driver's single-consumer mpsc receiver for the same reason as @@ -431,6 +412,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 // ============================================================================ @@ -516,7 +517,6 @@ impl TransportHandle { let proof_eligible_peers: Arc> = Arc::new(DashSet::new()); // Subscribe to address-related P2pEvents from the transport layer: - // - PeerAddressUpdated → mpsc, drained by the DHT bridge // - RelayEstablished → mpsc, drained by the DHT bridge // - RelayLost → mpsc, drained by the reachability driver // - DirectAddressPromoted → mpsc, drained by the reachability driver @@ -544,15 +544,14 @@ impl TransportHandle { self_address_updated_tx, self_address_updated_watch_tx.clone(), ); - let (peer_addr_update_rx, relay_established_rx, relay_lost_rx) = dual_node - .spawn_peer_address_update_forwarder( - Arc::clone(&external_addresses), - Arc::clone(&peer_observations), - Arc::clone(&proof_eligible_peers), - Arc::clone(&proven_externals), - direct_promoted_events.clone(), - self_address_updated_events.clone(), - ); + let (relay_established_rx, relay_lost_rx) = dual_node.spawn_address_event_forwarder( + Arc::clone(&external_addresses), + Arc::clone(&peer_observations), + Arc::clone(&proof_eligible_peers), + Arc::clone(&proven_externals), + direct_promoted_events.clone(), + self_address_updated_events.clone(), + ); dual_node.spawn_direct_reachability_classifier( Arc::clone(&dialed_addrs), @@ -621,7 +620,6 @@ impl TransportHandle { traffic, geo_provider, shutdown, - peer_address_update_rx: tokio::sync::Mutex::new(peer_addr_update_rx), relay_established_rx: tokio::sync::Mutex::new(relay_established_rx), relay_lost_rx: tokio::sync::Mutex::new(relay_lost_rx), direct_address_promoted_rx: tokio::sync::Mutex::new(direct_address_promoted_rx), @@ -703,12 +701,6 @@ impl TransportHandle { traffic: Arc::new(TrafficCounters::default()), geo_provider: Arc::new(BgpGeoProvider::new()), shutdown: CancellationToken::new(), - peer_address_update_rx: { - let (_tx, rx) = tokio::sync::mpsc::channel( - crate::transport::saorsa_transport_adapter::ADDRESS_EVENT_CHANNEL_CAPACITY, - ); - tokio::sync::Mutex::new(rx) - }, relay_established_rx: { let (_tx, rx) = tokio::sync::mpsc::channel( crate::transport::saorsa_transport_adapter::ADDRESS_EVENT_CHANNEL_CAPACITY, @@ -1020,19 +1012,6 @@ impl TransportHandle { .and_then(|p| p.value().iter().next().copied()) } - /// Drain pending peer address updates from ADD_ADDRESS frames. - /// - /// Returns (peer_connection_addr, advertised_addr) pairs. The caller - /// should look up the peer ID and update the DHT routing table. - pub async fn drain_peer_address_updates(&self) -> Vec<(SocketAddr, SocketAddr)> { - let mut rx = self.peer_address_update_rx.lock().await; - let mut updates = Vec::new(); - while let Ok(update) = rx.try_recv() { - updates.push(update); - } - updates - } - /// Drain any relay established events. Returns the relay address if this /// node has just established a MASQUE relay. pub async fn drain_relay_established(&self) -> Option { @@ -1041,18 +1020,6 @@ impl TransportHandle { rx.try_recv().ok() } - /// Wait for the next peer-address update from an ADD_ADDRESS frame. - /// - /// Returns `(peer_connection_addr, advertised_addr)` when one arrives, - /// or `None` if the underlying channel has closed (transport shut down). - /// - /// Use this in a `tokio::select!` against a shutdown token to react to - /// address updates immediately instead of polling. - pub async fn recv_peer_address_update(&self) -> Option<(SocketAddr, SocketAddr)> { - let mut rx = self.peer_address_update_rx.lock().await; - rx.recv().await - } - /// Wait for the next relay-established event. /// /// Resolves when this node has just set up a MASQUE relay (yielding @@ -1368,12 +1335,46 @@ impl TransportHandle { self.connect_peer_inner(address, Some(kind)).await } + /// Connect to a prospective relay during third-party canary validation. + /// + /// The transport semantics remain [`AddressType::Relay`], but the + /// dedicated structured log kind keeps a failed pre-publication probe + /// distinguishable from a failed dial of an already-published relay. + pub(crate) async fn probe_relay_canary_authenticated( + &self, + address: &MultiAddr, + ) -> Result { + let socket_addr = address.dialable_socket_addr().ok_or_else(|| { + P2PError::Network(NetworkError::InvalidAddress( + format!("relay canary requires a QUIC address, got {address}").into(), + )) + })?; + let target = normalize_wildcard_to_loopback(socket_addr); + let peer_public_key_spki = self + .dual_node + .probe_fresh_authenticated(target) + .await + .map_err(|error| P2PError::connection_failed(target, error.to_string()))?; + peer_id_from_public_key_spki(&peer_public_key_spki) + } + async fn connect_peer_inner( &self, address: &MultiAddr, kind: Option, ) -> Result { - let kind_label = address_kind_label(kind); + self.connect_peer_inner_authenticated(address, kind, None) + .await + .map(|(channel_id, _)| channel_id) + } + + async fn connect_peer_inner_authenticated( + &self, + address: &MultiAddr, + kind: Option, + log_kind: Option<&'static str>, + ) -> Result<(String, Option>)> { + let kind_label = log_kind.unwrap_or_else(|| address_kind_label(kind)); // Require a dialable (QUIC) transport. let socket_addr = address.dialable_socket_addr().ok_or_else(|| { @@ -1406,8 +1407,13 @@ impl TransportHandle { dial_target_normalized.ip(), )); - let peer_id = match self.dual_node.connect_happy_eyeballs(&addr_list).await { - Ok(addr) => { + let (peer_id, peer_public_key_spki) = match self + .dual_node + .connect_happy_eyeballs_authenticated(&addr_list) + .await + { + Ok(dialed_peer) => { + let addr = dialed_peer.remote_addr; let connected_peer_id = canonical_channel_id(addr); // Prevent self-connections by comparing against all listen @@ -1435,7 +1441,7 @@ impl TransportHandle { channel_id = %connected_peer_id, "Successfully connected to channel" ); - connected_peer_id + (connected_peer_id, dialed_peer.peer_public_key_spki) } Err(e) => { warn!( @@ -1468,7 +1474,7 @@ impl TransportHandle { // PeerConnected is emitted later when the peer's identity is // authenticated via a signed message — not at transport level. - Ok(peer_id) + Ok((peer_id, peer_public_key_spki)) } /// Check if the proactive relay session is still alive. @@ -1476,8 +1482,8 @@ impl TransportHandle { /// Returns `true` if no relay was established or the relay is healthy. /// Returns `false` if a relay was established but the QUIC connection /// has closed. Used by the relayer monitor (ADR-014 item 6). - pub fn is_relay_healthy(&self) -> bool { - self.dual_node.is_relay_healthy() + pub async fn is_relay_healthy(&self) -> bool { + self.dual_node.is_relay_healthy().await } /// Enable or disable relay serving on this node's MASQUE relay servers. @@ -1489,15 +1495,15 @@ impl TransportHandle { self.dual_node.set_relay_serving_enabled(enabled); } - /// Establish a proactive MASQUE relay session with the peer reachable at - /// `relay_addr`, returning the relay-allocated public socket address on - /// success. + /// Prepare a proactive MASQUE relay session with the peer reachable at + /// `relay_addr`, returning its provisional public socket address. /// /// This is the caller-driven entry point for ADR-014 relay acquisition. - /// It delegates through [`DualStackNetworkNode::setup_proactive_relay`] - /// to saorsa-transport's `NatTraversalEndpoint::setup_proactive_relay`, - /// which establishes the MASQUE `CONNECT-UDP` session and rebinds the - /// local Quinn endpoint onto the tunnel. + /// It delegates through [`DualStackNetworkNode::prepare_proactive_relay`] + /// to saorsa-transport's `NatTraversalEndpoint::prepare_proactive_relay`, + /// which establishes the MASQUE `CONNECT-UDP` session and a relay-backed + /// Quinn endpoint without advertising the address. The reachability driver + /// publishes or aborts the allocation after the canary verdict. /// /// Error conversion: saorsa-transport's `RelayAtCapacity` variant is /// mapped to [`RelaySessionEstablishError::AtCapacity`] so the acquisition @@ -1507,7 +1513,7 @@ impl TransportHandle { pub async fn setup_proactive_relay_session( &self, relay_addr: SocketAddr, - ) -> std::result::Result { + ) -> std::result::Result { use saorsa_transport::nat_traversal_api::NatTraversalError; use saorsa_transport::p2p_endpoint::EndpointError; @@ -1516,12 +1522,12 @@ impl TransportHandle { "requesting proactive MASQUE relay session from transport layer" ); - match self.dual_node.setup_proactive_relay(relay_addr).await { + match self.dual_node.prepare_proactive_relay(relay_addr).await { Ok(allocated) => { info!( relay = %relay_addr, - allocated = %allocated, - "proactive relay established" + allocated = %allocated.public_addr(), + "proactive relay prepared for canary verification" ); Ok(allocated) } @@ -1544,6 +1550,33 @@ impl TransportHandle { } } + /// Commit a canary-verified proactive relay and advertise it to peers. + pub async fn publish_proactive_relay_session(&self, allocation: PreparedRelay) -> Result<()> { + let relay_public_addr = allocation.public_addr(); + self.dual_node + .publish_proactive_relay(allocation) + .await + .map_err(|error| { + P2PError::Transport(TransportError::SetupFailed( + format!("Failed to publish proactive relay {relay_public_addr}: {error}") + .into(), + )) + }) + } + + /// Abort a proactive relay allocation and release its MASQUE resources. + pub async fn abort_proactive_relay_session(&self, allocation: PreparedRelay) -> Result<()> { + let relay_public_addr = allocation.public_addr(); + self.dual_node + .abort_proactive_relay(allocation) + .await + .map_err(|error| { + P2PError::Transport(TransportError::SetupFailed( + format!("Failed to abort proactive relay {relay_public_addr}: {error}").into(), + )) + }) + } + /// Disconnect from a peer, closing the underlying QUIC connection only /// when no other peers share the channel. /// @@ -1925,6 +1958,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(), @@ -1934,7 +1969,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(), )); @@ -1942,15 +1976,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 { @@ -1962,19 +1991,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. @@ -2762,6 +2780,23 @@ impl TransportHandle { .send_to_peer_optimized(&remote_address, &announce_bytes) .await { + if e + .downcast_ref::() + .is_some_and(|error| matches!( + error, + saorsa_transport::p2p_endpoint::EndpointError::PeerNotFound(_) + )) + { + // A one-shot reachability probe closes as + // soon as TLS exposes the target identity. + // Its inbound Established event can race this + // ordinary identity hook; by the time the send + // runs there is intentionally no peer left. + debug!( + "Skipping identity announce for closed channel {channel_id_for_send}" + ); + return; + } // {e:#} prints the full anyhow cause chain so we // can see the underlying reason (e.g. "peer did // not acknowledge stream data within 1s", @@ -2906,7 +2941,7 @@ impl RelaySessionEstablisher for TransportHandle { async fn establish( &self, relay_addr: SocketAddr, - ) -> std::result::Result { + ) -> std::result::Result { self.setup_proactive_relay_session(relay_addr).await } } @@ -2916,7 +2951,7 @@ impl RelaySessionEstablisher for Arc { async fn establish( &self, relay_addr: SocketAddr, - ) -> std::result::Result { + ) -> std::result::Result { self.setup_proactive_relay_session(relay_addr).await } } diff --git a/tests/dht_self_advertisement.rs b/tests/dht_self_advertisement.rs index fc86b03a..3bec4e41 100644 --- a/tests/dht_self_advertisement.rs +++ b/tests/dht_self_advertisement.rs @@ -584,7 +584,7 @@ async fn pinned_address_survives_connection_drop() { "ExternalAddressDiscovered event should reach the pinned external \ addresses within the timeout. If this fails, either saorsa-transport's \ poll_discovery_task is not firing the broadcast event, or the \ - ExternalAddressDiscovered branch in spawn_peer_address_update_forwarder \ + ExternalAddressDiscovered branch in spawn_address_event_forwarder \ is not pinning the address.", ); @@ -626,7 +626,7 @@ async fn pinned_address_survives_connection_drop() { {observed} after every live connection has dropped, but returned {after_drop:?}.\n\ \n\ Either the ExternalAddressDiscovered forwarder is not pinning the \ - address (check spawn_peer_address_update_forwarder in \ + address (check spawn_address_event_forwarder in \ saorsa_transport_adapter.rs), or the pinned path in \ TransportHandle::observed_external_address() is not reading it." );