syncweb_core/node/
beacon_lookup.rs

1//! A UDP beacon address lookup, inspired by Zyre's passive discovery beacon.
2//!
3//! The beacon broadcasts a node's current [`EndpointData`] (direct addresses and
4//! an optional relay URL) as a small JSON datagram on a fixed UDP port. Unlike
5//! mDNS there is no service-name registry; peers on the same LAN simply listen
6//! on the same port and accept datagrams that carry the expected scope token.
7//!
8//! Each configured network derives a 16-byte scope from its name. The scope is
9//! used twice:
10//! * to derive the beacon port (the scope offsets the base port), and
11//! * as a token inside every datagram so that different networks never accept
12//!   each other's beacons even when the derived ports collide.
13//!
14//! This is an [`AddressLookup`] and can be registered next to the mDNS lookup.
15//! Like mDNS it is a fallback for local discovery and is *not* an access
16//! boundary: any peer that knows the derived port and scope can still be
17//! reached, and membership is still enforced by the layer 2/3 peer allowlist.
18
19use std::{
20    collections::{BTreeSet, HashMap},
21    net::{IpAddr, Ipv4Addr, SocketAddr},
22    pin::Pin,
23    ptr,
24    str::FromStr,
25    sync::Arc,
26    task::{Context, Poll},
27    time::{Duration, Instant},
28};
29
30use iroh::{
31    PublicKey,
32    address_lookup::{
33        AddressLookup, EndpointData, EndpointInfo, Error as AddressLookupError, Item as AddressLookupItem,
34    },
35};
36use n0_future::{Stream, boxed::BoxStream, task::AbortOnDropHandle};
37use serde::{Deserialize, Serialize};
38use tokio::{net::UdpSocket, sync::mpsc};
39use tracing::{debug, warn};
40
41use crate::error::SyncwebError;
42
43/// Magic value that prefixes every beacon datagram. It both identifies the
44/// protocol and provides cheap sanity checking before JSON parsing.
45pub const BEACON_MAGIC: &str = "syncweb-beacon-v1";
46
47/// Default base UDP port for the beacon. The effective port is
48/// `base + (scope[0..2] % BEACON_PORT_SPREAD)`.
49pub const DEFAULT_BEACON_PORT: u16 = 15_200;
50
51/// Number of ports a scope may spread over when deriving the effective port.
52pub const BEACON_PORT_SPREAD: u16 = 2_048;
53
54/// How long a discovered peer's addresses stay usable without a refresh.
55pub const BEACON_PEER_TTL: Duration = Duration::from_secs(30);
56
57/// How long a resolve request waits for a fresh beacon before giving up.
58const LOOKUP_DURATION: Duration = Duration::from_secs(10);
59
60/// Maximum accepted datagram size.
61const MAX_DATAGRAM_SIZE: usize = 2_048;
62
63/// Number of beacons sent back-to-back when the published data changes.
64const BURST_COUNT: usize = 3;
65
66/// A single interface with its IPv4 address and directed broadcast address.
67struct InterfaceInfo {
68    name: String,
69    addr: Ipv4Addr,
70    broadcast: Ipv4Addr,
71}
72
73/// Datagram payload broadcast by a beacon.
74#[derive(Debug, Clone, Serialize, Deserialize)]
75struct BeaconMessage {
76    magic: String,
77    scope: [u8; 16],
78    id: String,
79    addrs: Vec<SocketAddr>,
80    relay: Option<String>,
81}
82
83/// Pending response channels for a resolve request.
84struct PendingResolve {
85    deadline: Instant,
86    sender: mpsc::Sender<Result<AddressLookupItem, AddressLookupError>>,
87}
88
89/// A peer's published data and when it was last heard from.
90struct PeerEntry {
91    data: Arc<EndpointData>,
92    last_seen: Instant,
93}
94
95/// Messages sent to the beacon actor.
96enum Message {
97    Publish(Arc<EndpointData>),
98    Resolve(PublicKey, mpsc::Sender<Result<AddressLookupItem, AddressLookupError>>),
99}
100
101/// An [`AddressLookup`] that publishes and resolves addresses over a UDP beacon.
102///
103/// The `_handle` field is never read; it exists only so that dropping the lookup
104/// aborts the background actor.
105#[derive(Debug, Clone)]
106pub struct BeaconAddressLookup {
107    sender: mpsc::Sender<Message>,
108    _handle: Arc<AbortOnDropHandle<()>>,
109}
110
111/// A small adapter exposing a [`tokio::sync::mpsc::Receiver`] as a stream.
112struct ReceiverStream<T>(mpsc::Receiver<T>);
113
114impl<T> ReceiverStream<T> {
115    const fn new(receiver: mpsc::Receiver<T>) -> Self {
116        Self(receiver)
117    }
118}
119
120impl<T> Stream for ReceiverStream<T> {
121    type Item = T;
122
123    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
124        self.0.poll_recv(cx)
125    }
126}
127
128impl BeaconAddressLookup {
129    /// Creates a beacon address lookup and starts its background task.
130    ///
131    /// The receiver socket is bound to the effective port on all interfaces
132    /// (or on the configured interface when one is given) and the sender
133    /// broadcasts to `255.255.255.255` plus the directed broadcast of every
134    /// up, non-loopback IPv4 interface (or only the configured one).
135    ///
136    /// # Errors
137    ///
138    /// Returns an error if the receiver or sender socket cannot be created.
139    pub fn new(
140        endpoint_id: PublicKey,
141        scope: Option<[u8; 16]>,
142        base_port: u16,
143        interval: Duration,
144        interface: Option<&str>,
145    ) -> crate::error::Result<Self> {
146        let port = beacon_port(base_port, scope.as_ref());
147        let bind_ip = interface.and_then(interface_ipv4).or_else(|| {
148            if interface.is_some() {
149                warn!(interface, "beacon interface not found; using all interfaces");
150            }
151            None
152        });
153        let bind_addr = SocketAddr::new(bind_ip.map_or(IpAddr::V4(Ipv4Addr::UNSPECIFIED), IpAddr::V4), port);
154        let targets = broadcast_targets(interface, port);
155        Self::with_bind_targets(endpoint_id, scope, bind_addr, targets, interval)
156    }
157
158    /// Creates a beacon with an explicit bind address and set of targets.
159    fn with_bind_targets(
160        endpoint_id: PublicKey,
161        scope: Option<[u8; 16]>,
162        bind_addr: SocketAddr,
163        targets: Vec<SocketAddr>,
164        beacon_interval: Duration,
165    ) -> crate::error::Result<Self> {
166        let recv_socket = tokio::net::UdpSocket::from_std(beacon_socket(bind_addr, true)?)?;
167        let send_socket = tokio::net::UdpSocket::from_std(beacon_socket(SocketAddr::new(bind_addr.ip(), 0), false)?)?;
168
169        let (sender, mut recv) = mpsc::channel(64);
170        let actor = async move {
171            let mut advertised: Option<Arc<EndpointData>> = None;
172            let mut peers: HashMap<PublicKey, PeerEntry> = HashMap::new();
173            let mut pending: HashMap<PublicKey, Vec<PendingResolve>> = HashMap::new();
174            let mut buf = vec![0_u8; MAX_DATAGRAM_SIZE];
175            let mut timer = tokio::time::interval(beacon_interval.max(Duration::from_millis(1)));
176            loop {
177                tokio::select! {
178                    _ = timer.tick() => {
179                        broadcast_beacon(&send_socket, &targets, advertised.as_deref(), scope, endpoint_id).await;
180                        let now = Instant::now();
181                        sweep(&mut peers, &mut pending, now);
182                    }
183                    message = recv.recv() => {
184                        match message {
185                            None => break,
186                            Some(Message::Publish(data)) => {
187                                debug!("beacon data published");
188                                advertised = Some(data.clone());
189                                for _ in 0..BURST_COUNT {
190                                    broadcast_beacon(&send_socket, &targets, Some(data.as_ref()), scope, endpoint_id).await;
191                                }
192                            }
193                            Some(Message::Resolve(id, responder)) => {
194                                if let Some(entry) = peers.get(&id) {
195                                    let _ = responder.send(Ok(make_item(id, entry.data.as_ref()))).await;
196                                } else {
197                                    pending.entry(id).or_default().push(PendingResolve {
198                                        deadline: Instant::now().checked_add(LOOKUP_DURATION).unwrap_or_else(Instant::now),
199                                        sender: responder,
200                                    });
201                                }
202                            }
203                        }
204                    }
205                    received = recv_socket.recv_from(&mut buf) => {
206                        match received {
207                            Ok((length, _source)) => {
208                                if let Some(message) = parse_beacon(buf.get(..length)) {
209                                    handle_discovered(&message, endpoint_id, scope, &mut peers, &mut pending).await;
210                                }
211                            }
212                            Err(error) => warn!(%error, "beacon receive failed"),
213                        }
214                    }
215                }
216            }
217        };
218        let handle = tokio::spawn(actor);
219        Ok(Self {
220            sender,
221            _handle: Arc::new(AbortOnDropHandle::new(handle)),
222        })
223    }
224}
225
226/// Derives the effective beacon port for a scope.
227#[must_use]
228pub const fn beacon_port(base: u16, scope: Option<&[u8; 16]>) -> u16 {
229    let offset = match scope {
230        Some(bytes) => u16::from_be_bytes([bytes[0], bytes[1]]) % BEACON_PORT_SPREAD,
231        None => 0,
232    };
233    base.saturating_add(offset)
234}
235
236/// Builds a bound UDP socket with broadcast enabled and non-blocking I/O.
237fn beacon_socket(bind_addr: SocketAddr, reuse: bool) -> crate::error::Result<std::net::UdpSocket> {
238    let socket = std::net::UdpSocket::bind(bind_addr)
239        .map_err(|error| SyncwebError::operation("failed to bind beacon socket", error))?;
240    socket
241        .set_broadcast(true)
242        .map_err(|error| SyncwebError::operation("failed to enable beacon broadcast", error))?;
243    socket
244        .set_nonblocking(true)
245        .map_err(|error| SyncwebError::operation("failed to set beacon socket non-blocking", error))?;
246    if reuse {
247        set_reuseaddr(&socket)?;
248    }
249    Ok(socket)
250}
251
252/// Sets `SO_REUSEADDR` on a UDP socket so multiple daemons on one host can
253/// share the same beacon port.
254fn set_reuseaddr(socket: &std::net::UdpSocket) -> crate::error::Result<()> {
255    use std::os::fd::AsRawFd;
256
257    let one: libc::c_int = 1;
258    let option_length = libc::socklen_t::try_from(std::mem::size_of::<libc::c_int>())
259        .map_err(|error| SyncwebError::operation("invalid beacon socket option length", error))?;
260    // SAFETY: `socket.as_raw_fd()` is a valid socket descriptor for the lifetime
261    // of `socket`, and setting SO_REUSEADDR on a UDP socket only affects the
262    // kernel's bind behavior; it cannot cause memory unsafety.
263    let result = unsafe {
264        libc::setsockopt(
265            socket.as_raw_fd(),
266            libc::SOL_SOCKET,
267            libc::SO_REUSEADDR,
268            ptr::from_ref(&one).cast(),
269            option_length,
270        )
271    };
272    if result == 0 {
273        Ok(())
274    } else {
275        Err(SyncwebError::operation(
276            "failed to set SO_REUSEADDR on beacon socket",
277            std::io::Error::last_os_error(),
278        ))
279    }
280}
281
282/// The addresses every beacon is broadcast to: the limited broadcast plus the
283/// directed broadcast of every relevant IPv4 interface.
284fn broadcast_targets(interface: Option<&str>, port: u16) -> Vec<SocketAddr> {
285    let mut targets = Vec::with_capacity(8);
286    targets.push(SocketAddr::new(Ipv4Addr::BROADCAST.into(), port));
287    for iface in ipv4_interfaces(interface) {
288        let target = SocketAddr::new(iface.broadcast.into(), port);
289        if !targets.contains(&target) {
290            targets.push(target);
291        }
292    }
293    targets
294}
295
296/// IPv4 interfaces that can carry beacon traffic.
297fn ipv4_interfaces(interface: Option<&str>) -> Vec<InterfaceInfo> {
298    let mut interfaces = Vec::new();
299    for iface in ipv4_interfaces_raw() {
300        if interface.is_none_or(|name| iface.name == name) {
301            interfaces.push(iface);
302        }
303    }
304    interfaces
305}
306
307/// Resolves an interface name to its IPv4 address, if any.
308fn interface_ipv4(interface: &str) -> Option<Ipv4Addr> {
309    ipv4_interfaces(Some(interface))
310        .into_iter()
311        .next()
312        .map(|iface| iface.addr)
313}
314
315/// Enumerates IPv4 interfaces from the operating system.
316#[cfg(unix)]
317fn ipv4_interfaces_raw() -> Vec<InterfaceInfo> {
318    use std::ffi::CStr;
319
320    let mut ifap: *mut libc::ifaddrs = ptr::null_mut();
321    // SAFETY: `getifaddrs` allocates a linked list of interface records that
322    // must be released with `freeifaddrs`; we guarantee that on every exit path.
323    if unsafe { libc::getifaddrs(ptr::from_mut(&mut ifap)) } != 0 {
324        warn!("failed to enumerate network interfaces for the beacon");
325        return Vec::new();
326    }
327    let mut interfaces = Vec::new();
328    let mut current = ifap;
329    while !current.is_null() {
330        // SAFETY: `current` points to a node in the list allocated by
331        // `getifaddrs` that remains valid until `freeifaddrs` is called below.
332        let ifa = unsafe { &*current };
333        let flags = ifa.ifa_flags;
334        if flag_is_set(flags, libc::IFF_UP)
335            && !flag_is_set(flags, libc::IFF_LOOPBACK)
336            && let Some(addr) = sockaddr_ipv4(ifa.ifa_addr)
337            && let Some(netmask) = sockaddr_ipv4(ifa.ifa_netmask)
338        {
339            let addr_bits = u32::from_be_bytes(addr.octets());
340            let mask_bits = u32::from_be_bytes(netmask.octets());
341            if mask_bits != 0 {
342                // SAFETY: `ifa_name` points to a NUL-terminated interface
343                // name that is valid for the lifetime of `ifa`.
344                let name = unsafe { CStr::from_ptr(ifa.ifa_name) }.to_string_lossy().into_owned();
345                interfaces.push(InterfaceInfo {
346                    name,
347                    addr,
348                    broadcast: Ipv4Addr::from(addr_bits | !mask_bits),
349                });
350            }
351        }
352        current = ifa.ifa_next;
353    }
354    // SAFETY: `ifap` still points to the list returned by `getifaddrs` above.
355    unsafe { libc::freeifaddrs(ifap) };
356    interfaces
357}
358
359/// Enumerates IPv4 interfaces from the operating system.
360#[cfg(not(unix))]
361fn ipv4_interfaces_raw() -> Vec<InterfaceInfo> {
362    Vec::new()
363}
364
365/// Tests whether an interface flag constant is set.
366fn flag_is_set(flags: u32, flag: i32) -> bool {
367    flags & u32::try_from(flag).unwrap_or(0) != 0
368}
369
370/// Reads the IPv4 address from a `sockaddr`, if it is an IPv4 address.
371fn sockaddr_ipv4(raw: *mut libc::sockaddr) -> Option<Ipv4Addr> {
372    if raw.is_null() {
373        return None;
374    }
375    // SAFETY: dereferencing `sa_family` is always safe for a valid `sockaddr`.
376    let family = unsafe { (*raw).sa_family };
377    if family != libc::AF_INET.try_into().unwrap_or(0) {
378        return None;
379    }
380    // SAFETY: the family is AF_INET, so `raw` points to a valid `sockaddr_in`.
381    let address = unsafe { ptr::read_unaligned(raw.cast::<libc::sockaddr_in>()) };
382    Some(Ipv4Addr::from(address.sin_addr.s_addr))
383}
384
385/// Serializes the current endpoint data into a beacon datagram.
386fn encode_message(data: &EndpointData, scope: Option<[u8; 16]>, endpoint_id: PublicKey) -> BeaconMessage {
387    BeaconMessage {
388        magic: BEACON_MAGIC.to_owned(),
389        scope: scope.unwrap_or([0; 16]),
390        id: endpoint_id.to_string(),
391        addrs: data.ip_addrs().copied().collect(),
392        relay: data.relay_urls().next().map(ToString::to_string),
393    }
394}
395
396/// Parses and sanity-checks a received datagram.
397fn parse_beacon(data: Option<&[u8]>) -> Option<BeaconMessage> {
398    let message: BeaconMessage = serde_json::from_slice(data?).ok()?;
399    (message.magic == BEACON_MAGIC).then_some(message)
400}
401
402/// Records a discovered peer and answers any pending resolve requests.
403async fn handle_discovered(
404    message: &BeaconMessage,
405    endpoint_id: PublicKey,
406    scope: Option<[u8; 16]>,
407    peers: &mut HashMap<PublicKey, PeerEntry>,
408    pending: &mut HashMap<PublicKey, Vec<PendingResolve>>,
409) {
410    if let Some(expected) = scope
411        && message.scope != expected
412    {
413        return;
414    }
415    let Ok(peer_id) = PublicKey::from_str(&message.id) else {
416        return;
417    };
418    if peer_id == endpoint_id {
419        return;
420    }
421    let data = message_to_endpoint_data(message);
422    peers.insert(
423        peer_id,
424        PeerEntry {
425            data: Arc::new(data.clone()),
426            last_seen: Instant::now(),
427        },
428    );
429    if let Some(senders) = pending.get_mut(&peer_id) {
430        for resolve in senders.drain(..) {
431            let _ = resolve.sender.send(Ok(make_item(peer_id, &data))).await;
432        }
433    }
434}
435
436/// Converts a parsed beacon message into endpoint data.
437fn message_to_endpoint_data(message: &BeaconMessage) -> EndpointData {
438    let addrs: BTreeSet<SocketAddr> = message.addrs.iter().copied().collect();
439    let mut data = EndpointData::from(addrs);
440    if let Some(relay) = &message.relay
441        && let Ok(relay_url) = relay.parse()
442    {
443        data.add_relay_url(relay_url);
444    }
445    data
446}
447
448/// Builds the item returned to resolve callers.
449fn make_item(endpoint_id: PublicKey, data: &EndpointData) -> AddressLookupItem {
450    AddressLookupItem::new(EndpointInfo::from_parts(endpoint_id, data.clone()), "beacon", None)
451}
452
453/// Sends the current endpoint data to every broadcast target.
454async fn broadcast_beacon(
455    socket: &UdpSocket,
456    targets: &[SocketAddr],
457    data: Option<&EndpointData>,
458    scope: Option<[u8; 16]>,
459    endpoint_id: PublicKey,
460) {
461    let Some(current) = data else {
462        return;
463    };
464    let Ok(payload) = serde_json::to_vec(&encode_message(current, scope, endpoint_id)) else {
465        warn!("failed to serialize beacon payload");
466        return;
467    };
468    for target in targets {
469        if let Err(error) = socket.send_to(&payload, *target).await {
470            warn!(%target, %error, "beacon broadcast failed");
471        }
472    }
473}
474
475/// Drops peers and pending resolves that have exceeded their lifetime.
476fn sweep(
477    peers: &mut HashMap<PublicKey, PeerEntry>,
478    pending: &mut HashMap<PublicKey, Vec<PendingResolve>>,
479    now: Instant,
480) {
481    peers.retain(|_, entry| {
482        entry
483            .last_seen
484            .checked_add(BEACON_PEER_TTL)
485            .is_some_and(|deadline| deadline > now)
486    });
487    pending.retain(|_, resolves| {
488        resolves.retain(|resolve| resolve.deadline > now);
489        !resolves.is_empty()
490    });
491}
492
493impl AddressLookup for BeaconAddressLookup {
494    fn resolve(&self, endpoint_id: PublicKey) -> Option<BoxStream<Result<AddressLookupItem, AddressLookupError>>> {
495        let (send, recv) = mpsc::channel(20);
496        let sender = self.sender.clone();
497        tokio::spawn(async move {
498            let _ = sender.send(Message::Resolve(endpoint_id, send)).await;
499        });
500        Some(Box::pin(ReceiverStream::new(recv)))
501    }
502
503    fn publish(&self, data: &EndpointData) {
504        let _ = self.sender.try_send(Message::Publish(Arc::new(data.clone())));
505    }
506}
507
508#[cfg(test)]
509mod tests {
510    use super::*;
511    use n0_future::StreamExt;
512    use std::net::{IpAddr, Ipv4Addr};
513    use tokio::time::timeout;
514
515    fn endpoint_id(seed: u8) -> PublicKey {
516        let mut bytes = [0_u8; 32];
517        bytes[0] = seed;
518        let secret = iroh::SecretKey::from_bytes(&bytes);
519        secret.public()
520    }
521
522    fn endpoint_data(port: u16) -> EndpointData {
523        EndpointData::from(BTreeSet::from([SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port)]))
524    }
525
526    fn make_beacon(id: PublicKey, scope: [u8; 16], bind_port: u16, target_port: u16) -> BeaconAddressLookup {
527        let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), bind_port);
528        let targets = vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), target_port)];
529        BeaconAddressLookup::with_bind_targets(id, Some(scope), bind_addr, targets, Duration::from_secs(1))
530            .expect("beacon should bind")
531    }
532
533    #[tokio::test]
534    async fn beacon_port_is_scoped() {
535        let scope = [7_u8; 16];
536        let port = beacon_port(DEFAULT_BEACON_PORT, Some(&scope));
537        let offset = u16::from_be_bytes([scope[0], scope[1]]) % BEACON_PORT_SPREAD;
538        assert_eq!(port, DEFAULT_BEACON_PORT.saturating_add(offset));
539        assert_eq!(beacon_port(DEFAULT_BEACON_PORT, None), DEFAULT_BEACON_PORT);
540    }
541
542    #[tokio::test]
543    async fn publish_and_resolve_over_loopback() {
544        let scope = [9_u8; 16];
545        let id_a = endpoint_id(1);
546        let id_b = endpoint_id(2);
547        let beacon_a = make_beacon(id_a, scope, 21_000, 21_001);
548        let beacon_b = make_beacon(id_b, scope, 21_001, 21_000);
549
550        beacon_a.publish(&endpoint_data(42_000));
551        beacon_b.publish(&endpoint_data(42_001));
552
553        let mut stream = beacon_a.resolve(id_b).expect("resolve stream");
554        let item = timeout(Duration::from_secs(5), stream.next())
555            .await
556            .expect("resolve should complete in time")
557            .expect("stream should yield an item")
558            .expect("resolve should succeed");
559        assert_eq!(item.endpoint_info().endpoint_id, id_b);
560        let published: BTreeSet<SocketAddr> = item.ip_addrs().copied().collect();
561        assert_eq!(
562            published,
563            BTreeSet::from([SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 42_001)])
564        );
565    }
566
567    #[tokio::test]
568    async fn different_scope_is_ignored() {
569        let scope_a = [11_u8; 16];
570        let scope_b = [13_u8; 16];
571        let id_a = endpoint_id(3);
572        let id_b = endpoint_id(4);
573
574        let message = BeaconMessage {
575            magic: BEACON_MAGIC.to_owned(),
576            scope: scope_b,
577            id: id_b.to_string(),
578            addrs: vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 42_011)],
579            relay: None,
580        };
581        let mut peers = HashMap::new();
582        let mut pending = HashMap::new();
583        handle_discovered(&message, id_a, Some(scope_a), &mut peers, &mut pending).await;
584        assert!(peers.is_empty());
585        assert!(pending.is_empty());
586    }
587}
syncweb_core/node/beacon_lookup.rs

syncweb_core/node/
beacon_lookup.rs
