diff --git a/src/dht_network_manager.rs b/src/dht_network_manager.rs index 433e0d38..7715a651 100644 --- a/src/dht_network_manager.rs +++ b/src/dht_network_manager.rs @@ -69,6 +69,8 @@ use uuid::Uuid; #[cfg(test)] mod address_v2_tests; +#[cfg(test)] +mod closer_peers_tests; /// Minimum concurrent operations for semaphore backpressure const MIN_CONCURRENT_OPERATIONS: usize = 10; @@ -86,6 +88,9 @@ const MAX_MESSAGE_SIZE: usize = 64 * 1024; /// Optional signed-record trailer on the unchanged FIND_NODE response. const LOOKUP_EXTENSION_MARKER: &[u8; 8] = b"ADDRSIG1"; const MAX_LOOKUP_EXTENSION_RECORDS: usize = 256; +/// Format byte leading an encoded closer-peers payload, so the format can +/// change without being mistaken for this one. +const CLOSER_PEERS_FORMAT: u8 = 1; /// Keep room for the largest native self-address set, including future relay /// acquisition and IPv4/IPv6 reachability changes. @@ -1378,6 +1383,14 @@ fn split_witnessed_transcript_views( (responder_node_views, missing_responders) } +/// The body of an encoded closer-peers payload: a FIND_NODE answer without +/// the DHT message around it. +#[derive(Serialize, Deserialize)] +struct CloserPeers { + key: Key, + nodes: Vec, +} + struct NativeFindNodeQuery<'a> { manager: &'a DhtNetworkManager, transcript_view_count: Option, @@ -1482,15 +1495,7 @@ impl LookupQuery for NativeFindNodeQuery<'_> { result: DhtNetworkResult::PeerRejected, .. }) => { - info!( - "[NETWORK] Peer {} rejected us — removing from routing table", - peer_id.to_hex() - ); - let mut dht = manager.dht.write().await; - let rt_events = dht.remove_node_by_id(&peer_id).await; - drop(dht); - manager.broadcast_routing_events(&rt_events); - let _ = manager.transport.disconnect_peer(&peer_id).await; + manager.forget_rejecting_peer(&peer_id).await; outcomes.push(LookupQueryOutcome::Failed { responder: *peer_id.as_bytes(), }); @@ -1606,8 +1611,7 @@ impl DhtNetworkManager { hex::encode(key) ); - let candidate_nodes = self.find_closest_nodes_local(key, self.k_value()).await; - let closer_nodes = Self::filter_response_nodes(candidate_nodes, requester); + let closer_nodes = self.find_node_answer(key, requester).await; // Log addresses being returned in FIND_NODE response for node in &closer_nodes { @@ -1625,6 +1629,198 @@ impl DhtNetworkManager { }) } + /// The K closest routing-table peers to `key`, without `requester`: what + /// this node answers a FIND_NODE with. + async fn find_node_answer(&self, key: &Key, requester: &PeerId) -> Vec { + let candidate_nodes = self.find_closest_nodes_local(key, self.k_value()).await; + Self::filter_response_nodes(candidate_nodes, requester) + } + + /// Encode this node's FIND_NODE answer for `key` to `requester`, for an + /// application response that returns closer peers in place of a value + /// (Kademlia's FIND_VALUE). [`Self::decode_closer_peers`] reads it. + /// + /// It holds what a FIND_NODE answer would: the K closest routing-table + /// peers other than `requester`, their QUIC addresses, and as many of + /// their owner-signed address records as fit the FIND_NODE size limit. + /// + /// # Errors + /// + /// Returns an error if the answer cannot be encoded within that limit. + pub async fn encode_closer_peers(&self, key: &Key, requester: &PeerId) -> Result> { + let mut nodes = self.find_node_answer(key, requester).await; + let records = Self::split_lookup_publications(&mut nodes); + let mut bytes = vec![CLOSER_PEERS_FORMAT]; + let body = CloserPeers { key: *key, nodes }; + bytes = postcard::to_extend(&body, bytes) + .map_err(|error| P2PError::Serialization(error.to_string().into()))?; + if bytes.len() > MAX_MESSAGE_SIZE { + return Err(P2PError::Validation( + "closer peers exceed the message size limit".into(), + )); + } + Self::append_lookup_extension(&mut bytes, records)?; + Ok(bytes) + } + + /// Lookup candidates from closer peers that `responder` encoded for `key` + /// with [`Self::encode_closer_peers`]. + /// + /// The payload is checked as a FIND_NODE answer from `responder` would + /// be: + /// - a payload larger than a FIND_NODE answer may be is rejected before + /// it is parsed + /// - owner-signed address records are verified + /// - the responder's own report and locally proven owner views are + /// applied + /// - LAN addresses are filtered against `transport_source` + /// - peers whose every address is in the dial-failure cache are left out + /// + /// The caller must have received `payload` from `responder` over an + /// authenticated channel. Combining reports from several responders, + /// and skipping peers already queried, is left to the caller's lookup. + /// + /// # Errors + /// + /// Returns an error if the payload is larger than a FIND_NODE answer may + /// be, is not in this format, or answers a different key. + pub async fn decode_closer_peers( + &self, + responder: &PeerId, + key: &Key, + payload: &[u8], + transport_source: Option<&MultiAddr>, + ) -> Result> { + // The encoder keeps the whole answer, trailer included, within the + // FIND_NODE limit. The application message carrying it may be far + // larger, so check the limit here, before deserialization. + if payload.len() > MAX_MESSAGE_SIZE { + return Err(P2PError::Validation( + format!( + "closer peers of {} bytes exceed the message size limit of {MAX_MESSAGE_SIZE} bytes", + payload.len() + ) + .into(), + )); + } + let Some((&CLOSER_PEERS_FORMAT, body)) = payload.split_first() else { + return Err(P2PError::Validation( + "unsupported closer-peers format".into(), + )); + }; + let (decoded, trailing): (CloserPeers, _) = postcard::take_from_bytes(body) + .map_err(|error| P2PError::Serialization(error.to_string().into()))?; + if decoded.key != *key { + return Err(P2PError::Validation( + "closer peers answer a different key".into(), + )); + } + let records = Self::decode_lookup_extension(trailing); + let nodes = self + .normalize_lookup_answer(decoded.nodes, &records, responder, transport_source) + .await; + Ok(self + .usable_lookup_candidates(nodes, transport_source, key) + .await) + } + + /// Ask one peer for its closest peers to `key`: one step of a lookup the + /// caller drives itself. Returns lookup candidates checked as + /// [`Self::decode_closer_peers`] checks them. + /// + /// # Errors + /// + /// Returns an error if the peer cannot be reached, does not answer, or + /// rejects us. A rejecting peer is removed from the routing table, as in + /// a network lookup. + pub async fn find_node_on_peer(&self, node: &DHTNode, key: &Key) -> Result> { + let (peer_id, response) = self + .send_find_node_lookup_request( + node.peer_id, + node.typed_addresses(), + DhtNetworkOperation::FindNode { key: *key }, + self.lookup_failures.subscribe(), + ) + .await; + match response? { + DhtResponseEnvelope { + result: DhtNetworkResult::NodesFound { nodes, .. }, + transport_source, + .. + } => { + self.merge_trusted_gossiped_typed_addresses(node).await; + Ok(self + .usable_lookup_candidates(nodes, transport_source.as_ref(), key) + .await) + } + DhtResponseEnvelope { + result: DhtNetworkResult::PeerRejected, + .. + } => { + self.forget_rejecting_peer(&peer_id).await; + Err(P2PError::Network(NetworkError::PeerNotFound( + format!("peer {} rejected the lookup", peer_id.to_hex()).into(), + ))) + } + _ => Ok(Vec::new()), + } + } + + /// Connect to `node` as a lookup query would, so its identity and user + /// agent are known before the caller chooses what to send it. + /// + /// # Errors + /// + /// Returns an error if no address of `node` can be dialled or its + /// identity cannot be confirmed. + pub async fn connect_lookup_peer(&self, node: &DHTNode) -> Result<()> { + self.ensure_peer_channel(&node.peer_id, &node.typed_addresses()) + .await + } + + /// Whether a lookup may query `node`: it is connected, or not every + /// address it would be dialled on is in the dial-failure cache. + pub async fn is_lookup_candidate_dialable(&self, node: &DHTNode) -> bool { + !self.lookup_candidate_dial_plan_is_exhausted(node).await + } + + /// The K closest trusted candidates in a normalized lookup answer, less + /// those that cannot be dialled now, with their addresses merged into + /// the routing table where the peers are already known. + async fn usable_lookup_candidates( + &self, + nodes: Vec, + transport_source: Option<&MultiAddr>, + key: &Key, + ) -> Vec { + let (candidates, _) = self + .trusted_find_node_response_nodes(nodes, transport_source, key, self.k_value(), 0) + .await; + let mut usable = Vec::with_capacity(candidates.len()); + for node in candidates { + if self.lookup_candidate_dial_plan_is_exhausted(&node).await { + continue; + } + self.merge_trusted_gossiped_typed_addresses(&node).await; + usable.push(node); + } + usable + } + + /// Remove a peer that rejected a lookup from the routing table and drop + /// its connection. + async fn forget_rejecting_peer(&self, peer_id: &PeerId) { + info!( + "[NETWORK] Peer {} rejected us — removing from routing table", + peer_id.to_hex() + ); + let mut dht = self.dht.write().await; + let rt_events = dht.remove_node_by_id(peer_id).await; + drop(dht); + self.broadcast_routing_events(&rt_events); + let _ = self.transport.disconnect_peer(peer_id).await; + } + /// Create a new DHT Network Manager using an existing transport handle. /// /// The caller is responsible for the transport lifecycle and must stop @@ -5083,58 +5279,17 @@ impl DhtNetworkManager { }; let result = match result { - DhtNetworkResult::NodesFound { key, nodes } => { - let base_peers: HashMap<_, _> = nodes - .iter() - .map(|node| (node.peer_id, node.reliability)) - .collect(); - let verified = message - .signed_records - .iter() - .filter_map(|record| { - let proof = record.verify().ok()?; - let reliability = *base_peers.get(&proof.owner())?; - Some((proof, reliability)) - }) - .collect(); - let mut signed_by_owner = HashMap::new(); - for node in self - .normalize_signed_lookup_nodes(verified, transport_source) - .await - { - signed_by_owner - .entry(node.peer_id) - .and_modify(|current: &mut DHTNode| current.merge_from(node.clone())) - .or_insert(node); - } - let mut normalized = Vec::new(); - for mut node in nodes { - let claimed_seq = advertised_publish_seq(&node); - node.address_authority = None; - node.distance = None; - if node.peer_id == sender_app_id && claimed_seq != 0 { - if let Some(typed) = self - .apply_native_self_report(&node, claimed_seq, transport_source) - .await - { - (node.addresses, node.address_types) = typed.into_iter().unzip(); - node.address_authority = - Some(AddressAuthority::AuthenticatedOwner(claimed_seq)); - } else { - node.addresses.clear(); - node.address_types.clear(); - } - } - if let Some(signed) = signed_by_owner.remove(&node.peer_id) { - node.merge_from(signed); - } - normalized.push(self.protect_owner_view(node).await); - } - DhtNetworkResult::NodesFound { - key, - nodes: normalized, - } - } + DhtNetworkResult::NodesFound { key, nodes } => DhtNetworkResult::NodesFound { + key, + nodes: self + .normalize_lookup_answer( + nodes, + &message.signed_records, + &sender_app_id, + transport_source, + ) + .await, + }, result => result, }; let response = DhtResponseEnvelope { @@ -5148,6 +5303,64 @@ impl DhtNetworkManager { Ok(()) } + /// Normalize the nodes of a FIND_NODE answer from authenticated + /// `responder`: verify owner-signed records, apply the responder's own + /// report, and protect locally proven owner views. + async fn normalize_lookup_answer( + &self, + nodes: Vec, + signed_records: &[SignedAddressRecord], + responder: &PeerId, + transport_source: Option<&MultiAddr>, + ) -> Vec { + let base_peers: HashMap<_, _> = nodes + .iter() + .map(|node| (node.peer_id, node.reliability)) + .collect(); + let verified = signed_records + .iter() + .filter_map(|record| { + let proof = record.verify().ok()?; + let reliability = *base_peers.get(&proof.owner())?; + Some((proof, reliability)) + }) + .collect(); + let mut signed_by_owner = HashMap::new(); + for node in self + .normalize_signed_lookup_nodes(verified, transport_source) + .await + { + signed_by_owner + .entry(node.peer_id) + .and_modify(|current: &mut DHTNode| current.merge_from(node.clone())) + .or_insert(node); + } + let mut normalized = Vec::new(); + for mut node in nodes { + let claimed_seq = advertised_publish_seq(&node); + node.address_authority = None; + node.distance = None; + if node.peer_id == *responder && claimed_seq != 0 { + if let Some(typed) = self + .apply_native_self_report(&node, claimed_seq, transport_source) + .await + { + (node.addresses, node.address_types) = typed.into_iter().unzip(); + node.address_authority = + Some(AddressAuthority::AuthenticatedOwner(claimed_seq)); + } else { + node.addresses.clear(); + node.address_types.clear(); + } + } + if let Some(signed) = signed_by_owner.remove(&node.peer_id) { + node.merge_from(signed); + } + normalized.push(self.protect_owner_view(node).await); + } + normalized + } + /// Handle DHT broadcast message async fn handle_dht_broadcast(&self, _message: &DhtNetworkMessage) -> Result<()> { // Handle broadcast messages (for network-wide announcements) @@ -5162,35 +5375,79 @@ impl DhtNetworkManager { .map_err(|error| P2PError::Serialization(error.to_string().into()))?; if matches!(message.message_type, DhtMessageType::Response) && matches!(message.result, Some(DhtNetworkResult::NodesFound { .. })) - && let Some(bundle) = trailing.strip_prefix(LOOKUP_EXTENSION_MARKER) { - message.signed_records = - crate::signed_address::decode_record_bundle(bundle, MAX_LOOKUP_EXTENSION_RECORDS) - .unwrap_or_default(); + message.signed_records = Self::decode_lookup_extension(trailing); } Ok(message) } + /// Owner-signed records from a lookup answer's trailer. Unknown or + /// malformed trailers yield none rather than invalidating the answer. + fn decode_lookup_extension(trailing: &[u8]) -> Vec { + trailing + .strip_prefix(LOOKUP_EXTENSION_MARKER) + .and_then(|bundle| { + crate::signed_address::decode_record_bundle(bundle, MAX_LOOKUP_EXTENSION_RECORDS) + .ok() + }) + .unwrap_or_default() + } + + /// Take each node's owner publication out as a signed record and keep + /// only its QUIC addresses, as the legacy lookup answer requires. + fn split_lookup_publications(nodes: &mut [DHTNode]) -> Vec { + let mut records = Vec::new(); + for node in nodes { + if let Some(proof) = node + .address_authority + .as_ref() + .and_then(AddressAuthority::publication) + && !records.contains(proof.signed()) + { + records.push(proof.signed().clone()); + } + if node.addresses.iter().any(|address| !address.is_quic()) { + (node.addresses, node.address_types) = node + .typed_addresses() + .into_iter() + .filter(|(address, _)| address.is_quic()) + .unzip(); + } + } + records + } + + /// Append as many complete owner-signed records as fit within the legacy + /// message size limit. + fn append_lookup_extension( + bytes: &mut Vec, + records: Vec, + ) -> Result<()> { + let mut extension = Vec::new(); + for record in records.into_iter().take(MAX_LOOKUP_EXTENSION_RECORDS) { + let encoded = crate::signed_address::encode_record_bundle(&[record]) + .map_err(|error| P2PError::Serialization(error.into()))?; + if bytes.len() + LOOKUP_EXTENSION_MARKER.len() + extension.len() + encoded.len() + <= MAX_MESSAGE_SIZE + { + extension.extend_from_slice(&encoded); + } + } + if !extension.is_empty() { + bytes.extend_from_slice(LOOKUP_EXTENSION_MARKER); + bytes.extend_from_slice(&extension); + } + Ok(()) + } + /// Preserve the complete QUIC-only legacy response, then append as many /// complete owner-signed records as fit within the legacy 64 KiB limit. fn encode_response_message(mut response: DhtNetworkMessage) -> Result> { let mut records = std::mem::take(&mut response.signed_records); if let Some(DhtNetworkResult::NodesFound { nodes, .. }) = &mut response.result { - for node in nodes { - if let Some(proof) = node - .address_authority - .as_ref() - .and_then(AddressAuthority::publication) - && !records.contains(proof.signed()) - { - records.push(proof.signed().clone()); - } - if node.addresses.iter().any(|address| !address.is_quic()) { - (node.addresses, node.address_types) = node - .typed_addresses() - .into_iter() - .filter(|(address, _)| address.is_quic()) - .unzip(); + for record in Self::split_lookup_publications(nodes) { + if !records.contains(&record) { + records.push(record); } } } @@ -5202,20 +5459,7 @@ impl DhtNetworkManager { )); } if matches!(response.result, Some(DhtNetworkResult::NodesFound { .. })) { - let mut extension = Vec::new(); - for record in records.into_iter().take(MAX_LOOKUP_EXTENSION_RECORDS) { - let encoded = crate::signed_address::encode_record_bundle(&[record]) - .map_err(|error| P2PError::Serialization(error.into()))?; - if bytes.len() + LOOKUP_EXTENSION_MARKER.len() + extension.len() + encoded.len() - <= MAX_MESSAGE_SIZE - { - extension.extend_from_slice(&encoded); - } - } - if !extension.is_empty() { - bytes.extend_from_slice(LOOKUP_EXTENSION_MARKER); - bytes.extend_from_slice(&extension); - } + Self::append_lookup_extension(&mut bytes, records)?; } Ok(bytes) } diff --git a/src/dht_network_manager/closer_peers_tests.rs b/src/dht_network_manager/closer_peers_tests.rs new file mode 100644 index 00000000..5efe09c4 --- /dev/null +++ b/src/dht_network_manager/closer_peers_tests.rs @@ -0,0 +1,219 @@ +// Copyright 2024 Saorsa Labs Limited +// +// This software is licensed under the MIT license or the Apache License, Version 2.0 +// , at your +// option. This file may not be copied, modified, or distributed except +// according to those terms. + +use super::*; +use crate::P2PNode; + +/// Copies of one peer in a closer-peers body no honest encoder produces: well +/// over the FIND_NODE size limit. +const OVERSIZED_NODES: usize = 2_000; + +async fn test_node() -> P2PNode { + P2PNode::new( + NodeConfig::builder() + .local(true) + .port(0) + .ipv6(false) + .build() + .unwrap(), + ) + .await + .unwrap() +} + +async fn listening_node() -> (P2PNode, MultiAddr) { + let node = test_node().await; + node.dht_manager() + .transport + .start_network_listeners() + .await + .unwrap(); + node.dht_manager().start().await.unwrap(); + let address = node + .listen_addrs() + .await + .into_iter() + .find(MultiAddr::is_ipv4) + .unwrap(); + (node, address) +} + +async fn seed_peer(manager: &DhtNetworkManager, owner: PeerId, address: &MultiAddr) { + manager + .dht + .write() + .await + .add_node_no_trust(NodeInfo { + id: owner, + addresses: vec![address.clone()], + address_types: vec![AddressType::Direct], + last_seen: AtomicInstant::now(), + }) + .await + .unwrap(); +} + +fn lookup_node(peer_id: PeerId, address: &MultiAddr) -> DHTNode { + DHTNode { + peer_id, + addresses: vec![address.clone()], + address_types: vec![AddressType::Direct], + distance: None, + reliability: 1.0, + address_authority: None, + } +} + +fn peer_ids(nodes: &[DHTNode]) -> Vec { + let mut ids: Vec<_> = nodes.iter().map(|node| node.peer_id).collect(); + ids.sort(); + ids +} + +#[tokio::test] +async fn closer_peers_match_a_find_node_answer() { + let (requester, _) = listening_node().await; + let (responder, responder_address) = listening_node().await; + let (known, known_address) = listening_node().await; + seed_peer(responder.dht_manager(), *known.peer_id(), &known_address).await; + let responder_node = lookup_node(*responder.peer_id(), &responder_address); + let key = *known.peer_id().as_bytes(); + + requester + .dht_manager() + .connect_lookup_peer(&responder_node) + .await + .unwrap(); + assert!( + requester + .dht_manager() + .is_lookup_candidate_dialable(&responder_node) + .await + ); + let answered = requester + .dht_manager() + .find_node_on_peer(&responder_node, &key) + .await + .unwrap(); + let payload = responder + .dht_manager() + .encode_closer_peers(&key, requester.peer_id()) + .await + .unwrap(); + let decoded = requester + .dht_manager() + .decode_closer_peers( + responder.peer_id(), + &key, + &payload, + Some(&responder_address), + ) + .await + .unwrap(); + + assert!(peer_ids(&answered).contains(known.peer_id())); + assert_eq!(peer_ids(&decoded), peer_ids(&answered)); + for node in [&requester, &responder, &known] { + node.stop().await.unwrap(); + } +} + +#[tokio::test] +async fn closer_peers_exclude_the_requester() { + let responder = test_node().await; + let requester = PeerId::from_bytes([7; 32]); + seed_peer( + responder.dht_manager(), + requester, + &"/ip4/127.0.0.1/udp/9/quic".parse().unwrap(), + ) + .await; + let payload = responder + .dht_manager() + .encode_closer_peers(requester.as_bytes(), &requester) + .await + .unwrap(); + let (body, _): (CloserPeers, _) = postcard::take_from_bytes(&payload[1..]).unwrap(); + assert!(body.nodes.iter().all(|node| node.peer_id != requester)); +} + +#[tokio::test] +async fn closer_peers_reject_another_key_or_format() { + let node = test_node().await; + let key = [1; 32]; + let responder = PeerId::from_bytes([2; 32]); + let payload = node + .dht_manager() + .encode_closer_peers(&key, &PeerId::from_bytes([3; 32])) + .await + .unwrap(); + + let other_key = node + .dht_manager() + .decode_closer_peers(&responder, &[9; 32], &payload, None) + .await; + assert!(other_key.is_err()); + + let mut other_format = payload; + other_format[0] = CLOSER_PEERS_FORMAT.wrapping_add(1); + let other_format = node + .dht_manager() + .decode_closer_peers(&responder, &key, &other_format, None) + .await; + assert!(other_format.is_err()); + + let empty = node + .dht_manager() + .decode_closer_peers(&responder, &key, &[], None) + .await; + assert!(empty.is_err()); +} + +#[tokio::test] +async fn closer_peers_over_the_size_limit_are_rejected_before_parsing() { + let node = test_node().await; + let key = [1; 32]; + let responder = PeerId::from_bytes([2; 32]); + let address: MultiAddr = "/ip4/127.0.0.1/udp/9/quic".parse().unwrap(); + let nodes = vec![lookup_node(PeerId::from_bytes([4; 32]), &address); OVERSIZED_NODES]; + let payload = + postcard::to_extend(&CloserPeers { key, nodes }, vec![CLOSER_PEERS_FORMAT]).unwrap(); + assert!(payload.len() > MAX_MESSAGE_SIZE); + + let decoded = node + .dht_manager() + .decode_closer_peers(&responder, &key, &payload, None) + .await; + assert!(decoded.is_err()); +} + +#[tokio::test] +async fn closer_peers_are_read_up_to_the_size_limit() { + let node = test_node().await; + let key = [1; 32]; + let responder = PeerId::from_bytes([2; 32]); + let mut payload = node + .dht_manager() + .encode_closer_peers(&key, &PeerId::from_bytes([3; 32])) + .await + .unwrap(); + // An unknown trailer is ignored, so padding reaches the limit exactly. + payload.resize(MAX_MESSAGE_SIZE, 0); + let at_limit = node + .dht_manager() + .decode_closer_peers(&responder, &key, &payload, None) + .await; + assert!(at_limit.is_ok()); + + payload.push(0); + let over_limit = node + .dht_manager() + .decode_closer_peers(&responder, &key, &payload, None) + .await; + assert!(over_limit.is_err()); +}