urma_node/
peer_transport.rs

1//! The URMA overlay swarm: TCP and QUIC on the same port, Noise, bounded
2//! yamux, the relay client transport and, when the system resolver
3//! configuration can be read, DNS. Without it the overlay still runs for
4//! IP-literal peers; named addresses (`/dns*`, built-in `/dnsaddr` seeds) are
5//! then skipped and the reason is reported, never a public resolver.
6//!
7//! NAT traversal: AutoNAT v2 client and server, a circuit relay v2 server
8//! that admits reservations and circuits only while this node is public, the
9//! relay client, and DCUtR hole punching.
10use crate::{
11    Error,
12    peer::REQUEST_TIMEOUT,
13    peer_codec::Codec,
14    peer_fence::{Candidates, Fenced},
15    peer_limits::{Classes, Limits},
16    peer_protocol::PROTOCOL,
17    peer_rewaken::Rewaken,
18};
19use libp2p::{
20    Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder, autonat, dcutr, identify,
21    identity::Keypair, multiaddr::Protocol, noise, quic, relay, request_response,
22    swarm::NetworkBehaviour, tcp, yamux,
23};
24use std::{
25    num::{NonZeroU8, NonZeroUsize},
26    sync::{
27        Arc,
28        atomic::{AtomicBool, Ordering},
29    },
30    time::{Duration, Instant},
31};
32
33#[derive(NetworkBehaviour)]
34pub(crate) struct Behaviour {
35    // Admission must precede every other behaviour's connection side effects:
36    // it classifies the connection the fenced behaviours read.
37    pub(crate) limits: Limits,
38    pub(crate) records: Fenced<request_response::Behaviour<Codec>>,
39    pub(crate) identify: Fenced<identify::Behaviour>,
40    pub(crate) candidates: Candidates,
41    pub(crate) autonat: Fenced<autonat::v2::client::Behaviour>,
42    pub(crate) autonat_server: Fenced<autonat::v2::server::Behaviour>,
43    pub(crate) relay: Rewaken<relay::Behaviour>,
44    pub(crate) relay_client: relay::client::Behaviour,
45    pub(crate) dcutr: dcutr::Behaviour,
46}
47
48impl Behaviour {
49    const IDENTIFY_PROTOCOL: &str = "/urma/node/4";
50    const STREAMS: u32 = 8;
51    const WINDOW: u32 = 256 * 1024;
52    // Fly's UDP path carries about 1300 bytes; quinn's default probe ceiling
53    // (1452) loses packets there.
54    const QUIC_MTU: u16 = 1280;
55}
56
57/// What the network loop and the behaviours share: connection classes and
58/// whether this node's relay server admits reservations right now.
59#[derive(Clone)]
60pub(crate) struct Wiring {
61    pub(crate) classes: Classes,
62    pub(crate) relay_open: Arc<AtomicBool>,
63}
64
65/// Admits reservations and circuits only while the relay is open (this node
66/// is public), and never for this node itself.
67struct RelayGate {
68    open: Arc<AtomicBool>,
69    local: PeerId,
70}
71
72impl relay::RateLimiter for RelayGate {
73    fn try_next(&mut self, peer: PeerId, _: &Multiaddr, _: Instant) -> bool {
74        self.open.load(Ordering::Acquire) && peer != self.local
75    }
76}
77
78/// §3.3 limits. libp2p denies a new reservation or circuit only when the
79/// peer already holds MORE than the per-peer maximum, so 0 admits one
80/// reservation per peer and 3 admits four circuits.
81fn relay_config(open: &Arc<AtomicBool>, local: PeerId) -> relay::Config {
82    let gate = || RelayGate {
83        open: Arc::clone(open),
84        local,
85    };
86    let mut config = relay::Config {
87        max_reservations: 128,
88        max_reservations_per_peer: 0,
89        max_circuits: 16,
90        max_circuits_per_peer: 3,
91        max_circuit_duration: Duration::from_secs(2 * 60),
92        max_circuit_bytes: 1 << 20,
93        ..relay::Config::default()
94    };
95    config.reservation_rate_limiters.push(Box::new(gate()));
96    config.circuit_src_rate_limiters.push(Box::new(gate()));
97    config
98}
99
100/// How the overlay resolves names. `Refuse` stands in for an unreadable
101/// system configuration (tests inject it); it is never a fallback resolver.
102#[derive(Clone, Debug, PartialEq, Eq)]
103pub enum DnsMode {
104    System,
105    Refuse(String),
106}
107
108/// Whether this swarm can resolve names, and why not.
109#[derive(Clone, Debug, PartialEq, Eq)]
110pub enum Dns {
111    Available,
112    Unavailable(String),
113}
114
115impl Dns {
116    /// Names need the resolver; IP literals never do.
117    pub fn admits(&self, address: &Multiaddr) -> Result<(), Error> {
118        let named = address.iter().any(|part| {
119            matches!(
120                part,
121                Protocol::Dns(_) | Protocol::Dns4(_) | Protocol::Dns6(_) | Protocol::Dnsaddr(_)
122            )
123        });
124        match (self, named) {
125            (Self::Unavailable(cause), true) => Err(Error::Transport(format!(
126                "system DNS resolver unavailable: {cause}"
127            ))),
128            (Self::Unavailable(_) | Self::Available, false) | (Self::Available, true) => Ok(()),
129        }
130    }
131}
132
133fn behaviour(
134    identity: &Keypair,
135    relay_client: relay::client::Behaviour,
136    wiring: &Wiring,
137) -> Behaviour {
138    let local = identity.public().to_peer_id();
139    let records = request_response::Behaviour::with_codec(
140        Codec,
141        [(
142            StreamProtocol::new(PROTOCOL),
143            request_response::ProtocolSupport::Full,
144        )],
145        request_response::Config::default()
146            .with_request_timeout(REQUEST_TIMEOUT)
147            .with_max_concurrent_streams(2),
148    );
149    let identify = identify::Behaviour::new(
150        identify::Config::new(Behaviour::IDENTIFY_PROTOCOL.into(), identity.public())
151            .with_agent_version(format!("urma-node/{}", env!("CARGO_PKG_VERSION")))
152            .with_push_listen_addr_updates(true),
153    );
154    Behaviour {
155        limits: Limits::new(wiring.classes.clone(), Arc::clone(&wiring.relay_open)),
156        records: Fenced::data_only(records, wiring.classes.clone()),
157        identify: Fenced::without_candidates(identify),
158        candidates: Candidates::default(),
159        autonat: Fenced::without_confirmations(autonat::v2::client::Behaviour::default()),
160        autonat_server: Fenced::public_dial_backs(autonat::v2::server::Behaviour::default()),
161        relay: Rewaken(relay::Behaviour::new(
162            local,
163            relay_config(&wiring.relay_open, local),
164        )),
165        relay_client,
166        dcutr: dcutr::Behaviour::new(local),
167    }
168}
169
170fn multiplexer() -> yamux::Config {
171    let mut config = yamux::Config::default();
172    // These setters select libp2p-yamux's bounded 0.12 backend.
173    config.set_max_num_streams(8);
174    config.set_max_buffer_size(256 * 1024);
175    config.set_receive_window_size(Behaviour::WINDOW);
176    config
177}
178
179/// QUIC with the same stream and window bounds as yamux.
180fn quic(mut config: quic::Config) -> quic::Config {
181    config.max_concurrent_stream_limit = Behaviour::STREAMS;
182    config.max_stream_data = Behaviour::WINDOW;
183    config.max_connection_data = Behaviour::WINDOW * Behaviour::STREAMS;
184    config.mtu_upper_bound(Behaviour::QUIC_MTU)
185}
186
187fn configure(config: libp2p::swarm::Config) -> libp2p::swarm::Config {
188    config
189        .with_notify_handler_buffer_size(NonZeroUsize::new(2).unwrap())
190        .with_per_connection_event_buffer_size(2)
191        .with_max_negotiating_inbound_streams(2)
192        .with_dial_concurrency_factor(NonZeroU8::MIN)
193        .with_idle_connection_timeout(Duration::from_secs(20))
194}
195
196fn transport_error(error: impl std::fmt::Display) -> Error {
197    Error::Invalid(format!("peer transport: {error}"))
198}
199
200fn behaviour_error(error: impl std::fmt::Display) -> Error {
201    Error::Invalid(format!("peer behaviour: {error}"))
202}
203
204/// Tries the system resolver first; if its configuration cannot be read the
205/// same swarm is built without DNS and the cause is returned.
206pub(crate) fn swarm(
207    identity: Keypair,
208    mode: &DnsMode,
209    wiring: &Wiring,
210) -> Result<(Swarm<Behaviour>, Dns), Error> {
211    let transports = || {
212        SwarmBuilder::with_existing_identity(identity.clone())
213            .with_tokio()
214            .with_tcp(
215                tcp::Config::default().nodelay(true),
216                noise::Config::new,
217                multiplexer,
218            )
219            .map(|builder| builder.with_quic_config(quic))
220            .map_err(transport_error)
221    };
222    let cause = match mode {
223        DnsMode::System => match transports()?.with_dns() {
224            Ok(resolving) => {
225                let swarm = resolving
226                    .with_relay_client(noise::Config::new, multiplexer)
227                    .map_err(transport_error)?
228                    .with_behaviour(|key, relay| behaviour(key, relay, wiring))
229                    .map_err(behaviour_error)?
230                    .with_swarm_config(configure)
231                    .with_connection_timeout(Duration::from_secs(5))
232                    .build();
233                return Ok((swarm, Dns::Available));
234            }
235            Err(error) => error.to_string(),
236        },
237        DnsMode::Refuse(cause) => cause.clone(),
238    };
239    tracing::warn!(%cause, "system DNS resolver unavailable; overlay runs for IP-literal peers only");
240    let swarm = transports()?
241        .with_relay_client(noise::Config::new, multiplexer)
242        .map_err(transport_error)?
243        .with_behaviour(|key, relay| behaviour(key, relay, wiring))
244        .map_err(behaviour_error)?
245        .with_swarm_config(configure)
246        .with_connection_timeout(Duration::from_secs(5))
247        .build();
248    Ok((swarm, Dns::Unavailable(cause)))
249}
urma_node/peer_transport.rs

urma_node/
peer_transport.rs
