syncweb_core/net/relay/
transport.rs

1//! The iroh [`CustomTransport`] that carries QUIC datagrams over bep-relay
2//! sessions.
3//!
4//! Each syncweb node advertises a [`CustomAddr`] whose payload carries both
5//! its iroh node id (so the receiving side can attribute traffic to a node)
6//! and its relay device id (so the receiving side can ask a Syncthing relay
7//! to broker a session). Datagrams handed over by iroh are framed into an
8//! existing session towards the peer, or (on first use) a session is dialed
9//! in the background while QUIC retransmits.
10
11use std::io::{self, IoSliceMut};
12use std::sync::Arc;
13use std::task::{Context, Poll};
14use std::time::Duration;
15
16use bytes::Bytes;
17use iroh::PublicKey;
18use iroh::endpoint::transports::{CustomEndpoint, CustomSender, CustomTransport, RecvInfo, Transmit};
19use iroh_base::CustomAddr;
20use noq_udp::RecvMeta;
21use parking_lot::RwLock;
22use tokio::sync::mpsc as tokio_mpsc;
23use tokio_rustls::TlsConnector;
24use tokio_util::sync::CancellationToken;
25use tracing::warn;
26
27use super::certs::RelayCerts;
28use super::manager::start_peer_dial;
29use super::protocol::Message;
30
31/// Transport id of the syncweb bep-relay transport ("SWSB").
32pub const BEP_TRANSPORT_ID: u64 = 0x5357_5342;
33/// Length of a relay device id (a SHA-256 digest).
34pub const RELAY_ID_LEN: usize = 32;
35/// Version marker for [`ADDR_LEN`] payloads carried in our custom addresses.
36const ADDR_VERSION: u8 = 1;
37/// Byte length of a custom address payload: version + node + relay id.
38pub const ADDR_LEN: usize = 1 + 32 + RELAY_ID_LEN;
39/// Length of the node-id prefix inside the payload.
40const NODE_LEN: usize = 32;
41/// Offset of the relay device id inside the payload.
42const RELAY_START: usize = 1 + NODE_LEN;
43/// Byte length of a relay device id inside the payload after the prefix.
44const RELAY_LEN: usize = 32;
45
46/// An incoming datagram delivered from a relay session to iroh.
47#[derive(Debug)]
48pub struct IncomingPacket {
49    /// Custom address of the peer that sent the datagram.
50    pub remote: CustomAddr,
51    /// Our own custom address that received it.
52    pub local: CustomAddr,
53    /// Complete datagram bytes.
54    pub data: Bytes,
55}
56
57/// A live relay session towards one peer.
58///
59/// The bounded queue lets the synchronous send path enqueue datagrams
60/// without blocking; overflow simply drops the frame and QUIC retransmits.
61#[derive(Debug)]
62pub struct Session {
63    /// Relay device id of the peer.
64    pub peer: [u8; 32],
65    /// Node id used to label incoming datagrams from this peer.
66    pub peer_node: [u8; 32],
67    /// Queue consumed by the session writer task.
68    pub queue: tokio_mpsc::Sender<Bytes>,
69    /// Cooperative cancellation for the session tasks.
70    pub cancel: CancellationToken,
71}
72
73/// State shared between the manager (session establishment) and the
74/// transport (socket I/O).
75pub struct Backend {
76    /// Sessions towards peers, keyed by the peer relay device id.
77    pub sessions: RwLock<std::collections::HashMap<[u8; 32], Arc<Session>>>,
78    /// Relay device id to iroh node id, for labeling inbound traffic.
79    pub node_ids: RwLock<std::collections::HashMap<[u8; 32], [u8; 32]>>,
80    /// The ingress sink into iroh, created when the transport is bound.
81    pub ingress: parking_lot::Mutex<Option<tokio_mpsc::UnboundedSender<IncomingPacket>>>,
82    /// Our published custom address.
83    pub local_addrs: n0_watcher::Watchable<Vec<CustomAddr>>,
84    /// Dials currently in flight, keyed by the peer relay device id.
85    pub dials: parking_lot::Mutex<std::collections::HashSet<[u8; 32]>>,
86    /// Round-robin index over [`Backend::relays`].
87    pub relay_index: parking_lot::Mutex<usize>,
88    /// The relays we are registered with, in dial order.
89    pub relays: RwLock<Vec<String>>,
90    /// Outbound control-message senders for the live registration connection
91    /// of each relay, keyed by relay url. `ConnectRequest`s are sent here so
92    /// the relay attributes them to our registered device id.
93    pub control: RwLock<std::collections::HashMap<String, tokio_mpsc::UnboundedSender<Message>>>,
94    /// Persistent identity used for control connections.
95    pub certs: Arc<RelayCerts>,
96    /// TLS connector for control connections.
97    pub connector: TlsConnector,
98    /// Timeout budget for connection and frame operations.
99    pub timeout: Duration,
100    /// Cooperative cancellation for all relay tasks.
101    pub cancel: CancellationToken,
102}
103
104impl std::fmt::Debug for Backend {
105    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        f.debug_struct("Backend").finish_non_exhaustive()
107    }
108}
109
110impl Backend {
111    /// Create an empty backend with the given identity and connector.
112    #[must_use]
113    pub(crate) fn new(certs: Arc<RelayCerts>, connector: TlsConnector, timeout: Duration) -> Self {
114        Self {
115            sessions: RwLock::new(std::collections::HashMap::new()),
116            node_ids: RwLock::new(std::collections::HashMap::new()),
117            ingress: parking_lot::Mutex::new(None),
118            local_addrs: n0_watcher::Watchable::new(Vec::new()),
119            dials: parking_lot::Mutex::new(std::collections::HashSet::new()),
120            relay_index: parking_lot::Mutex::new(0),
121            relays: RwLock::new(Vec::new()),
122            control: RwLock::new(std::collections::HashMap::new()),
123            certs,
124            connector,
125            timeout,
126            cancel: CancellationToken::new(),
127        }
128    }
129
130    /// Build the custom address this node publishes.
131    #[must_use]
132    pub(crate) fn local_addr(&self) -> CustomAddr {
133        self.local_addrs
134            .get()
135            .first()
136            .cloned()
137            .unwrap_or_else(|| CustomAddr::from_parts(BEP_TRANSPORT_ID, &[0_u8; ADDR_LEN]))
138    }
139
140    /// Update the address we publish, notifying the bind watcher.
141    pub(crate) fn update_local_addr(&self, addr: CustomAddr) {
142        let _ = self.local_addrs.set(vec![addr]);
143    }
144
145    /// Record a relay id to node id mapping for labeling inbound datagrams.
146    pub(crate) fn remember_node(&self, relay_id: [u8; 32], node: [u8; 32]) {
147        self.node_ids.write().insert(relay_id, node);
148    }
149
150    /// The next relay url in round-robin dial order.
151    #[must_use]
152    pub(crate) fn next_relay(&self) -> Option<String> {
153        if self.relays.read().is_empty() {
154            return None;
155        }
156        let count = self.relays.read().len();
157        let slot = {
158            let mut index = self.relay_index.lock();
159            let slot = *index;
160            *index = if slot.saturating_add(1) >= count {
161                0
162            } else {
163                slot.saturating_add(1)
164            };
165            slot
166        };
167        self.relays.read().get(slot).cloned()
168    }
169}
170
171/// A custom transport that talks bep-relay to Syncthing relays.
172#[derive(Clone, Debug)]
173pub struct SyncthingRelayTransport(Arc<Backend>);
174
175impl SyncthingRelayTransport {
176    /// Wrap a manager backend.
177    #[must_use]
178    pub(crate) const fn new(backend: Arc<Backend>) -> Self {
179        Self(backend)
180    }
181
182    /// The custom address this transport advertises locally.
183    #[must_use]
184    pub fn local_addr(&self) -> CustomAddr {
185        self.0.local_addr()
186    }
187
188    /// Record the iroh node id that owns the given relay device id, so
189    /// inbound datagrams from that peer can be attributed correctly.
190    pub fn remember_node(&self, relay_id: &[u8; 32], node: PublicKey) {
191        self.0.remember_node(*relay_id, *node.as_bytes());
192    }
193}
194
195impl CustomTransport for SyncthingRelayTransport {
196    fn bind(&self) -> io::Result<Box<dyn CustomEndpoint>> {
197        let (tx, rx) = tokio_mpsc::unbounded_channel();
198        *self.0.ingress.lock() = Some(tx);
199        Ok(Box::new(RelayEndpoint {
200            backend: self.0.clone(),
201            recv: rx,
202        }))
203    }
204}
205
206/// The bound endpoint for the bep-relay transport.
207#[derive(Debug)]
208struct RelayEndpoint {
209    backend: Arc<Backend>,
210    recv: tokio_mpsc::UnboundedReceiver<IncomingPacket>,
211}
212
213impl CustomEndpoint for RelayEndpoint {
214    fn watch_local_addrs(&self) -> n0_watcher::Direct<Vec<CustomAddr>> {
215        self.backend.local_addrs.watch()
216    }
217
218    fn create_sender(&self) -> Arc<dyn CustomSender> {
219        Arc::new(RelaySender(self.backend.clone()))
220    }
221
222    fn poll_recv(
223        &mut self,
224        cx: &mut Context<'_>,
225        bufs: &mut [IoSliceMut<'_>],
226        metas: &mut [RecvMeta],
227        recv_infos: &mut [RecvInfo],
228    ) -> Poll<io::Result<usize>> {
229        let mut count = 0;
230        while count < bufs.len() {
231            match self.recv.poll_recv(cx) {
232                Poll::Pending | Poll::Ready(None) => break,
233                Poll::Ready(Some(packet)) => {
234                    let len = packet.data.len();
235                    let Some(buffer) = bufs.get_mut(count) else {
236                        break;
237                    };
238                    let Some(dest) = buffer.get_mut(..len) else {
239                        warn!(
240                            bytes = len,
241                            capacity = buffer.len(),
242                            "datagram larger than consumer buffer, dropping"
243                        );
244                        continue;
245                    };
246                    dest.copy_from_slice(&packet.data);
247                    let Some(meta) = metas.get_mut(count) else {
248                        break;
249                    };
250                    *meta = RecvMeta::default();
251                    meta.len = len;
252                    meta.stride = len;
253                    let Some(recv_info) = recv_infos.get_mut(count) else {
254                        break;
255                    };
256                    *recv_info = RecvInfo::new(packet.remote, Some(packet.local));
257                    count = count.saturating_add(1);
258                }
259            }
260        }
261        if count == 0 {
262            return Poll::Pending;
263        }
264        Poll::Ready(Ok(count))
265    }
266}
267
268/// The sender half of the bep-relay transport.
269#[derive(Debug)]
270struct RelaySender(Arc<Backend>);
271
272impl CustomSender for RelaySender {
273    fn is_valid_send_addr(&self, addr: &CustomAddr) -> bool {
274        decode_peer_addr(addr).is_some()
275    }
276
277    fn poll_send(
278        &self,
279        _cx: &mut Context<'_>,
280        dst: &CustomAddr,
281        _src: Option<&CustomAddr>,
282        transmit: &Transmit<'_>,
283    ) -> Poll<io::Result<()>> {
284        let Some((peer, node)) = decode_peer_addr(dst) else {
285            return Poll::Ready(Err(io::Error::new(
286                io::ErrorKind::InvalidInput,
287                "custom address is not a syncweb relay address",
288            )));
289        };
290        let candidate = self.0.sessions.read().get(&peer).cloned();
291        if let Some(session) = candidate {
292            if session
293                .queue
294                .try_send(Bytes::copy_from_slice(transmit.contents))
295                .is_err()
296            {
297                start_peer_dial(self.0.clone(), peer, node);
298            }
299            return Poll::Ready(Ok(()));
300        }
301        start_peer_dial(self.0.clone(), peer, node);
302        Poll::Ready(Ok(()))
303    }
304}
305
306/// Encode a node id and relay device id into our custom address payload.
307#[must_use]
308pub fn encode_peer_addr(node: &PublicKey, relay_id: &[u8; 32]) -> Vec<u8> {
309    let mut payload = Vec::with_capacity(ADDR_LEN);
310    payload.push(ADDR_VERSION);
311    payload.extend_from_slice(node.as_bytes());
312    payload.extend_from_slice(relay_id);
313    payload
314}
315
316/// Build the custom address this node advertises.
317#[must_use]
318pub fn own_addr(node: &PublicKey, relay_id: &[u8; 32]) -> CustomAddr {
319    CustomAddr::from_parts(BEP_TRANSPORT_ID, &encode_peer_addr(node, relay_id))
320}
321
322/// Parse one of our custom addresses into its peer (relay id, node id).
323#[must_use]
324pub fn decode_peer_addr(addr: &CustomAddr) -> Option<([u8; 32], [u8; 32])> {
325    if addr.id() != BEP_TRANSPORT_ID {
326        return None;
327    }
328    let data = addr.data();
329    if data.len() != ADDR_LEN || data.first() != Some(&ADDR_VERSION) {
330        return None;
331    }
332    let mut node = [0_u8; NODE_LEN];
333    node.copy_from_slice(data.get(1..=NODE_LEN)?);
334    let mut relay = [0_u8; RELAY_LEN];
335    relay.copy_from_slice(data.get(RELAY_START..)?);
336    Some((relay, node))
337}
338
339/// Build the remote custom address to label datagrams on a session to a peer
340/// whose relay id is `relay` and node id `node`.
341#[must_use]
342pub fn peer_addr(relay: [u8; 32], node: [u8; 32]) -> CustomAddr {
343    let mut payload = Vec::with_capacity(ADDR_LEN);
344    payload.push(ADDR_VERSION);
345    payload.extend_from_slice(&node);
346    payload.extend_from_slice(&relay);
347    CustomAddr::from_parts(BEP_TRANSPORT_ID, &payload)
348}
349
350#[cfg(test)]
351mod tests {
352    use super::*;
353
354    #[test]
355    fn peer_addr_round_trips() {
356        let node_id = iroh::SecretKey::generate().public();
357        let node = *node_id.as_bytes();
358        let relay = [9_u8; 32];
359        let payload = encode_peer_addr(&node_id, &relay);
360        let addr = CustomAddr::from((BEP_TRANSPORT_ID, payload.as_ref()));
361        let (decoded_relay, decoded_node) = decode_peer_addr(&addr).expect("decode");
362        assert_eq!(decoded_relay, relay);
363        assert_eq!(decoded_node, node);
364    }
365
366    #[test]
367    fn foreign_addrs_are_rejected() {
368        let addr = CustomAddr::from((42, b"junk".as_slice()));
369        assert!(decode_peer_addr(&addr).is_none());
370    }
371
372    #[test]
373    fn short_payloads_are_rejected() {
374        let addr = CustomAddr::from((BEP_TRANSPORT_ID, b"1".as_slice()));
375        assert!(decode_peer_addr(&addr).is_none());
376    }
377}
syncweb_core/net/relay/transport.rs

syncweb_core/net/relay/
transport.rs
