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_punch::Punches,
18    peer_rewaken::Rewaken,
19};
20use libp2p::{
21    Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder, autonat, identify, identity::Keypair,
22    multiaddr::Protocol, noise, quic, relay, request_response, 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: Punches,
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 1 admits two
80/// reservations per peer (a restarted peer's new one beside the stale one we
81/// hold until its old connection times out) and 3 admits four circuits.
82fn relay_config(open: &Arc<AtomicBool>, local: PeerId) -> relay::Config {
83    let gate = || RelayGate {
84        open: Arc::clone(open),
85        local,
86    };
87    let mut config = relay::Config {
88        max_reservations: 128,
89        max_reservations_per_peer: 1,
90        max_circuits: 16,
91        max_circuits_per_peer: 3,
92        max_circuit_duration: Duration::from_secs(2 * 60),
93        max_circuit_bytes: 1 << 20,
94        ..relay::Config::default()
95    };
96    config.reservation_rate_limiters.push(Box::new(gate()));
97    config.circuit_src_rate_limiters.push(Box::new(gate()));
98    config
99}
100
101/// How the overlay resolves names. `Refuse` stands in for an unreadable
102/// system configuration (tests inject it); it is never a fallback resolver.
103#[derive(Clone, Debug, PartialEq, Eq)]
104pub enum DnsMode {
105    System,
106    Refuse(String),
107}
108
109/// Whether this swarm can resolve names, and why not.
110#[derive(Clone, Debug, PartialEq, Eq)]
111pub enum Dns {
112    Available,
113    Unavailable(String),
114}
115
116impl Dns {
117    /// Names need the resolver; IP literals never do.
118    pub fn admits(&self, address: &Multiaddr) -> Result<(), Error> {
119        let named = address.iter().any(|part| {
120            matches!(
121                part,
122                Protocol::Dns(_) | Protocol::Dns4(_) | Protocol::Dns6(_) | Protocol::Dnsaddr(_)
123            )
124        });
125        match (self, named) {
126            (Self::Unavailable(cause), true) => Err(Error::Transport(format!(
127                "system DNS resolver unavailable: {cause}"
128            ))),
129            (Self::Unavailable(_) | Self::Available, false) | (Self::Available, true) => Ok(()),
130        }
131    }
132}
133
134fn behaviour(
135    identity: &Keypair,
136    relay_client: relay::client::Behaviour,
137    wiring: &Wiring,
138) -> Behaviour {
139    let local = identity.public().to_peer_id();
140    let records = request_response::Behaviour::with_codec(
141        Codec,
142        [(
143            StreamProtocol::new(PROTOCOL),
144            request_response::ProtocolSupport::Full,
145        )],
146        request_response::Config::default()
147            .with_request_timeout(REQUEST_TIMEOUT)
148            .with_max_concurrent_streams(2),
149    );
150    let identify = identify::Behaviour::new(
151        identify::Config::new(Behaviour::IDENTIFY_PROTOCOL.into(), identity.public())
152            .with_agent_version(format!("urma-node/{}", env!("CARGO_PKG_VERSION")))
153            .with_push_listen_addr_updates(true),
154    );
155    Behaviour {
156        limits: Limits::new(wiring.classes.clone(), Arc::clone(&wiring.relay_open)),
157        records: Fenced::data_only(records, wiring.classes.clone()),
158        identify: Fenced::without_candidates(identify),
159        candidates: Candidates::default(),
160        autonat: Fenced::without_confirmations(autonat::v2::client::Behaviour::default()),
161        autonat_server: Fenced::public_dial_backs(autonat::v2::server::Behaviour::default()),
162        relay: Rewaken(relay::Behaviour::new(
163            local,
164            relay_config(&wiring.relay_open, local),
165        )),
166        relay_client,
167        dcutr: Punches::new(local),
168    }
169}
170
171fn multiplexer() -> yamux::Config {
172    let mut config = yamux::Config::default();
173    // These setters select libp2p-yamux's bounded 0.12 backend.
174    config.set_max_num_streams(8);
175    config.set_max_buffer_size(256 * 1024);
176    config.set_receive_window_size(Behaviour::WINDOW);
177    config
178}
179
180/// QUIC with the same stream and window bounds as yamux.
181fn quic(mut config: quic::Config) -> quic::Config {
182    config.max_concurrent_stream_limit = Behaviour::STREAMS;
183    config.max_stream_data = Behaviour::WINDOW;
184    config.max_connection_data = Behaviour::WINDOW * Behaviour::STREAMS;
185    config.mtu_upper_bound(Behaviour::QUIC_MTU)
186}
187
188fn configure(config: libp2p::swarm::Config) -> libp2p::swarm::Config {
189    config
190        .with_notify_handler_buffer_size(NonZeroUsize::new(2).unwrap())
191        .with_per_connection_event_buffer_size(2)
192        .with_max_negotiating_inbound_streams(2)
193        .with_dial_concurrency_factor(NonZeroU8::MIN)
194        .with_idle_connection_timeout(Duration::from_secs(20))
195}
196
197fn transport_error(error: impl std::fmt::Display) -> Error {
198    Error::Invalid(format!("peer transport: {error}"))
199}
200
201fn behaviour_error(error: impl std::fmt::Display) -> Error {
202    Error::Invalid(format!("peer behaviour: {error}"))
203}
204
205/// Tries the system resolver first; if its configuration cannot be read the
206/// same swarm is built without DNS and the cause is returned.
207pub(crate) fn swarm(
208    identity: Keypair,
209    mode: &DnsMode,
210    wiring: &Wiring,
211) -> Result<(Swarm<Behaviour>, Dns), Error> {
212    let transports = || {
213        SwarmBuilder::with_existing_identity(identity.clone())
214            .with_tokio()
215            .with_tcp(
216                tcp::Config::default().nodelay(true),
217                noise::Config::new,
218                multiplexer,
219            )
220            .map(|builder| builder.with_quic_config(quic))
221            .map_err(transport_error)
222    };
223    let cause = match mode {
224        DnsMode::System => match transports()?.with_dns() {
225            Ok(resolving) => {
226                let swarm = resolving
227                    .with_relay_client(noise::Config::new, multiplexer)
228                    .map_err(transport_error)?
229                    .with_behaviour(|key, relay| behaviour(key, relay, wiring))
230                    .map_err(behaviour_error)?
231                    .with_swarm_config(configure)
232                    .with_connection_timeout(Duration::from_secs(5))
233                    .build();
234                return Ok((swarm, Dns::Available));
235            }
236            Err(error) => error.to_string(),
237        },
238        DnsMode::Refuse(cause) => cause.clone(),
239    };
240    tracing::warn!(%cause, "system DNS resolver unavailable; overlay runs for IP-literal peers only");
241    let swarm = transports()?
242        .with_relay_client(noise::Config::new, multiplexer)
243        .map_err(transport_error)?
244        .with_behaviour(|key, relay| behaviour(key, relay, wiring))
245        .map_err(behaviour_error)?
246        .with_swarm_config(configure)
247        .with_connection_timeout(Duration::from_secs(5))
248        .build();
249    Ok((swarm, Dns::Unavailable(cause)))
250}
urma_node/peer_transport.rs

urma_node/
peer_transport.rs
