From da0d9d39a0eada3aa338bc60f3a0f5411d84fe1f Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sat, 30 May 2026 00:46:38 +0000 Subject: [PATCH 1/2] node: refresh active peer paths without dropping links Add Node::update_peers for runtime peer-list refresh. It re-derives the active peer connections from a new peer configuration, adding newly configured peers and removing those no longer present, while keeping links to peers that remain in the set rather than tearing every connection down. The call returns an UpdatePeersOutcome summarizing the added, removed, and retained peers. PeerAddress gains a seen_at_ms recency field (with_seen_at_ms). Active path selection now sorts address candidates by recency so the most recently observed address wins when concurrent path probes race. complete_rekey_msg2 now returns the remote peer's startup epoch alongside the new Noise session, letting the rekey path detect a peer restart and clear stale session state. A stale FSP session is cleared when a peer restart is detected during FMP rekey or cross-connection promotion, so the session-layer map no longer lingers out of sync with the freshly promoted peer. Per-tick work budgets bound the connection churn in a single node tick (MAX_DISCOVERY_CONNECTS_PER_TICK, MAX_RETRY_CONNECTIONS_PER_TICK, MAX_PARALLEL_PATH_CANDIDATES_PER_PEER); work beyond a tick's budget is deferred to the next tick rather than discarded. Co-authored-by: Johnathan Corgan --- CHANGELOG.md | 22 ++ src/config/peer.rs | 29 ++ src/lib.rs | 2 +- src/node/handlers/discovery.rs | 17 +- src/node/handlers/handshake.rs | 49 ++- src/node/lifecycle.rs | 562 +++++++++++++++++++++++++++++---- src/node/mod.rs | 13 + src/node/retry.rs | 12 +- src/node/tests/unit.rs | 218 +++++++++++++ src/peer/active.rs | 15 +- 10 files changed, 872 insertions(+), 67 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 474c1da..fe9f9cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `Node::update_peers` for runtime peer-list refresh, returning an + `UpdatePeersOutcome` summarizing added, removed, and retained peers. + Re-derives active peer connections from a new peer configuration + without dropping links to peers that remain in the set. + `PeerAddress` gains a `seen_at_ms` recency field (with + `with_seen_at_ms`) used to prefer more recently observed addresses. - `pool_inbound` and `pool_outbound` counters on the TCP and Tor transport stats (`TcpStats`, `TorStats`). Per-direction accounting is updated at every pool-insert and receive-loop-exit site, plus on @@ -25,6 +31,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `complete_rekey_msg2` now returns the remote peer's startup epoch + alongside the new Noise session, so the rekey path can detect a peer + restart and clear stale session state. +- Active-peer path selection now sorts address candidates by recency + (`seen_at_ms`), preferring the most recently observed address when + racing concurrent path probes. +- Per-tick work budgets bound the connection churn done in a single + node tick: `MAX_DISCOVERY_CONNECTS_PER_TICK`, + `MAX_RETRY_CONNECTIONS_PER_TICK`, and + `MAX_PARALLEL_PATH_CANDIDATES_PER_PEER`. Work beyond a tick's budget + is deferred to the next tick rather than discarded. - Nostr discovery startup is now non-blocking. `Node::start` no longer waits for relay connect, subscribe, or initial advert publish before returning. A slow or unreachable relay no longer @@ -138,6 +155,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- A stale FSP (session-layer) session is now cleared when a peer + restart is detected during FMP rekey or cross-connection promotion. + Previously the old session could linger after the peer came back + with a new startup epoch, leaving the session-layer map out of sync + with the freshly promoted peer. - TCP and Tor `max_inbound_connections` admission cap is now compared against the per-direction inbound count (`pool_inbound`) rather than the combined pool size. Outbound connect-on-send connections share diff --git a/src/config/peer.rs b/src/config/peer.rs index 51a0903..3247737 100644 --- a/src/config/peer.rs +++ b/src/config/peer.rs @@ -45,8 +45,28 @@ pub struct PeerAddress { /// are tried first. #[serde(default = "default_priority")] pub priority: u8, + + /// Wall-clock observation timestamp (Unix ms) for ranking by recency. + /// + /// `None` means "no freshness signal", typically an operator-edited + /// static config. The dialer sorts candidates by this field descending + /// so freshly observed overlay or runtime hints can be tried before stale + /// static addresses. This field is runtime-only and is ignored when + /// comparing peer-address lists for config changes. + #[serde(default, skip_serializing_if = "Option::is_none", skip_deserializing)] + pub seen_at_ms: Option, } +impl PartialEq for PeerAddress { + fn eq(&self, other: &Self) -> bool { + self.transport == other.transport + && self.addr == other.addr + && self.priority == other.priority + } +} + +impl Eq for PeerAddress {} + fn default_priority() -> u8 { 100 } @@ -62,6 +82,7 @@ impl PeerAddress { transport: transport.into(), addr: addr.into(), priority: default_priority(), + seen_at_ms: None, } } @@ -75,8 +96,16 @@ impl PeerAddress { transport: transport.into(), addr: addr.into(), priority, + seen_at_ms: None, } } + + /// Tag this address with a freshness timestamp for source-agnostic + /// candidate ranking. + pub fn with_seen_at_ms(mut self, seen_at_ms: u64) -> Self { + self.seen_at_ms = Some(seen_at_ms); + self + } } /// Configuration for a known peer. diff --git a/src/lib.rs b/src/lib.rs index d3906e5..54252f3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -67,4 +67,4 @@ pub use peer::{ }; // Re-export node types -pub use node::{Node, NodeError, NodeState}; +pub use node::{Node, NodeError, NodeState, UpdatePeersOutcome}; diff --git a/src/node/handlers/discovery.rs b/src/node/handlers/discovery.rs index d4ff004..933dda2 100644 --- a/src/node/handlers/discovery.rs +++ b/src/node/handlers/discovery.rs @@ -11,6 +11,8 @@ use crate::transport::{TransportAddr, TransportId}; use crate::{NodeAddr, PeerIdentity}; use tracing::{debug, info, trace, warn}; +const MAX_RECENT_DISCOVERY_REQUESTS: usize = 4096; + impl Node { /// Handle an incoming LookupRequest from a peer. /// @@ -34,6 +36,7 @@ impl Node { }; let now_ms = Self::now_ms(); + self.purge_expired_requests(now_ms); // Dedup: drop if we've already seen this request_id. // Also serves as loop protection — tree routing is loop-free, @@ -48,13 +51,21 @@ impl Node { return; } + if self.recent_requests.len() >= MAX_RECENT_DISCOVERY_REQUESTS { + debug!( + request_id = request.request_id, + from = %self.peer_display_name(from), + recent_requests = self.recent_requests.len(), + max_recent_requests = MAX_RECENT_DISCOVERY_REQUESTS, + "Discovery request dedup cache full, dropping LookupRequest" + ); + return; + } + // Record for reverse-path forwarding and dedup self.recent_requests .insert(request.request_id, RecentRequest::new(*from, now_ms)); - // Lazy purge expired entries - self.purge_expired_requests(now_ms); - // Are we the target? if request.target == *self.node_addr() { self.stats_mut().discovery.req_target_is_us += 1; diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 0743bea..e76ebe5 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -658,8 +658,15 @@ impl Node { // Complete the rekey handshake on the ActivePeer if let Some(peer) = self.peers.get_mut(&peer_node_addr) { match peer.complete_rekey_msg2(noise_msg2) { - Ok(session) => { + Ok((session, remote_epoch)) => { let our_index = peer.rekey_our_index().unwrap_or(header.receiver_idx); + let remote_epoch_changed = matches!( + (peer.remote_epoch(), remote_epoch), + (Some(old), Some(new)) if old != new + ); + if remote_epoch.is_some() { + peer.set_remote_epoch(remote_epoch); + } peer.set_pending_session(session, our_index, header.sender_idx); if let Some(transport_id) = peer.transport_id() { @@ -667,6 +674,19 @@ impl Node { .insert((transport_id, our_index.as_u32()), peer_node_addr); } + if remote_epoch_changed { + if self.sessions.remove(&peer_node_addr).is_some() { + debug!( + peer = %display_name, + "Cleared stale FSP session after peer restart during FMP rekey" + ); + } + info!( + peer = %display_name, + "Peer restart detected during FMP rekey, replacing stale endpoint session" + ); + } + debug!( peer = %display_name, new_our_index = %our_index, @@ -1024,9 +1044,15 @@ impl Node { if let Some(existing_peer) = self.peers.get(&peer_node_addr) { let existing_link_id = existing_peer.link_id(); - // Determine which connection wins - let this_wins = - cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound); + let remote_epoch_changed = matches!((existing_peer.remote_epoch(), remote_epoch), (Some(old), Some(new)) if old != new); + + // Determine which connection wins. A peer restart (different + // startup epoch) is not a normal cross-connection: the old link + // and FSP sessions are cryptographically stale, so the freshly + // authenticated connection must replace them regardless of the + // tie-breaker direction. + let this_wins = remote_epoch_changed + || cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound); if this_wins { // This connection wins, replace the existing peer @@ -1052,6 +1078,21 @@ impl Node { let _ = self.index_allocator.free(old_idx); } + if remote_epoch_changed { + if self.sessions.remove(&peer_node_addr).is_some() { + debug!( + peer = %self.peer_display_name(&peer_node_addr), + "Cleared stale FSP session after peer restart during promotion" + ); + } + info!( + peer = %self.peer_display_name(&peer_node_addr), + winner_link = %link_id, + loser_link = %loser_link_id, + "Peer restart detected during promotion, replacing stale active peer" + ); + } + self.seed_path_mtu_for_link_peer(&peer_node_addr, transport_id, ¤t_addr); let mut new_peer = ActivePeer::with_session( diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 103b87e..4d3c326 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -14,14 +14,177 @@ use crate::protocol::{Disconnect, DisconnectReason}; use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel}; use crate::upper::tun::{TunDevice, TunState, run_tun_reader, shutdown_tun_interface}; use crate::{NodeAddr, PeerIdentity}; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::thread; use std::time::Duration; use tracing::{debug, info, warn}; const OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER: u64 = 2; +const MAX_PARALLEL_PATH_CANDIDATES_PER_PEER: usize = 4; +const MAX_DISCOVERY_CONNECTS_PER_TICK: usize = 16; impl Node { + /// Replace the runtime peer list. + /// + /// Newly added auto-connect peers are dialed immediately, removed peers + /// are dropped from retry bookkeeping, and existing peers get fresh + /// address hints without tearing down an active link. If an existing peer + /// is already connected and a new concrete candidate appears, FIPS starts + /// an alternate handshake in parallel; promotion switches only after that + /// handshake authenticates. + pub async fn update_peers( + &mut self, + new_peers: Vec, + ) -> Result { + let mut new_by_addr: HashMap = + HashMap::with_capacity(new_peers.len()); + for peer in new_peers { + let identity = + PeerIdentity::from_npub(&peer.npub).map_err(|e| NodeError::InvalidPeerNpub { + npub: peer.npub.clone(), + reason: e.to_string(), + })?; + new_by_addr.insert(*identity.node_addr(), peer); + } + + let current_by_addr: HashMap = self + .config + .peers() + .iter() + .filter_map(|peer| { + PeerIdentity::from_npub(&peer.npub) + .ok() + .map(|identity| (*identity.node_addr(), peer.clone())) + }) + .collect(); + + let new_addrs: HashSet<_> = new_by_addr.keys().copied().collect(); + let current_addrs: HashSet<_> = current_by_addr.keys().copied().collect(); + + let removed: Vec<_> = current_addrs.difference(&new_addrs).copied().collect(); + let added: Vec<_> = new_addrs.difference(¤t_addrs).copied().collect(); + let kept: Vec<_> = new_addrs.intersection(¤t_addrs).copied().collect(); + + let mut outcome = crate::node::UpdatePeersOutcome::default(); + let mut refresh_configs = Vec::new(); + + for node_addr in &removed { + if self.retry_pending.remove(node_addr).is_some() { + debug!( + peer = %self.peer_display_name(node_addr), + "Dropping retry entry for peer removed from runtime peer list" + ); + } + self.peer_aliases.remove(node_addr); + outcome.removed += 1; + } + + for node_addr in &kept { + let new_peer = &new_by_addr[node_addr]; + let current_peer = ¤t_by_addr[node_addr]; + let changed = new_peer.addresses != current_peer.addresses + || new_peer.alias != current_peer.alias + || new_peer.connect_policy != current_peer.connect_policy + || new_peer.auto_reconnect != current_peer.auto_reconnect + || new_peer.via_nostr != current_peer.via_nostr; + + if changed { + outcome.updated += 1; + if let Some(state) = self.retry_pending.get_mut(node_addr) { + state.peer_config = new_peer.clone(); + state.retry_after_ms = Self::now_ms(); + } + if let Some(alias) = new_peer.alias.clone() { + self.peer_aliases.insert(*node_addr, alias); + } + } else { + outcome.unchanged += 1; + } + + if new_peer.is_auto_connect() && (!new_peer.addresses.is_empty() || new_peer.via_nostr) + { + refresh_configs.push(new_peer.clone()); + } + } + + let added_configs: Vec<_> = added + .iter() + .map(|node_addr| new_by_addr[node_addr].clone()) + .collect(); + + self.config.peers = new_by_addr.into_values().collect(); + + for peer_config in added_configs { + outcome.added += 1; + let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else { + continue; + }; + let name = peer_config + .alias + .clone() + .unwrap_or_else(|| identity.short_npub()); + self.peer_aliases.insert(*identity.node_addr(), name); + self.register_identity(*identity.node_addr(), identity.pubkey_full()); + + if peer_config.is_auto_connect() + && let Err(err) = self.initiate_peer_connection(&peer_config).await + { + debug!( + npub = %peer_config.npub, + error = %err, + "Failed to initiate connection for newly added runtime peer" + ); + self.schedule_retry(*identity.node_addr(), Self::now_ms()); + } + } + + for peer_config in refresh_configs { + let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else { + continue; + }; + let node_addr = *identity.node_addr(); + + if self.peers.contains_key(&node_addr) { + match self + .try_active_peer_alternative_addresses(&peer_config, identity) + .await + { + Ok(true) => debug!( + peer = %self.peer_display_name(&node_addr), + "Started alternate-path handshake for active peer" + ), + Ok(false) => {} + Err(err) => debug!( + npub = %peer_config.npub, + error = %err, + "Active peer alternate-path refresh did not start" + ), + } + } else { + match self.initiate_peer_connection(&peer_config).await { + Ok(()) => { + if let Some(state) = self.retry_pending.get_mut(&node_addr) { + state.peer_config = peer_config; + state.retry_after_ms = Self::now_ms().saturating_add( + self.config.node.rate_limit.handshake_timeout_secs * 1000, + ); + } + } + Err(err) => { + debug!( + npub = %peer_config.npub, + error = %err, + "Refreshed peer addresses did not initiate a direct connection" + ); + self.schedule_retry(node_addr, Self::now_ms()); + } + } + } + } + + Ok(outcome) + } + /// Initiate connections to configured static peers. /// /// For each peer configured with AutoConnect policy, creates a link and @@ -142,6 +305,25 @@ impl Node { }) } + fn is_connecting_to_peer_on_path( + &self, + peer_node_addr: &NodeAddr, + transport_id: TransportId, + remote_addr: &TransportAddr, + ) -> bool { + self.connections.values().any(|conn| { + conn.expected_identity() + .map(|id| id.node_addr() == peer_node_addr) + .unwrap_or(false) + && conn.transport_id() == Some(transport_id) + && conn.source_addr() == Some(remote_addr) + }) || self.pending_connects.iter().any(|pending| { + pending.peer_identity.node_addr() == peer_node_addr + && pending.transport_id == transport_id + && pending.remote_addr == *remote_addr + }) + } + /// Initiate a connection to a peer on a specific transport and address. /// /// For connectionless transports (UDP, Ethernet): allocates a link, starts @@ -340,6 +522,9 @@ impl Node { pub(super) async fn poll_transport_discovery(&mut self) { // Collect discoveries first to avoid borrow conflict with self let mut to_connect = Vec::new(); + let mut queued_per_peer: HashMap = HashMap::new(); + let mut connect_budget = self.discovery_connect_budget(); + let mut skipped_budget = 0usize; for (transport_id, transport) in &self.transports { if !transport.is_operational() { @@ -366,29 +551,80 @@ impl Node { if node_addr == *self.identity.node_addr() { continue; } - // Skip if already connected + + let candidate_transport_id = *transport_id; + let remote_addr = peer.addr; + if self.peers.contains_key(&node_addr) { - continue; - } - // Skip if connection already in progress - let connecting = self.connections.values().any(|c| { - c.expected_identity() - .map(|id| id.node_addr() == &node_addr) - .unwrap_or(false) - }); - if connecting { + let transport_name = transport.transport_type().name; + let candidate = PeerAddress::new(transport_name, remote_addr.to_string()); + if self.active_peer_candidate_is_fresh_enough_to_skip( + &node_addr, + std::slice::from_ref(&candidate), + ) { + continue; + } + if self.is_connecting_to_peer_on_path( + &node_addr, + candidate_transport_id, + &remote_addr, + ) { + continue; + } + let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0); + if connect_budget == 0 + || self + .path_candidate_attempt_budget(&node_addr) + .saturating_sub(queued_for_peer) + == 0 + { + skipped_budget = skipped_budget.saturating_add(1); + continue; + } + to_connect.push((candidate_transport_id, remote_addr, identity, true)); + *queued_per_peer.entry(node_addr).or_default() += 1; + connect_budget = connect_budget.saturating_sub(1); continue; } - to_connect.push((*transport_id, peer.addr, identity)); + if self.is_connecting_to_peer_on_path( + &node_addr, + candidate_transport_id, + &remote_addr, + ) { + continue; + } + let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0); + if connect_budget == 0 + || self + .path_candidate_attempt_budget(&node_addr) + .saturating_sub(queued_for_peer) + == 0 + { + skipped_budget = skipped_budget.saturating_add(1); + continue; + } + + to_connect.push((candidate_transport_id, remote_addr, identity, false)); + *queued_per_peer.entry(node_addr).or_default() += 1; + connect_budget = connect_budget.saturating_sub(1); } } - for (transport_id, remote_addr, identity) in to_connect { + if skipped_budget > 0 { + debug!( + skipped = skipped_budget, + queued = to_connect.len(), + "Transport discovery connect budget exhausted" + ); + } + + for (transport_id, remote_addr, identity, active_refresh) in to_connect { info!( peer = %self.peer_display_name(identity.node_addr()), transport_id = %transport_id, remote_addr = %remote_addr, + active_refresh, "Auto-connecting to discovered peer" ); if let Err(e) = self @@ -1228,8 +1464,10 @@ impl Node { .max() .unwrap_or(100) .saturating_add(1); + let seen_at_ms = Self::now_ms(); for endpoint in endpoints { - let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, next_priority) + let Some(candidate) = + Self::overlay_endpoint_to_peer_address(&endpoint, next_priority, seen_at_ms) else { continue; }; @@ -1251,17 +1489,27 @@ impl Node { fn overlay_endpoint_to_peer_address( endpoint: &OverlayEndpointAdvert, priority: u8, + seen_at_ms: u64, ) -> Option { let transport = match endpoint.transport { OverlayTransportKind::Udp => "udp", OverlayTransportKind::Tcp => "tcp", OverlayTransportKind::Tor => "tor", }; - Some(PeerAddress::with_priority( - transport, - endpoint.addr.clone(), - priority, - )) + Some( + PeerAddress::with_priority(transport, endpoint.addr.clone(), priority) + .with_seen_at_ms(seen_at_ms), + ) + } + + async fn request_nostr_bootstrap(&self, peer_config: &PeerConfig) -> bool { + let Some(bootstrap) = self.nostr_discovery.clone() else { + debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address"); + return false; + }; + bootstrap.request_connect(peer_config.clone()).await; + info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt"); + true } async fn attempt_peer_address_list( @@ -1271,18 +1519,28 @@ impl Node { allow_bootstrap_nat: bool, addresses: &[PeerAddress], ) -> Result<(), NodeError> { + let peer_node_addr = *peer_identity.node_addr(); + let mut attempted = 0usize; + let max_attempts = self.path_candidate_attempt_budget(&peer_node_addr); + if max_attempts == 0 { + return Err(NodeError::NoTransportForType(format!( + "no outbound slots available for {}", + peer_config.npub + ))); + } + for addr in addresses { + if attempted >= max_attempts { + break; + } if addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat") { if !allow_bootstrap_nat { continue; } - let Some(bootstrap) = self.nostr_discovery.clone() else { - debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address"); - continue; - }; - bootstrap.request_connect(peer_config.clone()).await; - info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt"); - return Ok(()); + if self.request_nostr_bootstrap(peer_config).await { + attempted = attempted.saturating_add(1); + } + continue; } let (transport_id, remote_addr) = if addr.transport == "ethernet" { @@ -1334,11 +1592,15 @@ impl Node { (tid, TransportAddr::from_string(&addr.addr)) }; + if self.is_connecting_to_peer_on_path(&peer_node_addr, transport_id, &remote_addr) { + continue; + } + match self .initiate_connection(transport_id, remote_addr, peer_identity) .await { - Ok(()) => return Ok(()), + Ok(()) => attempted = attempted.saturating_add(1), Err(e @ NodeError::AccessDenied(_)) => return Err(e), Err(e) => { debug!( @@ -1351,6 +1613,10 @@ impl Node { } } + if attempted > 0 { + return Ok(()); + } + Err(NodeError::NoTransportForType(format!( "no operational transport for any of {}'s addresses", peer_config.npub @@ -1427,6 +1693,21 @@ impl Node { } if configured_npubs.contains(&npub) { + if let Ok(peer_identity) = PeerIdentity::from_npub(&npub) { + let node_addr = *peer_identity.node_addr(); + if !self.peers.contains_key(&node_addr) + && !self.is_connecting_to_peer(&node_addr) + && let Some(state) = self.retry_pending.get_mut(&node_addr) + && state.retry_after_ms > now_ms + { + state.retry_after_ms = now_ms; + debug!( + peer = %peer_identity.short_npub(), + caller = %caller, + "open-discovery sweep: fresh configured-peer advert expedited retry" + ); + } + } skipped_configured = skipped_configured.saturating_add(1); continue; } @@ -1467,8 +1748,10 @@ impl Node { let mut addresses = Vec::new(); let mut priority = 120u8; + let seen_at_ms = Self::now_ms(); for endpoint in endpoints { - let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, priority) + let Some(candidate) = + Self::overlay_endpoint_to_peer_address(&endpoint, priority, seen_at_ms) else { continue; }; @@ -1607,6 +1890,61 @@ impl Node { connection_slots.min(peer_slots) } + fn outbound_handshake_slots(&self) -> usize { + let used = self + .connections + .len() + .saturating_add(self.pending_connects.len()); + if self.max_connections == 0 { + usize::MAX + } else { + self.max_connections.saturating_sub(used) + } + } + + fn outbound_link_slots(&self) -> usize { + if self.max_links == 0 { + usize::MAX + } else { + self.max_links.saturating_sub(self.links.len()) + } + } + + fn path_candidate_attempt_budget(&self, peer_node_addr: &NodeAddr) -> usize { + if !self.peers.contains_key(peer_node_addr) + && self.max_peers > 0 + && self.peers.len() >= self.max_peers + { + return 0; + } + + let in_flight_for_peer = self + .connections + .values() + .filter(|conn| { + conn.expected_identity() + .map(|identity| identity.node_addr() == peer_node_addr) + .unwrap_or(false) + }) + .count() + .saturating_add( + self.pending_connects + .iter() + .filter(|pending| pending.peer_identity.node_addr() == peer_node_addr) + .count(), + ); + + self.outbound_handshake_slots() + .min(self.outbound_link_slots()) + .min(MAX_PARALLEL_PATH_CANDIDATES_PER_PEER.saturating_sub(in_flight_for_peer)) + } + + fn discovery_connect_budget(&self) -> usize { + self.outbound_handshake_slots() + .min(self.outbound_link_slots()) + .min(MAX_DISCOVERY_CONNECTS_PER_TICK) + } + fn open_discovery_enqueue_budget(&self, configured_npubs: &HashSet) -> usize { let current_open_discovery_pending = self .retry_pending @@ -1841,47 +2179,159 @@ impl Node { return Ok(()); } - // Static-first dialing: avoid delaying configured address attempts on - // advert fetch/network latency. - let static_addresses = self.static_peer_addresses(peer_config); + let candidates = self.peer_address_candidates(peer_config).await; + + if candidates.is_empty() { + return Err(NodeError::NoTransportForType(format!( + "no addresses known for {}", + peer_config.npub + ))); + } + if self - .attempt_peer_address_list( - peer_config, - peer_identity, - allow_bootstrap_nat, - &static_addresses, - ) + .attempt_peer_address_list(peer_config, peer_identity, allow_bootstrap_nat, &candidates) .await .is_ok() { return Ok(()); } - { - let fallback = self - .nostr_peer_fallback_addresses(peer_config, &static_addresses) - .await; - if !fallback.is_empty() - && self - .attempt_peer_address_list( - peer_config, - peer_identity, - allow_bootstrap_nat, - &fallback, - ) - .await - .is_ok() - { - return Ok(()); - } - } - Err(NodeError::NoTransportForType(format!( "no operational transport for any of {}'s addresses", peer_config.npub ))) } + async fn try_active_peer_alternative_addresses( + &mut self, + peer_config: &PeerConfig, + peer_identity: PeerIdentity, + ) -> Result { + let peer_node_addr = *peer_identity.node_addr(); + let candidates = self.peer_address_candidates(peer_config).await; + + if candidates.is_empty() { + return Err(NodeError::NoTransportForType(format!( + "no addresses known for {}", + peer_config.npub + ))); + } + + let concrete: Vec<_> = candidates + .into_iter() + .filter(|addr| !(addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat"))) + .collect(); + let has_alternative = concrete + .iter() + .any(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr)); + let attempt_candidates: Vec<_> = if has_alternative { + concrete + .into_iter() + .filter(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr)) + .collect() + } else if self.active_peer_needs_same_path_refresh(&peer_node_addr) { + concrete + } else { + Vec::new() + }; + + if attempt_candidates.is_empty() { + return Ok(false); + } + + self.attempt_peer_address_list(peer_config, peer_identity, false, &attempt_candidates) + .await?; + Ok(true) + } + + async fn peer_address_candidates(&self, peer_config: &PeerConfig) -> Vec { + let static_addresses = self.static_peer_addresses(peer_config); + let overlay_addresses = self + .nostr_peer_fallback_addresses(peer_config, &static_addresses) + .await; + + let mut candidates = Vec::with_capacity(overlay_addresses.len() + static_addresses.len()); + for addr in overlay_addresses.into_iter().chain(static_addresses) { + if !candidates.iter().any(|existing: &PeerAddress| { + existing.transport == addr.transport && existing.addr == addr.addr + }) { + candidates.push(addr); + } + } + + candidates.sort_by(|a, b| match (a.seen_at_ms, b.seen_at_ms) { + (Some(a_ts), Some(b_ts)) => b_ts.cmp(&a_ts), + (Some(_), None) => std::cmp::Ordering::Less, + (None, Some(_)) => std::cmp::Ordering::Greater, + (None, None) => std::cmp::Ordering::Equal, + }); + candidates + } + + pub(in crate::node) fn active_peer_candidate_is_fresh_enough_to_skip( + &self, + peer_node_addr: &NodeAddr, + candidates: &[PeerAddress], + ) -> bool { + if !self.active_peer_matches_any_candidate(peer_node_addr, candidates) { + return false; + } + !self.active_peer_needs_same_path_refresh(peer_node_addr) + } + + fn active_peer_needs_same_path_refresh(&self, peer_node_addr: &NodeAddr) -> bool { + let Some(peer) = self.peers.get(peer_node_addr) else { + return false; + }; + let stale_after_ms = self + .config + .node + .heartbeat_interval_secs + .saturating_mul(1000) + .max(1000); + peer.idle_time(Self::now_ms()) > stale_after_ms + } + + fn active_peer_matches_any_candidate( + &self, + peer_node_addr: &NodeAddr, + candidates: &[PeerAddress], + ) -> bool { + candidates + .iter() + .any(|candidate| self.active_peer_matches_candidate(peer_node_addr, candidate)) + } + + fn active_peer_matches_candidate( + &self, + peer_node_addr: &NodeAddr, + candidate: &PeerAddress, + ) -> bool { + let Some(peer) = self.peers.get(peer_node_addr) else { + return false; + }; + let Some(current_addr) = peer.current_addr() else { + return false; + }; + if peer + .transport_id() + .map(|id| self.bootstrap_transports.contains(&id)) + .unwrap_or(false) + { + return false; + } + let current_addr = current_addr.to_string(); + let current_transport = peer + .transport_id() + .and_then(|id| self.transports.get(&id)) + .map(|transport| transport.transport_type().name); + + candidate.addr == current_addr + && current_transport + .map(|transport| transport == candidate.transport) + .unwrap_or(true) + } + // === Control API methods === /// Connect to a peer via the control API. diff --git a/src/node/mod.rs b/src/node/mod.rs index 84d3bef..b66a27f 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -199,6 +199,19 @@ impl fmt::Display for NodeState { } } +/// Reports what changed when replacing the runtime peer list. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct UpdatePeersOutcome { + /// Peers present in the new list but not the previous list. + pub added: usize, + /// Peers removed from the previous list. + pub removed: usize, + /// Existing peers whose configured behavior changed. + pub updated: usize, + /// Existing peers whose comparable config did not change. + pub unchanged: usize, +} + /// Recent request tracking for dedup and reverse-path forwarding. /// /// When a LookupRequest is forwarded through a node, the node stores the diff --git a/src/node/retry.rs b/src/node/retry.rs index 34f7e0e..91f2f08 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -11,6 +11,7 @@ use crate::identity::NodeAddr; use tracing::{debug, info, warn}; // MAX_BACKOFF_MS is now derived from config: node.retry.max_backoff_secs * 1000 +const MAX_RETRY_CONNECTIONS_PER_TICK: usize = 16; /// Tracks retry state for a peer across connection attempts. pub struct RetryState { @@ -248,8 +249,17 @@ impl Node { .filter(|(_, state)| now_ms >= state.retry_after_ms) .map(|(addr, _)| *addr) .collect(); + let deferred = due.len().saturating_sub(MAX_RETRY_CONNECTIONS_PER_TICK); + if deferred > 0 { + debug!( + due = due.len(), + processing = MAX_RETRY_CONNECTIONS_PER_TICK, + deferred, + "Retry processing budget exhausted; deferring remaining peers" + ); + } - for node_addr in due { + for node_addr in due.into_iter().take(MAX_RETRY_CONNECTIONS_PER_TICK) { // Peer may have connected inbound while we waited if self.peers.contains_key(&node_addr) { self.retry_pending.remove(&node_addr); diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index ed654a4..acbb403 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -97,6 +97,58 @@ async fn test_nat_bootstrap_failure_falls_back_to_direct_udp_address() { } } +#[tokio::test] +async fn test_try_peer_addresses_races_all_concrete_udp_candidates() { + let peer_identity = Identity::generate(); + let mut node = make_node(); + let (packet_tx, packet_rx) = packet_channel(64); + node.packet_tx = Some(packet_tx.clone()); + node.packet_rx = Some(packet_rx); + + let transport_id = TransportId::new(1); + let mut udp = UdpTransport::new( + transport_id, + Some("main".to_string()), + crate::config::UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }, + packet_tx, + ); + udp.start_async().await.unwrap(); + node.transports + .insert(transport_id, TransportHandle::Udp(udp)); + + let peer_config = crate::config::PeerConfig { + npub: peer_identity.npub(), + alias: None, + addresses: vec![ + crate::config::PeerAddress::with_priority("udp", "127.0.0.1:9", 1), + crate::config::PeerAddress::with_priority("udp", "127.0.0.1:10", 2), + ], + connect_policy: crate::config::ConnectPolicy::AutoConnect, + auto_reconnect: true, + via_nostr: false, + }; + let peer_identity = PeerIdentity::from_npub(&peer_config.npub).unwrap(); + + node.try_peer_addresses(&peer_config, peer_identity, false) + .await + .unwrap(); + + let mut addrs = node + .connections + .values() + .filter_map(|conn| conn.source_addr().and_then(|addr| addr.as_str())) + .collect::>(); + addrs.sort(); + assert_eq!(addrs, vec!["127.0.0.1:10", "127.0.0.1:9"]); + + for transport in node.transports.values_mut() { + transport.stop().await.ok(); + } +} + #[tokio::test] async fn test_node_state_transitions() { let mut node = make_node(); @@ -712,6 +764,48 @@ fn test_schedule_retry_increments() { assert_eq!(state.retry_after_ms, 11_000 + 20_000); } +/// Retry processing is paced so a large due set cannot start every +/// handshake candidate in one maintenance tick. +#[tokio::test] +async fn test_process_pending_retries_is_budgeted_per_tick() { + let mut node = make_node(); + let mut addrs = Vec::new(); + + for _ in 0..20 { + let identity = Identity::generate(); + let npub = identity.npub(); + let peer_identity = PeerIdentity::from_npub(&npub).unwrap(); + let node_addr = *peer_identity.node_addr(); + node.retry_pending.insert( + node_addr, + crate::node::retry::RetryState { + peer_config: crate::config::PeerConfig::new(npub, "udp", "10.0.0.2:2121"), + retry_count: 0, + retry_after_ms: 0, + reconnect: true, + expires_at_ms: None, + }, + ); + addrs.push(node_addr); + } + + node.process_pending_retries(1).await; + + let processed = addrs + .iter() + .filter(|addr| { + node.retry_pending + .get(addr) + .is_some_and(|state| state.retry_count > 0) + }) + .count(); + let deferred = addrs.len().saturating_sub(processed); + + assert_eq!(processed, 16); + assert_eq!(deferred, 4); + assert_eq!(node.retry_pending.len(), 20); +} + /// Test that auto-connect peers retry indefinitely (never exhaust). #[test] fn test_schedule_retry_auto_connect_never_exhausts() { @@ -864,6 +958,130 @@ async fn test_try_peer_addresses_skips_connecting_peer() { ); } +#[test] +fn active_peer_same_path_discovery_skips_fresh_peer() { + let mut node = make_node(); + let peer_full = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full()); + let peer_node_addr = *peer_identity.node_addr(); + let transport_id = TransportId::new(1); + let current_addr = TransportAddr::from_string("127.0.0.1:9"); + let mut active_peer = ActivePeer::new(peer_identity, LinkId::new(7), Node::now_ms()); + active_peer.set_current_addr(transport_id, current_addr.clone()); + node.peers.insert(peer_node_addr, active_peer); + let candidate = crate::config::PeerAddress::new("udp", "127.0.0.1:9"); + + assert!(node.active_peer_candidate_is_fresh_enough_to_skip( + &peer_node_addr, + std::slice::from_ref(&candidate), + )); +} + +#[test] +fn active_peer_same_path_discovery_refreshes_stale_peer() { + let mut node = make_node(); + let peer_full = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full()); + let peer_node_addr = *peer_identity.node_addr(); + let transport_id = TransportId::new(1); + let current_addr = TransportAddr::from_string("127.0.0.1:9"); + let stale_at = Node::now_ms().saturating_sub( + node.config + .node + .heartbeat_interval_secs + .saturating_add(1) + .saturating_mul(1000), + ); + let mut active_peer = ActivePeer::new(peer_identity, LinkId::new(7), stale_at); + active_peer.set_current_addr(transport_id, current_addr.clone()); + node.peers.insert(peer_node_addr, active_peer); + let candidate = crate::config::PeerAddress::new("udp", "127.0.0.1:9"); + + assert!(!node.active_peer_candidate_is_fresh_enough_to_skip( + &peer_node_addr, + std::slice::from_ref(&candidate), + )); +} + +#[tokio::test] +async fn update_peers_races_new_alternative_without_dropping_active_peer() { + let mut node = make_node(); + let (packet_tx, packet_rx) = packet_channel(64); + node.packet_tx = Some(packet_tx.clone()); + node.packet_rx = Some(packet_rx); + + let transport_id = TransportId::new(1); + let mut udp = UdpTransport::new( + transport_id, + Some("main".to_string()), + crate::config::UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }, + packet_tx, + ); + udp.start_async().await.unwrap(); + node.transports + .insert(transport_id, TransportHandle::Udp(udp)); + + let peer_full = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full()); + let peer_node_addr = *peer_identity.node_addr(); + let current_addr = TransportAddr::from_string("127.0.0.1:9"); + let new_addr = TransportAddr::from_string("127.0.0.1:10"); + let old_link_id = LinkId::new(7); + let mut active_peer = ActivePeer::new(peer_identity, old_link_id, Node::now_ms()); + active_peer.set_current_addr(transport_id, current_addr.clone()); + node.peers.insert(peer_node_addr, active_peer); + node.links.insert( + old_link_id, + Link::connectionless( + old_link_id, + transport_id, + current_addr.clone(), + LinkDirection::Outbound, + Duration::from_millis(100), + ), + ); + + let old_peer = crate::config::PeerConfig { + npub: peer_full.npub(), + alias: None, + addresses: vec![crate::config::PeerAddress::new("udp", "127.0.0.1:9")], + connect_policy: crate::config::ConnectPolicy::AutoConnect, + auto_reconnect: true, + via_nostr: false, + }; + let new_peer = crate::config::PeerConfig { + addresses: vec![ + crate::config::PeerAddress::new("udp", "127.0.0.1:9"), + crate::config::PeerAddress::new("udp", "127.0.0.1:10"), + ], + ..old_peer.clone() + }; + node.config.peers = vec![old_peer]; + + let outcome = node.update_peers(vec![new_peer]).await.unwrap(); + + assert_eq!(outcome.updated, 1); + assert_eq!(node.peer_count(), 1, "existing link must stay live"); + assert_eq!(node.connection_count(), 1); + assert_eq!( + node.connections + .values() + .next() + .and_then(|conn| conn.source_addr()), + Some(&new_addr) + ); + let active = node.get_peer(&peer_node_addr).unwrap(); + assert_eq!(active.link_id(), old_link_id); + assert_eq!(active.current_addr(), Some(¤t_addr)); + + for transport in node.transports.values_mut() { + transport.stop().await.ok(); + } +} + #[tokio::test] async fn test_nostr_traversal_failure_skips_connected_peer() { let mut node = make_node(); diff --git a/src/peer/active.rs b/src/peer/active.rs index 2af00f0..3fa1535 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -593,6 +593,13 @@ impl ActivePeer { self.remote_epoch } + /// Update the remote peer's startup epoch after a successful in-place + /// rekey. Initial handshakes set this through `with_session`, but recovery + /// rekeys also exchange epochs and must keep restart detection current. + pub(crate) fn set_remote_epoch(&mut self, remote_epoch: Option<[u8; 8]>) { + self.remote_epoch = remote_epoch; + } + // === Tree Accessors === /// Get the peer's tree coordinates, if known. @@ -1099,7 +1106,10 @@ impl ActivePeer { /// Takes the stored handshake state, reads msg2, and returns the /// completed NoiseSession. Clears the handshake-related fields but /// leaves rekey_our_index for set_pending_session to use. - pub fn complete_rekey_msg2(&mut self, msg2_bytes: &[u8]) -> Result { + pub fn complete_rekey_msg2( + &mut self, + msg2_bytes: &[u8], + ) -> Result<(NoiseSession, Option<[u8; 8]>), NoiseError> { let mut hs = self .rekey_handshake .take() @@ -1109,13 +1119,14 @@ impl ActivePeer { })?; hs.read_message_2(msg2_bytes)?; + let remote_epoch = hs.remote_epoch(); let session = hs.into_session()?; // Clear msg1 resend state self.rekey_msg1 = None; self.rekey_msg1_next_resend = 0; - Ok(session) + Ok((session, remote_epoch)) } /// Check if msg1 needs resending. From 7d7b551ca19407ce2954a46aa6ca43aa7ab30051 Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Sat, 30 May 2026 01:32:06 +0000 Subject: [PATCH 2/2] discovery: add opt-in mDNS LAN discovery Add scoped mDNS / DNS-SD discovery for peers on the same local link, giving sub-second pairing without a relay or NAT-traversal roundtrip. A node advertises its npub, protocol version, and an optional network scope over link-local multicast, and browses for matching adverts to initiate Noise handshakes against same-LAN peers. LAN discovery is disabled by default; operators enable it with node.discovery.lan.enabled: true. Default-off avoids reintroducing a per-LAN identity broadcast on nodes that have deliberately disabled other discovery channels, and avoids any multicast surprise on upgrade. The startup advertised-port picker now excludes bootstrap transports and selects a non-bootstrap operational UDP transport with a stable lowest-id selector, so the advertised port is deterministic across restarts rather than dependent on HashMap iteration order. This matches the per-dial transport selection used for discovered peers. Co-authored-by: Johnathan Corgan --- CHANGELOG.md | 8 + Cargo.lock | 151 +++++++++++++- Cargo.toml | 1 + src/config/node.rs | 10 + src/discovery.rs | 1 + src/discovery/lan/mod.rs | 368 +++++++++++++++++++++++++++++++++++ src/discovery/lan/tests.rs | 213 ++++++++++++++++++++ src/node/handlers/rx_loop.rs | 1 + src/node/lifecycle.rs | 199 ++++++++++++++++++- src/node/mod.rs | 7 + 10 files changed, 940 insertions(+), 19 deletions(-) create mode 100644 src/discovery/lan/mod.rs create mode 100644 src/discovery/lan/tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index fe9f9cd..03f7739 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 without dropping links to peers that remain in the set. `PeerAddress` gains a `seen_at_ms` recency field (with `with_seen_at_ms`) used to prefer more recently observed addresses. +- Opt-in mDNS / DNS-SD LAN discovery for sub-second pairing of peers on + the same local link, without a relay or NAT-traversal roundtrip. + Disabled by default; operators enable it with + `node.discovery.lan.enabled: true`. Configurable service type and an + optional `node.discovery.lan.scope` that isolates discovery to peers + sharing the same private-network scope. The advertised UDP port is + chosen from a non-bootstrap operational UDP transport using a stable + selector, so it is deterministic across restarts. - `pool_inbound` and `pool_outbound` counters on the TCP and Tor transport stats (`TcpStats`, `TorStats`). Per-direction accounting is updated at every pool-insert and receive-loop-exit site, plus on diff --git a/Cargo.lock b/Cargo.lock index c632966..65e9d65 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1073,6 +1073,7 @@ dependencies = [ "hex", "hkdf", "libc", + "mdns-sd", "nostr", "nostr-sdk", "portable-atomic", @@ -1106,6 +1107,17 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1505,6 +1517,16 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "if-addrs" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "indexmap" version = "2.14.0" @@ -1764,6 +1786,21 @@ dependencies = [ "regex-automata", ] +[[package]] +name = "mdns-sd" +version = "0.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2bb8ce26633738d98ffcef71ec58bff967c6675be50229823c2835f6316e67e" +dependencies = [ + "fastrand", + "flume", + "if-addrs", + "log", + "mio", + "socket-pktinfo", + "socket2", +] + [[package]] name = "memchr" version = "2.8.0" @@ -3008,6 +3045,17 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "socket-pktinfo" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "927136cc2ae6a1b0e66ac6b1210902b75c3f726db004a73bc18686dcd0dcd22f" +dependencies = [ + "libc", + "socket2", + "windows-sys 0.60.2", +] + [[package]] name = "socket2" version = "0.6.3" @@ -3018,6 +3066,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -3914,7 +3971,7 @@ version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -3923,7 +3980,16 @@ version = "0.59.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" dependencies = [ - "windows-targets", + "windows-targets 0.52.6", +] + +[[package]] +name = "windows-sys" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" +dependencies = [ + "windows-targets 0.53.5", ] [[package]] @@ -3941,14 +4007,31 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_aarch64_gnullvm 0.52.6", + "windows_aarch64_msvc 0.52.6", + "windows_i686_gnu 0.52.6", + "windows_i686_gnullvm 0.52.6", + "windows_i686_msvc 0.52.6", + "windows_x86_64_gnu 0.52.6", + "windows_x86_64_gnullvm 0.52.6", + "windows_x86_64_msvc 0.52.6", +] + +[[package]] +name = "windows-targets" +version = "0.53.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" +dependencies = [ + "windows-link", + "windows_aarch64_gnullvm 0.53.1", + "windows_aarch64_msvc 0.53.1", + "windows_i686_gnu 0.53.1", + "windows_i686_gnullvm 0.53.1", + "windows_i686_msvc 0.53.1", + "windows_x86_64_gnu 0.53.1", + "windows_x86_64_gnullvm 0.53.1", + "windows_x86_64_msvc 0.53.1", ] [[package]] @@ -3957,48 +4040,96 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_aarch64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" + [[package]] name = "windows_i686_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" +[[package]] +name = "windows_i686_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" + [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" + [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_i686_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" +[[package]] +name = "windows_x86_64_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "windows_x86_64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" + [[package]] name = "winreg" version = "0.55.0" diff --git a/Cargo.toml b/Cargo.toml index dd1c5d6..bb6b82c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1", features = ["rt", "macros", "signal", "sync", "net", "time", "process", "io-util"] } futures = "0.3" simple-dns = "0.11.2" +mdns-sd = "0.19" socket2 = { version = "0.6.2", features = ["all"] } tokio-socks = "0.5" portable-atomic = { version = "1", features = ["std"] } diff --git a/src/config/node.rs b/src/config/node.rs index 0f6421f..984ac71 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -219,6 +219,12 @@ pub struct DiscoveryConfig { /// Nostr-mediated overlay endpoint discovery. #[serde(default = "DiscoveryConfig::default_nostr")] pub nostr: NostrDiscoveryConfig, + /// mDNS / DNS-SD peer discovery on the local link. Identity surface + /// is a strict subset of what `nostr.advertise` already publishes + /// publicly, so there's no marginal privacy cost; the latency win + /// for same-LAN peers is large (sub-second pairing, no relay). + #[serde(default = "DiscoveryConfig::default_lan")] + pub lan: crate::discovery::lan::LanDiscoveryConfig, } impl Default for DiscoveryConfig { @@ -231,6 +237,7 @@ impl Default for DiscoveryConfig { backoff_max_secs: 0, forward_min_interval_secs: 2, nostr: NostrDiscoveryConfig::default(), + lan: crate::discovery::lan::LanDiscoveryConfig::default(), } } } @@ -257,6 +264,9 @@ impl DiscoveryConfig { fn default_nostr() -> NostrDiscoveryConfig { NostrDiscoveryConfig::default() } + fn default_lan() -> crate::discovery::lan::LanDiscoveryConfig { + crate::discovery::lan::LanDiscoveryConfig::default() + } } /// Nostr advert discovery policy. diff --git a/src/discovery.rs b/src/discovery.rs index 2d86d22..3612fb5 100644 --- a/src/discovery.rs +++ b/src/discovery.rs @@ -6,6 +6,7 @@ //! it hands the live socket and selected remote endpoint to FIPS so the //! existing Noise/FMP transport path can take over. +pub mod lan; pub mod nostr; use crate::config::UdpConfig; diff --git a/src/discovery/lan/mod.rs b/src/discovery/lan/mod.rs new file mode 100644 index 0000000..f5e3b2d --- /dev/null +++ b/src/discovery/lan/mod.rs @@ -0,0 +1,368 @@ +//! LAN peer discovery via mDNS / DNS-SD (RFC 6762 / RFC 6763). +//! +//! Publishes a `_fips._udp.local.` service advert carrying our `npub` and +//! optional discovery scope on the local link, and concurrently browses for the +//! same service type to learn peers reachable on the same broadcast +//! domain. The result is sub-second peer pairing without any Nostr-relay +//! roundtrip, STUN observation, or NAT traversal — the observed +//! endpoint is by construction routable from the consumer's LAN. +//! +//! ## Trust model +//! +//! mDNS adverts are unauthenticated: anyone on the LAN can multicast a +//! TXT carrying `npub=...`. Identity is still proven end-to-end by the +//! Noise XX handshake the Node initiates against the observed endpoint +//! — a spoofed advert with another peer's npub fails the handshake and +//! is silently dropped. Treat the mDNS advert as a routing hint, not as +//! identity. LAN discovery is link-local mDNS only. It is not a Nostr advert +//! and does not leave the broadcast domain unless the operator's LAN bridges +//! mDNS. +//! +//! ## Scope filtering +//! +//! When a `discovery_scope` is configured, the advert carries it in a +//! `scope=` TXT entry and the browser only surfaces peers with a +//! matching scope. Nodes on the same physical LAN but configured for +//! different mesh networks don't cross-feed each other. + +use std::collections::HashMap; +use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6}; +use std::sync::Arc; +use std::time::Instant; + +use mdns_sd::{ScopedIp, ServiceDaemon, ServiceEvent, ServiceInfo}; +use thiserror::Error; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +use crate::Identity; + +/// DNS-SD service type for the FIPS LAN advert. RFC 6763 §4.1.2: must +/// end with `.local.`. The `_udp` is the IP transport, not the upper +/// protocol — both UDP and TCP FIPS endpoints announce under the same +/// service type because the link-layer punch/handshake travels over UDP +/// either way. +pub const SERVICE_TYPE: &str = "_fips._udp.local."; + +/// TXT key carrying the bech32-encoded npub of the publishing node. +pub const TXT_KEY_NPUB: &str = "npub"; + +/// TXT key carrying the publishing node's `discovery_scope`, if any. +pub const TXT_KEY_SCOPE: &str = "scope"; + +/// TXT key carrying the FIPS protocol version (matches the Nostr advert +/// `PROTOCOL_VERSION`). +pub const TXT_KEY_VERSION: &str = "v"; + +#[derive(Debug, Error)] +pub enum LanDiscoveryError { + #[error("mDNS daemon init failed: {0}")] + Daemon(String), + #[error("mDNS register failed: {0}")] + Register(String), + #[error("mDNS browse failed: {0}")] + Browse(String), + #[error("no advertised UDP port — start a UDP transport first")] + NoAdvertisedPort, + #[error("LAN discovery disabled in config")] + Disabled, +} + +/// A peer we learned about via mDNS. Identity is unverified at this +/// point; the Node initiates a Noise XX handshake against `addr` to +/// confirm `npub` actually controls the matching private key. +#[derive(Debug, Clone)] +pub struct LanDiscoveredPeer { + pub npub: String, + pub scope: Option, + pub addr: SocketAddr, + pub observed_at: Instant, +} + +/// Browser-side events surfaced by `LanDiscovery::drain_events`. +#[derive(Debug, Clone)] +pub enum LanEvent { + Discovered(LanDiscoveredPeer), +} + +/// Runtime configuration for the mDNS responder + browser. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct LanDiscoveryConfig { + /// Master switch. Default: `false` — LAN discovery is opt-in. Operators + /// who want sub-second same-LAN pairing enable it via + /// `node.discovery.lan.enabled: true`. Default-off avoids reintroducing + /// a per-LAN identity broadcast on nodes that have deliberately disabled + /// other discovery channels, and avoids any multicast surprise on upgrade. + #[serde(default = "LanDiscoveryConfig::default_enabled")] + pub enabled: bool, + /// Overridable service type, primarily so integration tests can run + /// multiple isolated services on the same loopback interface. + #[serde(default = "LanDiscoveryConfig::default_service_type")] + pub service_type: String, + /// Optional application/network scope carried in the LAN-only TXT + /// record. Browsers that set a scope ignore adverts for other scopes. + /// + /// This is intentionally separate from Nostr discovery's public `app` + /// tag so applications can keep relay-visible adverts generic while + /// still isolating LAN discovery per private network. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scope: Option, +} + +impl Default for LanDiscoveryConfig { + fn default() -> Self { + Self { + enabled: Self::default_enabled(), + service_type: Self::default_service_type(), + scope: None, + } + } +} + +impl LanDiscoveryConfig { + fn default_enabled() -> bool { + false + } + fn default_service_type() -> String { + SERVICE_TYPE.to_string() + } +} + +/// Running mDNS responder + browser bound to the node's UDP advert port. +pub struct LanDiscovery { + daemon: ServiceDaemon, + own_npub: String, + instance_fullname: String, + events_rx: Mutex>, + event_pump: tokio::task::JoinHandle<()>, +} + +impl LanDiscovery { + /// Start the mDNS responder and browser. + /// + /// `advertised_port` is the UDP port the operational UDP transport + /// is bound to — peers receiving our advert will initiate Noise XX + /// against that port. `scope` mirrors the Nostr discovery scope and + /// is used to filter the browser stream. + pub async fn start( + identity: &Identity, + scope: Option, + advertised_port: u16, + config: LanDiscoveryConfig, + ) -> Result, LanDiscoveryError> { + if !config.enabled { + return Err(LanDiscoveryError::Disabled); + } + if advertised_port == 0 { + return Err(LanDiscoveryError::NoAdvertisedPort); + } + + let daemon = ServiceDaemon::new().map_err(|e| LanDiscoveryError::Daemon(e.to_string()))?; + + let npub = identity.npub(); + // mDNS DNS labels are capped at 63 bytes. 16 bech32 chars of npub + // give 80 bits of effective entropy — collisions on a single LAN + // are vanishingly unlikely. Prefixed for human-readable logs. + let label_npub = &npub[..16.min(npub.len())]; + let instance_name = format!("fips-{label_npub}"); + let host_name = format!("{instance_name}.local."); + + let mut props: HashMap = HashMap::new(); + props.insert(TXT_KEY_NPUB.to_string(), npub.clone()); + if let Some(s) = scope.as_deref() + && !s.is_empty() + { + props.insert(TXT_KEY_SCOPE.to_string(), s.to_string()); + } + props.insert( + TXT_KEY_VERSION.to_string(), + super::nostr::PROTOCOL_VERSION.to_string(), + ); + + // host_ipv4 is set to "127.0.0.1" *and* enable_addr_auto() is + // called: the loopback seed makes the advert resolve for + // same-host peers (and same-host integration tests) while the + // auto-flag still appends every non-loopback interface address + // mdns-sd discovers. Belt-and-braces because addr_auto alone + // skips loopback by default on some platforms. + let service_info = ServiceInfo::new( + &config.service_type, + &instance_name, + &host_name, + "127.0.0.1", + advertised_port, + Some(props), + ) + .map_err(|e| LanDiscoveryError::Register(e.to_string()))? + .enable_addr_auto(); + + let instance_fullname = service_info.get_fullname().to_string(); + + daemon + .register(service_info) + .map_err(|e| LanDiscoveryError::Register(e.to_string()))?; + + let browse_rx = daemon + .browse(&config.service_type) + .map_err(|e| LanDiscoveryError::Browse(e.to_string()))?; + + let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel(); + let own_npub = npub.clone(); + let scope_filter = scope.clone().filter(|s| !s.is_empty()); + let event_pump = tokio::spawn(async move { + // mdns-sd browse returns a flume::Receiver; pump until the + // daemon shuts down and the channel closes. + loop { + let event = match browse_rx.recv_async().await { + Ok(e) => e, + Err(_) => break, + }; + match event { + ServiceEvent::ServiceResolved(info) => { + let mut peer_npub: Option = None; + let mut peer_scope: Option = None; + for prop in info.get_properties().iter() { + match prop.key() { + TXT_KEY_NPUB => { + peer_npub = Some(prop.val_str().to_string()); + } + TXT_KEY_SCOPE => { + peer_scope = Some(prop.val_str().to_string()); + } + _ => {} + } + } + let Some(peer_npub) = peer_npub else { + debug!( + instance = info.get_fullname(), + "lan: skip advert without npub TXT" + ); + continue; + }; + if peer_npub == own_npub { + // Our own advert echoed back on a loopback + // or multi-homed interface. + continue; + } + if scope_filter.is_some() && scope_filter != peer_scope { + debug!( + npub = %short(&peer_npub), + their_scope = ?peer_scope, + our_scope = ?scope_filter, + "lan: skip cross-scope advert" + ); + continue; + } + let port = info.get_port(); + if port == 0 { + continue; + } + let observed_at = Instant::now(); + // mdns-sd may report multiple interface IPs for + // a multi-homed responder. Surface all routable + // candidates — the Node side filters/dedups and + // only dials addresses compatible with an active + // UDP socket family. IPv6 link-local addresses + // require an interface scope; preserve it when + // mdns-sd provides one, and skip unusable + // scope-less link-local records. + for scoped in info.get_addresses() { + let Some(addr) = socket_addr_from_scoped_ip(scoped, port) else { + debug!( + npub = %short(&peer_npub), + addr = %scoped.to_ip_addr(), + "lan: skip scope-less IPv6 link-local advert" + ); + continue; + }; + if events_tx + .send(LanEvent::Discovered(LanDiscoveredPeer { + npub: peer_npub.clone(), + scope: peer_scope.clone(), + addr, + observed_at, + })) + .is_err() + { + return; + } + } + } + ServiceEvent::ServiceRemoved(_, fullname) => { + debug!(fullname = %fullname, "lan: service removed"); + } + other => { + debug!(?other, "lan: mDNS event"); + } + } + } + }); + + info!( + instance = %instance_fullname, + port = advertised_port, + scope = ?scope, + "lan: mDNS discovery started" + ); + Ok(Arc::new(Self { + daemon, + own_npub: npub, + instance_fullname, + events_rx: Mutex::new(events_rx), + event_pump, + })) + } + + /// Bech32 npub published by this node. + pub fn own_npub(&self) -> &str { + &self.own_npub + } + + /// Drain pending browser events. Called once per Node tick. + pub async fn drain_events(&self) -> Vec { + let mut rx = self.events_rx.lock().await; + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + events + } + + /// Tear down the responder, browser, and event pump. + pub async fn shutdown(self: &Arc) { + if let Err(e) = self.daemon.unregister(&self.instance_fullname) { + warn!(error = %e, "lan: unregister failed"); + } + if let Err(e) = self.daemon.shutdown() { + warn!(error = %e, "lan: daemon shutdown failed"); + } + self.event_pump.abort(); + } +} + +fn short(npub: &str) -> &str { + let end = 16.min(npub.len()); + &npub[..end] +} + +fn socket_addr_from_scoped_ip(scoped: &ScopedIp, port: u16) -> Option { + match scoped { + ScopedIp::V4(v4) => Some(SocketAddr::V4(SocketAddrV4::new(*v4.addr(), port))), + ScopedIp::V6(v6) => { + let ip = *v6.addr(); + let scope_id = v6.scope_id().index; + if ipv6_is_unicast_link_local(ip) && scope_id == 0 { + return None; + } + Some(SocketAddr::V6(SocketAddrV6::new(ip, port, 0, scope_id))) + } + _ => None, + } +} + +fn ipv6_is_unicast_link_local(ip: std::net::Ipv6Addr) -> bool { + (ip.segments()[0] & 0xffc0) == 0xfe80 +} + +#[cfg(test)] +mod tests; diff --git a/src/discovery/lan/tests.rs b/src/discovery/lan/tests.rs new file mode 100644 index 0000000..1e4867c --- /dev/null +++ b/src/discovery/lan/tests.rs @@ -0,0 +1,213 @@ +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::time::Duration; + +use crate::Identity; +use mdns_sd::ScopedIp; + +use super::{LanDiscovery, LanDiscoveryConfig, LanEvent}; + +/// Distinct service type per test run so concurrent cargo-test workers +/// on the same machine don't cross-feed each other's adverts via the +/// shared 224.0.0.251 multicast group. The trailing `.local.` is +/// required by RFC 6763 — mdns-sd will reject anything else. +fn isolated_service_type(tag: &str) -> String { + let rand: u32 = rand::random(); + format!("_fipstest-{tag}-{rand:08x}._udp.local.") +} + +fn config_for(service_type: String) -> LanDiscoveryConfig { + LanDiscoveryConfig { + enabled: true, + service_type, + scope: None, + } +} + +#[test] +fn scoped_ipv4_advert_becomes_socket_addr() { + let scoped = ScopedIp::from(IpAddr::V4(Ipv4Addr::new(192, 168, 178, 91))); + let addr = super::socket_addr_from_scoped_ip(&scoped, 51820); + + assert_eq!(addr, Some(SocketAddr::from(([192, 168, 178, 91], 51820)))); +} + +#[test] +fn scope_less_ipv6_link_local_advert_is_skipped() { + let scoped = ScopedIp::from(IpAddr::V6("fe80::32c5:99ff:fea7:5fe9".parse().unwrap())); + + assert!(super::socket_addr_from_scoped_ip(&scoped, 51820).is_none()); +} + +#[test] +fn non_link_local_ipv6_advert_is_preserved() { + let scoped = ScopedIp::from(IpAddr::V6(Ipv6Addr::LOCALHOST)); + let addr = super::socket_addr_from_scoped_ip(&scoped, 51820); + + assert_eq!(addr, Some("[::1]:51820".parse().unwrap())); +} + +async fn wait_for_peer( + discovery: &LanDiscovery, + expected_npub: &str, + timeout: Duration, +) -> Option { + let deadline = tokio::time::Instant::now() + timeout; + while tokio::time::Instant::now() < deadline { + for event in discovery.drain_events().await { + let LanEvent::Discovered(peer) = event; + if peer.npub == expected_npub { + return Some(peer); + } + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + None +} + +/// Two LanDiscovery instances on isolated service types — `a` browses +/// only its own type and never sees `b`, and vice versa. Sanity check +/// that the scope-isolation defense works (we'd lose isolation if mdns- +/// sd ever leaked across service types). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn isolated_service_types_do_not_cross_feed() { + let identity_a = Identity::generate(); + let identity_b = Identity::generate(); + + let service_a = isolated_service_type("isolated-a"); + let service_b = isolated_service_type("isolated-b"); + + let lan_a = LanDiscovery::start( + &identity_a, + Some("scope-x".to_string()), + 61001, + config_for(service_a.clone()), + ) + .await + .expect("start a"); + let lan_b = LanDiscovery::start( + &identity_b, + Some("scope-x".to_string()), + 61002, + config_for(service_b.clone()), + ) + .await + .expect("start b"); + + // Give mDNS multicast time to settle, then confirm neither side saw + // the other (different service type isolates them). + tokio::time::sleep(Duration::from_secs(2)).await; + let saw_b_from_a = wait_for_peer( + &lan_a, + identity_b.npub().as_str(), + Duration::from_millis(500), + ) + .await + .is_some(); + let saw_a_from_b = wait_for_peer( + &lan_b, + identity_a.npub().as_str(), + Duration::from_millis(500), + ) + .await + .is_some(); + + lan_a.shutdown().await; + lan_b.shutdown().await; + + assert!(!saw_b_from_a, "isolated service types must not cross-feed"); + assert!(!saw_a_from_b, "isolated service types must not cross-feed"); +} + +/// Two LanDiscovery instances on the same service type and the same +/// scope: each should observe the other's advert within a few seconds. +/// Exercises the responder + browser + TXT plumbing end-to-end. +/// +/// Ignored by default: relies on multicast-loopback semantics that +/// vary across macOS/Linux/Windows when two `ServiceDaemon` instances +/// run in the same process. Real cross-host LAN deployment exercises +/// the same code path correctly — verify with `cargo test -- --ignored +/// matched_scope_peers_observe_each_other` on a setup where this +/// matters, or via end-to-end integration with two daemons. +#[ignore] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn matched_scope_peers_observe_each_other() { + let identity_a = Identity::generate(); + let identity_b = Identity::generate(); + + let service = isolated_service_type("matched"); + + let lan_a = LanDiscovery::start( + &identity_a, + Some("scope-shared".to_string()), + 61101, + config_for(service.clone()), + ) + .await + .expect("start a"); + let lan_b = LanDiscovery::start( + &identity_b, + Some("scope-shared".to_string()), + 61102, + config_for(service.clone()), + ) + .await + .expect("start b"); + + // Loopback mDNS resolution on macOS/Linux takes a moment. + let observed_b = + wait_for_peer(&lan_a, identity_b.npub().as_str(), Duration::from_secs(10)).await; + let observed_a = + wait_for_peer(&lan_b, identity_a.npub().as_str(), Duration::from_secs(10)).await; + + lan_a.shutdown().await; + lan_b.shutdown().await; + + let observed_b = observed_b.expect("a must see b"); + let observed_a = observed_a.expect("b must see a"); + + assert_eq!(observed_b.scope.as_deref(), Some("scope-shared")); + assert_eq!(observed_a.scope.as_deref(), Some("scope-shared")); + assert_eq!(observed_b.addr.port(), 61102); + assert_eq!(observed_a.addr.port(), 61101); +} + +/// Different scopes on the same service type must be filtered out by +/// the browser — peer in scope X does not surface to a browser in +/// scope Y, even if both adverts arrive on the same multicast group. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cross_scope_advert_is_filtered() { + let identity_a = Identity::generate(); + let identity_b = Identity::generate(); + + let service = isolated_service_type("cross-scope"); + + let lan_a = LanDiscovery::start( + &identity_a, + Some("scope-a".to_string()), + 61201, + config_for(service.clone()), + ) + .await + .expect("start a"); + let lan_b = LanDiscovery::start( + &identity_b, + Some("scope-b".to_string()), + 61202, + config_for(service.clone()), + ) + .await + .expect("start b"); + + tokio::time::sleep(Duration::from_secs(3)).await; + let saw_b = wait_for_peer( + &lan_a, + identity_b.npub().as_str(), + Duration::from_millis(500), + ) + .await; + + lan_a.shutdown().await; + lan_b.shutdown().await; + + assert!(saw_b.is_none(), "cross-scope advert must be filtered"); +} diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index ce3dccc..33812af 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -252,6 +252,7 @@ impl Node { self.reload_peer_acl(); self.poll_pending_connects().await; self.poll_nostr_discovery().await; + self.poll_lan_discovery().await; self.resend_pending_handshakes(now_ms).await; self.resend_pending_rekeys(now_ms).await; self.resend_pending_session_handshakes(now_ms).await; diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 4d3c326..771cb32 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -15,6 +15,7 @@ use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, use crate::upper::tun::{TunDevice, TunState, run_tun_reader, shutdown_tun_interface}; use crate::{NodeAddr, PeerIdentity}; use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; use std::thread; use std::time::Duration; use tracing::{debug, info, warn}; @@ -23,6 +24,13 @@ const OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER: u64 = 2; const MAX_PARALLEL_PATH_CANDIDATES_PER_PEER: usize = 4; const MAX_DISCOVERY_CONNECTS_PER_TICK: usize = 16; +fn socket_addr_families_compatible(local: SocketAddr, remote: SocketAddr) -> bool { + matches!( + (local, remote), + (SocketAddr::V4(_), SocketAddr::V4(_)) | (SocketAddr::V6(_), SocketAddr::V6(_)) + ) +} + impl Node { /// Replace the runtime peer list. /// @@ -324,6 +332,31 @@ impl Node { }) } + /// Find a UDP transport whose bound socket can send to `remote_addr`. + /// + /// LAN discovery can surface both IPv4 and IPv6 addresses for the same + /// service. A wildcard IPv4 socket cannot send to an IPv6 link-local + /// target, and vice versa, so callers must choose by socket family rather + /// than by transport type alone. + fn find_udp_transport_for_remote_addr( + &self, + remote_addr: SocketAddr, + ) -> Option<(TransportId, SocketAddr)> { + self.transports + .iter() + .filter(|(id, handle)| { + handle.transport_type().name == "udp" + && handle.is_operational() + && !self.bootstrap_transports.contains(id) + }) + .filter_map(|(id, handle)| { + let local_addr = handle.local_addr()?; + socket_addr_families_compatible(local_addr, remote_addr) + .then_some((*id, local_addr)) + }) + .min_by_key(|(id, _)| id.as_u32()) + } + /// Initiate a connection to a peer on a specific transport and address. /// /// For connectionless transports (UDP, Ethernet): allocates a link, starts @@ -829,6 +862,90 @@ impl Node { self.queue_open_discovery_retries(&bootstrap).await; } + /// Resolve the LAN-only discovery scope. Applications with explicit + /// connectivity config can set `node.discovery.lan.scope` without + /// changing the public Nostr discovery `app` tag. The older fallback + /// extracts a scope from the Nostr app tag used by default scoped + /// discovery. + pub(super) fn lan_discovery_scope(&self) -> Option { + if let Some(scope) = self.config.node.discovery.lan.scope.as_deref() { + let scope = scope.trim(); + if !scope.is_empty() { + return Some(scope.to_string()); + } + } + + let app = self.config.node.discovery.nostr.app.trim(); + if app.is_empty() { + return None; + } + if let Some(rest) = app.strip_prefix("fips-overlay-v1:") { + let scope = rest.trim(); + if scope.is_empty() { + None + } else { + Some(scope.to_string()) + } + } else { + Some(app.to_string()) + } + } + + /// Drain mDNS-discovered peers and initiate Noise XX handshakes. + /// The handshake itself is the authentication — a spoofed mDNS advert + /// with someone else's npub fails the XX exchange and is dropped. + pub(super) async fn poll_lan_discovery(&mut self) { + let Some(runtime) = self.lan_discovery.clone() else { + return; + }; + let events = runtime.drain_events().await; + if events.is_empty() { + return; + } + for event in events { + let crate::discovery::lan::LanEvent::Discovered(peer) = event; + let Some((transport_id, local_addr)) = + self.find_udp_transport_for_remote_addr(peer.addr) + else { + debug!( + addr = %peer.addr, + "lan: skip discovered peer with no compatible UDP transport" + ); + continue; + }; + let identity = match crate::PeerIdentity::from_npub(&peer.npub) { + Ok(id) => id, + Err(err) => { + debug!(npub = %peer.npub, error = %err, "lan: skip bad npub"); + continue; + } + }; + let peer_node_addr = *identity.node_addr(); + let remote_addr = crate::transport::TransportAddr::from_string(&peer.addr.to_string()); + if self.peers.contains_key(&peer_node_addr) + || self.is_connecting_to_peer(&peer_node_addr) + { + continue; + } + info!( + npub = %identity.short_npub(), + addr = %peer.addr, + local_addr = %local_addr, + "lan: initiating handshake to discovered peer" + ); + if let Err(err) = self + .initiate_connection(transport_id, remote_addr, identity) + .await + { + debug!( + npub = %peer.npub, + error = %err, + "lan: failed to initiate connection to discovered peer" + ); + } + } + } + /// Poll pending transport connects and initiate handshakes for ready ones. /// /// Called from the tick handler. For each pending connect, queries the @@ -1029,6 +1146,47 @@ impl Node { } } + // mDNS / DNS-SD LAN discovery. Independent of Nostr — runs even + // when Nostr is disabled, since it gives us sub-second pairing + // on the same link without any relay or NAT-traversal roundtrip. + if self.config.node.discovery.lan.enabled { + // Advertise the port of a non-bootstrap operational UDP transport. + // Bootstrap transports must be excluded (they are not the node's + // listening data-plane socket), and a stable selector (lowest + // TransportId) is used so the advertised port is deterministic + // across restarts rather than dependent on HashMap iteration + // order. This mirrors find_udp_transport_for_remote_addr. + let advertised_udp_port = self + .transports + .iter() + .filter(|(id, h)| { + h.transport_type().name == "udp" + && h.is_operational() + && !self.bootstrap_transports.contains(id) + }) + .filter_map(|(id, h)| h.local_addr().map(|addr| (*id, addr.port()))) + .min_by_key(|(id, _)| id.as_u32()) + .map(|(_, port)| port) + .unwrap_or(0); + let scope = self.lan_discovery_scope(); + match crate::discovery::lan::LanDiscovery::start( + &self.identity, + scope, + advertised_udp_port, + self.config.node.discovery.lan.clone(), + ) + .await + { + Ok(runtime) => { + self.lan_discovery = Some(runtime); + info!("LAN mDNS discovery enabled"); + } + Err(err) => { + debug!(error = %err, "LAN mDNS discovery not started"); + } + } + } + // Connect to static peers before TUN is active // This allows handshake messages to be sent before we start accepting packets self.initiate_peer_connections().await; @@ -1314,6 +1472,13 @@ impl Node { warn!(error = %e, "Failed to shutdown Nostr overlay discovery"); } + // Tear down LAN mDNS responder + browser. Best-effort: the + // OS will eventually time the advert out via its TTL even if + // we don't get a clean unregister out before the daemon exits. + if let Some(lan) = self.lan_discovery.take() { + lan.shutdown().await; + } + // Shutdown transports (they're packet producers) let transport_ids: Vec<_> = self.transports.keys().cloned().collect(); for transport_id in transport_ids { @@ -1578,15 +1743,31 @@ impl Node { continue; } } else { - let tid = match self.find_transport_for_type(&addr.transport) { - Some(id) => id, - None => { - debug!( - transport = %addr.transport, - addr = %addr.addr, - "No operational transport for address type" - ); - continue; + let tid = if addr.transport == "udp" + && let Ok(remote_socket_addr) = addr.addr.parse::() + { + match self.find_udp_transport_for_remote_addr(remote_socket_addr) { + Some((id, _)) => id, + None => { + debug!( + transport = %addr.transport, + addr = %addr.addr, + "No compatible operational UDP transport for address" + ); + continue; + } + } + } else { + match self.find_transport_for_type(&addr.transport) { + Some(id) => id, + None => { + debug!( + transport = %addr.transport, + addr = %addr.addr, + "No operational transport for address type" + ); + continue; + } } }; (tid, TransportAddr::from_string(&addr.addr)) diff --git a/src/node/mod.rs b/src/node/mod.rs index b66a27f..f53ae19 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -463,6 +463,11 @@ pub struct Node { /// Optional Nostr/STUN overlay discovery coordinator for `udp:nat` peers. nostr_discovery: Option>, + /// mDNS / DNS-SD responder + browser for local-link peer discovery. + /// Identity is unverified at this layer — the Noise XX handshake + /// initiated against an mDNS-observed endpoint is what proves the + /// peer holds the matching private key. + lan_discovery: Option>, /// Wall-clock ms when Nostr discovery successfully started, used to /// schedule the one-shot startup advert sweep after a settle delay. /// `None` until discovery comes up; remains `None` if discovery is @@ -682,6 +687,7 @@ impl Node { retry_pending: HashMap::new(), nostr_discovery: None, nostr_discovery_started_at_ms: None, + lan_discovery: None, startup_open_discovery_sweep_done: false, bootstrap_transports: HashSet::new(), bootstrap_transport_npubs: HashMap::new(), @@ -827,6 +833,7 @@ impl Node { retry_pending: HashMap::new(), nostr_discovery: None, nostr_discovery_started_at_ms: None, + lan_discovery: None, startup_open_discovery_sweep_done: false, bootstrap_transports: HashSet::new(), bootstrap_transport_npubs: HashMap::new(),