1use crate::{
6 Error,
7 peer_external::{Externals, Verdict},
8 peer_gossip::{is_loopback, is_public},
9 peer_limits::{Class, Classes},
10 peer_nat::{Nat, PeerLink, Reachability},
11 peer_transport::{Behaviour, BehaviourEvent},
12};
13use libp2p::{
14 Multiaddr, PeerId, Swarm, autonat,
15 core::{ConnectedPoint, transport::ListenerId},
16 dcutr,
17 multiaddr::Protocol,
18 relay,
19 swarm::{ConnectionId, SwarmEvent},
20};
21use std::{
22 collections::{HashMap, HashSet},
23 io,
24 sync::{
25 Arc,
26 atomic::{AtomicBool, Ordering},
27 },
28 time::{Duration, Instant},
29};
30
31pub(crate) struct Remote {
33 pub(crate) peer: PeerId,
34 pub(crate) address: Multiaddr,
35 pub(crate) outbound: bool,
36 pub(crate) class: Class,
37 since: Instant,
38 demoted: Option<Instant>,
39}
40
41#[derive(Default)]
44pub(crate) struct Remotes(HashMap<ConnectionId, Remote>);
45
46impl Remotes {
47 pub(crate) fn admit(
49 &mut self,
50 (id, peer): (ConnectionId, PeerId),
51 endpoint: &ConnectedPoint,
52 classes: &Classes,
53 ) -> Class {
54 let class = match classes.of(id) {
55 Some(class) => class,
56 None => {
57 tracing::warn!(%peer, "connection without an admission class; kept off data");
58 Class::Auxiliary
59 }
60 };
61 let remote = Remote {
62 peer,
63 address: endpoint.get_remote_address().clone(),
64 outbound: endpoint.is_dialer(),
65 class,
66 since: Instant::now(),
67 demoted: None,
68 };
69 self.0.insert(id, remote);
70 class
71 }
72
73 pub(crate) fn remove(&mut self, id: ConnectionId) -> Option<Remote> {
74 self.0.remove(&id)
75 }
76
77 pub(crate) fn loopback(&self, id: ConnectionId) -> bool {
78 self.0
79 .get(&id)
80 .is_some_and(|remote| is_loopback(&remote.address))
81 }
82
83 pub(crate) fn outbound_direct(&self, id: ConnectionId) -> bool {
85 self.0
86 .get(&id)
87 .is_some_and(|remote| remote.outbound && remote.class != Class::Relayed)
88 }
89
90 pub(crate) fn superseded(
97 &self,
98 id: ConnectionId,
99 peer: PeerId,
100 class: Class,
101 ) -> Vec<ConnectionId> {
102 let inbound = self.0.get(&id).is_some_and(|remote| !remote.outbound);
103 match class {
104 Class::Data => self.relayed_of(peer),
105 Class::Relayed if inbound => self.direct_of(peer),
106 Class::Relayed if self.has_data(peer) => vec![id],
107 Class::Relayed | Class::Auxiliary | Class::Holder | Class::Demoted => Vec::new(),
108 }
109 }
110
111 fn direct_of(&self, peer: PeerId) -> Vec<ConnectionId> {
112 self.0
113 .iter()
114 .filter(|(_, remote)| remote.peer == peer && remote.class != Class::Relayed)
115 .map(|(id, _)| *id)
116 .collect()
117 }
118
119 fn retirable(&self, grace: Duration) -> Vec<(ConnectionId, PeerId, bool)> {
123 let mut retirable: Vec<(ConnectionId, PeerId, bool)> = self
124 .0
125 .iter()
126 .filter(|(_, remote)| match remote.class {
127 Class::Holder => remote.since.elapsed() >= grace,
128 Class::Demoted => remote.demoted.is_some_and(|at| at.elapsed() >= grace),
129 Class::Data | Class::Auxiliary | Class::Relayed => false,
130 })
131 .map(|(id, remote)| {
132 let seeker = remote.class == Class::Demoted && Self::candidate(remote);
133 (*id, remote.peer, seeker)
134 })
135 .collect();
136 retirable.sort_by_key(|(id, ..)| *id);
137 retirable
138 }
139
140 pub(crate) fn data_of(&self, peer: PeerId) -> Vec<(ConnectionId, bool)> {
142 self.0
143 .iter()
144 .filter(|(_, remote)| remote.peer == peer && remote.class == Class::Data)
145 .map(|(id, remote)| (*id, remote.outbound))
146 .collect()
147 }
148
149 pub(crate) fn reclass(&mut self, id: ConnectionId, class: Class) {
150 if let Some(remote) = self.0.get_mut(&id) {
151 remote.class = class;
152 remote.demoted = (class == Class::Demoted).then(Instant::now);
153 }
154 }
155
156 pub(crate) fn unproven(
159 &self,
160 grace: Duration,
161 proven: impl Fn(PeerId) -> bool,
162 ) -> HashSet<PeerId> {
163 self.0
164 .values()
165 .filter(|remote| remote.class == Class::Data && remote.since.elapsed() >= grace)
166 .map(|remote| remote.peer)
167 .filter(|peer| !proven(*peer))
168 .collect()
169 }
170
171 pub(crate) fn demoted_after(&self, grace: Duration) -> Vec<(ConnectionId, PeerId, bool)> {
173 self.0
174 .iter()
175 .filter(|(_, remote)| remote.demoted.is_some_and(|at| at.elapsed() >= grace))
176 .map(|(id, remote)| (*id, remote.peer, remote.outbound))
177 .collect()
178 }
179
180 fn has_data(&self, peer: PeerId) -> bool {
181 self.0
182 .values()
183 .any(|remote| remote.peer == peer && remote.class == Class::Data)
184 }
185
186 fn relayed_of(&self, peer: PeerId) -> Vec<ConnectionId> {
187 self.0
188 .iter()
189 .filter(|(_, remote)| remote.peer == peer && remote.class == Class::Relayed)
190 .map(|(id, _)| *id)
191 .collect()
192 }
193
194 fn candidate(remote: &Remote) -> bool {
195 remote.outbound
196 && matches!(remote.class, Class::Data | Class::Demoted)
197 && is_public(&remote.address)
198 }
199
200 fn relay_candidates(&self) -> Vec<(PeerId, Multiaddr)> {
203 let mut candidates: Vec<(&ConnectionId, &Remote)> = self
204 .0
205 .iter()
206 .filter(|(_, remote)| Self::candidate(remote))
207 .collect();
208 candidates.sort_by_key(|(id, _)| **id);
209 candidates
210 .into_iter()
211 .map(|(_, remote)| (remote.peer, without_peer(&remote.address)))
212 .collect()
213 }
214}
215
216pub(crate) fn without_peer(address: &Multiaddr) -> Multiaddr {
217 address
218 .iter()
219 .filter(|part| !matches!(part, Protocol::P2p(_)))
220 .collect()
221}
222
223pub(crate) fn direct_shape(address: &Multiaddr) -> bool {
226 let mut parts = address.iter();
227 let ip = matches!(parts.next(), Some(Protocol::Ip4(_) | Protocol::Ip6(_)));
228 let transport = match (parts.next(), parts.next()) {
229 (Some(Protocol::Tcp(port)), None) => port != 0,
230 (Some(Protocol::Udp(port)), Some(Protocol::QuicV1)) => port != 0 && parts.next().is_none(),
231 _ => false,
232 };
233 ip && transport && is_public(address)
234}
235
236pub(crate) fn validate_announce(address: &Multiaddr, local: PeerId) -> Result<Multiaddr, Error> {
240 let mut bare = address.clone();
241 if let Some(Protocol::P2p(named)) = address.iter().last() {
242 if named != local {
243 return Err(Error::Invalid("announce address names another peer".into()));
244 }
245 bare.pop();
246 }
247 if !is_public(&bare) {
248 return Err(Error::Invalid("announce address must be public".into()));
249 }
250 if !direct_shape(&bare) {
251 return Err(Error::Invalid(
252 "announce address must be an IP literal with TCP or UDP/QUIC v1".into(),
253 ));
254 }
255 Ok(bare)
256}
257
258#[derive(Clone, Copy, Debug, PartialEq, Eq)]
261pub enum ReachMode {
262 AutoNat,
263 Forced(Reachability),
264}
265
266pub(crate) struct Reach {
267 forced: ReachMode,
268 externals: Externals,
269 probe_failed: bool,
270 relay_open: Arc<AtomicBool>,
271 local: PeerId,
272 listen: Option<Multiaddr>,
273 quic: bool,
274 reservations: HashMap<ListenerId, PeerId>,
275 backoff: HashMap<PeerId, Instant>,
276 failures: HashMap<PeerId, u32>,
277 circuits: HashMap<PeerId, u32>,
278}
279
280fn retry_delay(failures: u32) -> Duration {
285 let doublings = failures.saturating_sub(1).min(16);
286 Reach::RETRY
287 .saturating_mul(1 << doublings)
288 .min(Reach::BACKOFF)
289}
290
291impl Reach {
292 const RESERVATIONS: usize = 2;
293 const RETRY: Duration = Duration::from_secs(30);
294 const BACKOFF: Duration = Duration::from_secs(10 * 60);
295 const HOLDER_GRACE: Duration = Duration::from_secs(10);
296
297 pub(crate) fn new(
298 forced: ReachMode,
299 announce: Vec<Multiaddr>,
300 relay_open: Arc<AtomicBool>,
301 (local, listen): (PeerId, Option<Multiaddr>),
302 ) -> Self {
303 Self {
304 forced,
305 externals: Externals::new(local, announce),
306 probe_failed: false,
307 relay_open,
308 local,
309 listen,
310 quic: false,
311 reservations: HashMap::new(),
312 backoff: HashMap::new(),
313 failures: HashMap::new(),
314 circuits: HashMap::new(),
315 }
316 }
317
318 fn back_off(&mut self, relay: PeerId, now: Instant) {
319 let failures = self.failures.entry(relay).or_default();
320 *failures = failures.saturating_add(1);
321 self.backoff.insert(relay, now + retry_delay(*failures));
322 }
323
324 pub(crate) fn update(&self, nat: &Nat) {
327 let observed = match (
328 self.externals.announced() || !nat.external().is_empty(),
329 self.probe_failed,
330 ) {
331 (true, _) => Reachability::Public,
332 (false, true) => Reachability::Private,
333 (false, false) => Reachability::Unknown,
334 };
335 let reachability = match self.forced {
336 ReachMode::Forced(reachability) => reachability,
337 ReachMode::AutoNat => observed,
338 };
339 nat.set_reachability(reachability);
340 self.relay_open
341 .store(reachability == Reachability::Public, Ordering::Release);
342 }
343
344 pub(crate) fn absorb(
346 &mut self,
347 swarm: &mut Swarm<Behaviour>,
348 remotes: &Remotes,
349 nat: &Nat,
350 event: SwarmEvent<BehaviourEvent>,
351 ) -> Option<SwarmEvent<BehaviourEvent>> {
352 match event {
353 SwarmEvent::Behaviour(BehaviourEvent::Autonat(event)) => {
354 self.autonat(swarm, &event, nat)
355 }
356 SwarmEvent::Behaviour(BehaviourEvent::AutonatServer(event)) => {
357 tracing::debug!(?event, "AutoNAT probe served")
358 }
359 SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => self.relay_server(&event, nat),
360 SwarmEvent::Behaviour(BehaviourEvent::RelayClient(event)) => {
361 self.relay_client(&event, nat)
362 }
363 SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => {
364 Self::dcutr(swarm, remotes, event)
365 }
366 SwarmEvent::ExternalAddrConfirmed { address } => self.external(address, true, nat),
367 SwarmEvent::ExternalAddrExpired { address } => self.external(address, false, nat),
368 SwarmEvent::NewListenAddr { ref address, .. }
369 if self.listen_addr(swarm, address, nat) => {}
370 SwarmEvent::ExpiredListenAddr { ref address, .. }
371 if self.listen_expired(address, nat) => {}
372 SwarmEvent::ListenerClosed {
373 listener_id,
374 ref addresses,
375 ref reason,
376 } => {
377 for address in addresses {
378 self.listen_expired(address, nat);
379 }
380 self.listener_closed(listener_id, reason, nat);
381 return Some(event);
382 }
383 other => return Some(other),
384 }
385 None
386 }
387
388 fn autonat(
391 &mut self,
392 swarm: &mut Swarm<Behaviour>,
393 event: &autonat::v2::client::Event,
394 nat: &Nat,
395 ) {
396 if let Err(error) = &event.result {
397 tracing::debug!(%error, address = %event.tested_addr, "AutoNAT probe failed");
398 self.probe_failed = true;
399 self.update(nat);
400 return;
401 }
402 match self.externals.verdict(&event.tested_addr, event.server) {
403 Verdict::Confirm(address) => self.advertise(swarm, address, nat),
404 Verdict::Retry(address) => swarm.behaviour_mut().candidates.offer(address),
405 Verdict::Ignore => {
406 tracing::debug!(address = %event.tested_addr, server = %event.server,
407 "AutoNAT confirmation not counted")
408 }
409 }
410 }
411
412 pub(crate) fn observed(
416 &mut self,
417 (swarm, remotes): (&mut Swarm<Behaviour>, &Remotes),
418 (observed, witness): (Multiaddr, PeerId),
419 nat: &Nat,
420 ) {
421 if !direct_shape(&observed) {
422 return;
423 }
424 let reported = (witness, Instant::now());
425 let data = |peer: &PeerId| remotes.has_data(*peer);
426 for stale in self
427 .externals
428 .witnessed(&observed, reported, &nat.external(), data)
429 {
430 tracing::info!(%stale, now = %observed, "external address mapping changed");
431 swarm.remove_external_address(&stale);
432 self.external(stale, false, nat);
433 }
434 if self.externals.flooded(witness) {
435 tracing::debug!(%witness, "witness reports too many addresses; ignored");
436 return;
437 }
438 swarm.behaviour_mut().candidates.offer(observed);
439 }
440
441 pub(crate) fn advertise(&self, swarm: &mut Swarm<Behaviour>, address: Multiaddr, nat: &Nat) {
444 swarm.add_external_address(address.clone());
445 self.external(address, true, nat);
446 }
447
448 pub(crate) fn external(&self, address: Multiaddr, confirmed: bool, nat: &Nat) {
451 if PeerLink::of(&address) == PeerLink::Direct && self.externals.admits(&address) {
452 nat.set_external(address, confirmed);
453 self.update(nat);
454 }
455 }
456
457 fn relay_server(&mut self, event: &relay::Event, nat: &Nat) {
458 match event {
459 relay::Event::ReservationReqAccepted { src_peer_id, .. } => {
460 nat.set_reserved_by(*src_peer_id, true)
461 }
462 relay::Event::ReservationClosed { src_peer_id }
463 | relay::Event::ReservationTimedOut { src_peer_id } => {
464 nat.set_reserved_by(*src_peer_id, false)
465 }
466 relay::Event::CircuitReqAccepted {
467 src_peer_id,
468 dst_peer_id,
469 } => {
470 for peer in [src_peer_id, dst_peer_id] {
471 *self.circuits.entry(*peer).or_default() += 1;
472 }
473 }
474 relay::Event::CircuitClosed {
475 src_peer_id,
476 dst_peer_id,
477 ..
478 } => {
479 for peer in [src_peer_id, dst_peer_id] {
480 self.circuit_closed(*peer);
481 }
482 }
483 _ => (),
484 }
485 }
486
487 fn circuit_closed(&mut self, peer: PeerId) {
488 let Some(count) = self.circuits.get_mut(&peer) else {
489 return;
490 };
491 *count -= 1;
492 if *count == 0 {
493 self.circuits.remove(&peer);
494 }
495 }
496
497 fn retire_holders(&self, swarm: &mut Swarm<Behaviour>, remotes: &Remotes, nat: &Nat) {
504 let relays = nat.relays();
505 let seeking = match nat.reachability() {
506 Reachability::Public => 0,
507 Reachability::Unknown => Self::RESERVATIONS,
508 Reachability::Private => Self::RESERVATIONS.saturating_sub(relays.len()),
509 };
510 let mut holding = nat.reserved_by_others();
511 holding.extend(relays);
512 let mut kept = 0;
513 for (id, peer, seeker) in remotes.retirable(Self::HOLDER_GRACE) {
514 if holding.contains(&peer) || self.circuits.contains_key(&peer) {
515 continue;
516 }
517 if seeker && kept < seeking {
518 kept += 1;
519 continue;
520 }
521 tracing::debug!(%peer, "holder carries no reservation or circuit; closing");
522 swarm.close_connection(id);
523 }
524 }
525
526 pub(crate) fn relay_client(&mut self, event: &relay::client::Event, nat: &Nat) {
528 if let relay::client::Event::ReservationReqAccepted { relay_peer_id, .. } = event {
529 nat.set_relay(*relay_peer_id, true);
530 self.failures.remove(relay_peer_id);
531 }
532 }
533
534 pub(crate) fn dcutr(swarm: &mut Swarm<Behaviour>, remotes: &Remotes, event: dcutr::Event) {
537 match event.result {
538 Ok(_direct) => {
539 for id in remotes.relayed_of(event.remote_peer_id) {
540 swarm.close_connection(id);
541 }
542 }
543 Err(error) => {
544 tracing::debug!(peer = %event.remote_peer_id, %error, "hole punch failed")
545 }
546 }
547 }
548
549 fn punches(swarm: &mut Swarm<Behaviour>, remotes: &Remotes, nat: &Nat) {
553 for (peer, outcome) in swarm.behaviour_mut().dcutr.drain(Instant::now()) {
554 match outcome {
555 Ok(()) => {
556 nat.punch_succeeded();
557 for id in remotes.relayed_of(peer) {
558 swarm.close_connection(id);
559 }
560 }
561 Err(error) => nat.punch_failed(format!("{peer}: {error}")),
562 }
563 }
564 }
565
566 pub(crate) fn listen_addr(
569 &mut self,
570 swarm: &mut Swarm<Behaviour>,
571 address: &Multiaddr,
572 nat: &Nat,
573 ) -> bool {
574 if PeerLink::of(address) == PeerLink::Relayed {
575 nat.set_circuit(self.with_local(address), true);
576 return true;
577 }
578 self.open_quic(swarm, address);
579 if let Some(candidate) = self.externals.bound(address) {
580 swarm.behaviour_mut().candidates.offer(candidate);
581 }
582 false
583 }
584
585 pub(crate) fn listen_expired(&self, address: &Multiaddr, nat: &Nat) -> bool {
586 let circuit = PeerLink::of(address) == PeerLink::Relayed;
587 if circuit {
588 nat.set_circuit(self.with_local(address), false);
589 }
590 circuit
591 }
592
593 pub(crate) fn listener_closed(
594 &mut self,
595 id: ListenerId,
596 reason: &Result<(), io::Error>,
597 nat: &Nat,
598 ) {
599 let Some(relay) = self.reservations.remove(&id) else {
600 return;
601 };
602 nat.set_relay(relay, false);
603 if let Err(error) = reason {
604 tracing::debug!(%relay, %error, "relay reservation ended; retrying after backoff");
605 self.back_off(relay, Instant::now());
606 }
607 }
608
609 fn with_local(&self, address: &Multiaddr) -> Multiaddr {
610 match address.iter().last() {
611 Some(Protocol::P2p(_)) => address.clone(),
612 _ => address.clone().with(Protocol::P2p(self.local)),
613 }
614 }
615
616 fn open_quic(&mut self, swarm: &mut Swarm<Behaviour>, bound: &Multiaddr) {
617 let Some(listen) = &self.listen else {
618 return;
619 };
620 let port = bound.iter().find_map(|part| match part {
621 Protocol::Tcp(port) => Some(port),
622 _ => None,
623 });
624 let (Some(port), Some(ip), false) = (port, listen.iter().next(), self.quic) else {
625 return;
626 };
627 self.quic = true;
628 let quic = Multiaddr::empty()
629 .with(ip)
630 .with(Protocol::Udp(port))
631 .with(Protocol::QuicV1);
632 if let Err(error) = swarm.listen_on(quic.clone()) {
633 tracing::warn!(%error, %quic, "QUIC listener unavailable; TCP only");
634 }
635 }
636
637 pub(crate) fn maintain(&mut self, swarm: &mut Swarm<Behaviour>, remotes: &Remotes, nat: &Nat) {
640 Self::punches(swarm, remotes, nat);
641 self.retire_holders(swarm, remotes, nat);
642 if nat.reachability() != Reachability::Private {
643 for (id, _) in self.reservations.drain() {
644 swarm.remove_listener(id);
645 }
646 return;
647 }
648 let now = Instant::now();
649 self.backoff.retain(|_, until| *until > now);
650 for (relay, address) in remotes.relay_candidates() {
651 if self.reservations.len() >= Self::RESERVATIONS {
652 return;
653 }
654 let held = self.reservations.values().any(|known| *known == relay);
655 if held || relay == self.local || self.backoff.contains_key(&relay) {
656 continue;
657 }
658 let circuit = address
659 .with(Protocol::P2p(relay))
660 .with(Protocol::P2pCircuit);
661 match swarm.listen_on(circuit) {
662 Ok(id) => {
663 self.reservations.insert(id, relay);
664 }
665 Err(error) => {
666 tracing::warn!(%error, %relay, "relay reservation not attempted");
667 self.back_off(relay, now);
668 }
669 }
670 }
671 }
672}
673
674#[cfg(test)]
675mod tests {
676 use super::{Remotes, retry_delay};
677 use crate::peer_limits::{Class, Classes};
678 use libp2p::{
679 Multiaddr, PeerId,
680 core::{ConnectedPoint, Endpoint, transport::PortUse},
681 swarm::ConnectionId,
682 };
683
684 #[test]
685 fn a_failed_reservation_retries_after_thirty_seconds_doubling_to_ten_minutes() {
686 let delays = [1, 2, 3, 4, 5, 6, 40].map(|failures| retry_delay(failures).as_secs());
687 assert_eq!(delays, [30, 60, 120, 240, 480, 600, 600]);
688 }
689
690 fn admit(
691 (remotes, classes): (&mut Remotes, &Classes),
692 (id, peer): (usize, PeerId),
693 (class, outbound): (Class, bool),
694 ) -> ConnectionId {
695 let id = ConnectionId::new_unchecked(id);
696 classes.set(id, class);
697 let address: Multiaddr = "/ip4/11.0.0.1/udp/1/quic-v1".parse().unwrap();
698 let endpoint = match outbound {
699 true => ConnectedPoint::Dialer {
700 address,
701 role_override: Endpoint::Dialer,
702 port_use: PortUse::Reuse,
703 },
704 false => ConnectedPoint::Listener {
705 local_addr: address.clone(),
706 send_back_addr: address,
707 },
708 };
709 remotes.admit((id, peer), &endpoint, classes);
710 id
711 }
712
713 #[test]
714 fn a_circuit_the_peer_dialed_retires_our_stale_direct_links_not_itself() {
715 let (mut remotes, classes) = (Remotes::default(), Classes::default());
716 let (peer, bystander) = (PeerId::random(), PeerId::random());
717 let mut set = |entry, kind| admit((&mut remotes, &classes), entry, kind);
718 let stale = set((1, peer), (Class::Data, true));
719 let holder = set((2, peer), (Class::Holder, false));
720 set((3, bystander), (Class::Data, true));
721 let circuit = set((4, peer), (Class::Relayed, false));
722 let mut retired = remotes.superseded(circuit, peer, Class::Relayed);
723 retired.sort();
724 assert_eq!(
725 retired,
726 vec![stale, holder],
727 "the peer dialed: ours are stale"
728 );
729 remotes.remove(stale);
730 remotes.remove(holder);
731 let punched = admit((&mut remotes, &classes), (5, peer), (Class::Data, true));
732 assert_eq!(
733 remotes.superseded(punched, peer, Class::Data),
734 vec![circuit]
735 );
736 let ours = admit((&mut remotes, &classes), (6, peer), (Class::Relayed, true));
737 assert_eq!(
738 remotes.superseded(ours, peer, Class::Relayed),
739 vec![ours],
740 "a circuit we dialed beside a data link is redundant"
741 );
742 }
743}