From a2adc104add744a34eb6c231b281771f39959fc6 Mon Sep 17 00:00:00 2001 From: Lucien Loiseau Date: Thu, 27 Nov 2025 14:47:23 +0100 Subject: [PATCH 1/2] mdns discovery --- Cargo.lock | 69 ++++++ core/dtn7/Cargo.toml | 1 + core/dtn7/src/bin/dtnd.rs | 15 +- core/dtn7/src/dtnconfig.rs | 43 +++- core/dtn7/src/ipnd/mdns_discovery.rs | 281 ++++++++++++++++++++++ core/dtn7/src/ipnd/mod.rs | 2 + core/dtn7/src/ipnd/neighbour_discovery.rs | 241 ++----------------- core/dtn7/src/ipnd/udp_discovery.rs | 233 ++++++++++++++++++ doc/mdns-discovery.md | 111 +++++++++ examples/dtn7.toml.example | 6 + 10 files changed, 778 insertions(+), 224 deletions(-) create mode 100644 core/dtn7/src/ipnd/mdns_discovery.rs create mode 100644 core/dtn7/src/ipnd/udp_discovery.rs create mode 100644 doc/mdns-discovery.md diff --git a/Cargo.lock b/Cargo.lock index 20355cbc5..8492ddd16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -392,6 +392,15 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7" +[[package]] +name = "concurrent-queue" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "config" version = "0.14.0" @@ -595,6 +604,7 @@ dependencies = [ "hyper", "lazy_static", "log", + "mdns-sd", "num-derive", "num-traits", "parking_lot 0.12.1", @@ -785,6 +795,17 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1135,6 +1156,16 @@ dependencies = [ "unicode-normalization", ] +[[package]] +name = "if-addrs" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a78a89907582615b19f6f0da1af18abf6ff08be259395669b834b057a7ee92d8" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "indexmap" version = "1.9.3" @@ -1272,6 +1303,19 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" +[[package]] +name = "mdns-sd" +version = "0.11.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fe7c11a1eb3cfbfcf702d1601c1f5f4c102cdc8665b8a557783ef634741676e" +dependencies = [ + "flume", + "if-addrs", + "log", + "polling", + "socket2", +] + [[package]] name = "memchr" version = "2.7.2" @@ -1485,6 +1529,22 @@ version = "0.3.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d231b230927b5e4ad203db57bbcbee2802f6bce620b1e4a9024a07d94e2907ec" +[[package]] +name = "polling" +version = "2.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b2d323e8ca7996b3e23126511a523f7e62924d93ecd5ae73b333815b0eb3dce" +dependencies = [ + "autocfg", + "bitflags 1.3.2", + "cfg-if", + "concurrent-queue", + "libc", + "log", + "pin-project-lite", + "windows-sys 0.48.0", +] + [[package]] name = "ppv-lite86" version = "0.2.17" @@ -1896,6 +1956,15 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] + [[package]] name = "stdweb" version = "0.4.20" diff --git a/core/dtn7/Cargo.toml b/core/dtn7/Cargo.toml index f36540c19..7a4b84028 100644 --- a/core/dtn7/Cargo.toml +++ b/core/dtn7/Cargo.toml @@ -78,6 +78,7 @@ dtn7-codegen = { path = "../codegen", version = "0.1.2" } sha1 = "0.10.5" glob-match = "0.2.1" tower-http = { version = "0.3.4", features = ["cors"] } +mdns-sd = "0.11" [lib] name = "dtn7" diff --git a/core/dtn7/src/bin/dtnd.rs b/core/dtn7/src/bin/dtnd.rs index a689fc725..83e02f94f 100644 --- a/core/dtn7/src/bin/dtnd.rs +++ b/core/dtn7/src/bin/dtnd.rs @@ -129,7 +129,16 @@ async fn main() -> Result<(), std::io::Error> { .help("Sets listening port for IPND node discovery (default = 3003)") .value_parser(value_parser!(u16)) .action(ArgAction::Set), - ).arg( + ) + .arg( + Arg::new("discoverytransport") + .long("discovery-transport") + .value_name("TRANSPORT") + .help("Sets discovery transport: udp-multicast (default), mdns, or both") + .value_parser(value_parser!(String)) + .action(ArgAction::Set), + ) + .arg( Arg::new("webport") .short('w') .long("web-port") @@ -354,6 +363,10 @@ Tag 255 takes 5 arguments and is interpreted as address. Usage: -S 255:'Samplest if let Some(i) = matches.get_one::("discoveryport") { cfg.discovery_listen_port = *i; } + if let Some(transport) = matches.get_one::("discoverytransport") { + cfg.discovery_transport = transport.parse() + .expect("Invalid discovery transport! Use: udp-multicast, mdns, or both"); + } if let Some(i) = matches.get_one::("webport") { cfg.webport = *i; } diff --git a/core/dtn7/src/dtnconfig.rs b/core/dtn7/src/dtnconfig.rs index 2d605d16c..eabe37c56 100644 --- a/core/dtn7/src/dtnconfig.rs +++ b/core/dtn7/src/dtnconfig.rs @@ -6,13 +6,44 @@ use config::{Config, File}; use log::{debug, error}; use rand::distributions::Alphanumeric; use rand::{thread_rng, Rng}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap}; use std::net::SocketAddr; use std::path::PathBuf; use std::str::FromStr; use std::{convert::TryInto, time::Duration}; +/// Discovery transport mechanism +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum DiscoveryTransport { + /// UDP multicast (legacy IPND) + UdpMulticast, + /// mDNS/DNS-SD service discovery + Mdns, + /// Both UDP multicast and mDNS simultaneously + Both, +} + +impl Default for DiscoveryTransport { + fn default() -> Self { + Self::UdpMulticast + } +} + +impl FromStr for DiscoveryTransport { + type Err = String; + + fn from_str(s: &str) -> Result { + match s.to_lowercase().as_str() { + "udp-multicast" | "udp" => Ok(Self::UdpMulticast), + "mdns" | "dns-sd" => Ok(Self::Mdns), + "both" => Ok(Self::Both), + _ => Err(format!("Unknown discovery transport: {}", s)), + } + } +} + #[derive(Debug, Default, Clone, Serialize)] pub struct DtnConfig { pub debug: bool, @@ -26,6 +57,7 @@ pub struct DtnConfig { pub webport: u16, pub announcement_interval: Duration, pub disable_neighbour_discovery: bool, + pub discovery_transport: DiscoveryTransport, pub discovery_destinations: BTreeMap, pub discovery_listen_port: u16, pub janitor_interval: Duration, @@ -156,6 +188,13 @@ impl From for DtnConfig { }; debug!("discovery-interval: {:?}", dtncfg.announcement_interval); + dtncfg.discovery_transport = if let Ok(transport) = s.get_string("discovery.transport") { + DiscoveryTransport::from_str(&transport).unwrap_or_default() + } else { + dtncfg.discovery_transport + }; + debug!("discovery-transport: {:?}", dtncfg.discovery_transport); + dtncfg.peer_timeout = if let Ok(interval) = s.get_string("discovery.peer-timeout") { humantime::parse_duration(&interval).unwrap_or_else(|_| Duration::new(0, 0)) } else { @@ -277,6 +316,7 @@ impl DtnConfig { host_eid: local_node_id, announcement_interval: "2s".parse::().unwrap().into(), disable_neighbour_discovery: false, + discovery_transport: DiscoveryTransport::default(), discovery_destinations: BTreeMap::new(), discovery_listen_port: 3003, webport: 3000, @@ -309,6 +349,7 @@ impl DtnConfig { self.webport = cfg.webport; self.announcement_interval = cfg.announcement_interval; self.disable_neighbour_discovery = cfg.disable_neighbour_discovery; + self.discovery_transport = cfg.discovery_transport; self.discovery_destinations = cfg.discovery_destinations; self.discovery_listen_port = cfg.discovery_listen_port; self.janitor_interval = cfg.janitor_interval; diff --git a/core/dtn7/src/ipnd/mdns_discovery.rs b/core/dtn7/src/ipnd/mdns_discovery.rs new file mode 100644 index 000000000..6110ba4d4 --- /dev/null +++ b/core/dtn7/src/ipnd/mdns_discovery.rs @@ -0,0 +1,281 @@ +use crate::cla::ConvergenceLayerAgent; +use crate::core::{DtnPeer, PeerType}; +use crate::routing::RoutingNotifcation; +use crate::{peers_add, peers_remove, peers_touch, routing_notify, CLAS, CONFIG, DTNCORE}; +use anyhow::Result; +use log::{debug, error, info, trace, warn}; +use std::collections::HashMap; +use std::net::IpAddr; +use tokio::time::interval; + +const MDNS_SERVICE_TYPE: &str = "_dtn._udp.local."; + +pub async fn spawn_mdns_discovery() -> Result<()> { + use mdns_sd::ServiceDaemon; + + let config = CONFIG.lock(); + let nodeid = config.nodeid.clone(); + let eid = config.host_eid.clone(); + let interval_duration = config.announcement_interval; + drop(config); + + info!("Starting mDNS discovery for service type: {}", MDNS_SERVICE_TYPE); + + // Create mDNS daemon + let mdns = ServiceDaemon::new().expect("Failed to create mDNS daemon"); + + // Build and register our service + let service_info = build_mdns_service_info(&nodeid, &eid); + mdns.register(service_info) + .expect("Failed to register mDNS service"); + + info!("Registered mDNS service type: {} (Node: {})", MDNS_SERVICE_TYPE, nodeid); + + // Browse for other DTN nodes + let receiver = mdns.browse(MDNS_SERVICE_TYPE) + .expect("Failed to browse for mDNS services"); + + // Spawn listener task + tokio::spawn(async move { + loop { + match receiver.recv_async().await { + Ok(event) => { + if let Err(e) = handle_mdns_event(event).await { + error!("Error handling mDNS event: {}", e); + } + } + Err(e) => { + error!("mDNS receiver error: {}", e); + break; + } + } + } + }); + + // Spawn periodic refresh task (update service info when CLAs change) + tokio::spawn(async move { + let mut task = interval(interval_duration); + loop { + task.tick().await; + + let config = CONFIG.lock(); + let nodeid = config.nodeid.clone(); + let eid = config.host_eid.clone(); + drop(config); + + // Re-register service with updated info (including fresh timestamp) + let service_info = build_mdns_service_info(&nodeid, &eid); + if let Err(e) = mdns.register(service_info) { + warn!("Failed to refresh mDNS service: {}", e); + } else { + debug!("Refreshed mDNS service registration with updated timestamp"); + } + } + }); + + Ok(()) +} + +fn build_mdns_service_info(nodeid: &str, eid: &bp7::EndpointID) -> mdns_sd::ServiceInfo { + use mdns_sd::ServiceInfo; + + let mut properties: Vec<(&str, String)> = Vec::new(); + + // Add endpoint ID + let eid_str = eid.to_string(); + properties.push(("eid", eid_str)); + + // Add timestamp to force mDNS to broadcast updates even when other data hasn't changed + // This ensures remote nodes receive periodic ServiceResolved events for keepalive + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + properties.push(("ts", timestamp.to_string())); + + // Add available CLAs with their ports + let clas = CLAS.lock(); + let cla_props: Vec<(String, String)> = clas.iter() + .enumerate() + .map(|(idx, cla)| { + (format!("cla{}", idx), format!("{}:{}", cla.name(), cla.port())) + }) + .collect(); + drop(clas); + + for (key, val) in cla_props { + properties.push((Box::leak(key.into_boxed_str()), val)); + } + + // Add custom services + let services = DTNCORE.lock().service_list.clone(); + let svc_props: Vec<(String, String)> = services.iter() + .map(|(tag, payload)| (format!("svc{}", tag), payload.clone())) + .collect(); + + for (key, val) in svc_props { + properties.push((Box::leak(key.into_boxed_str()), val)); + } + + // Convert to format mdns-sd expects + let props_ref: Vec<(&str, &str)> = properties.iter() + .map(|(k, v)| (*k, v.as_str())) + .collect(); + + // Use a dummy port (actual CLAs have their own ports) + // The hostname will be .local + ServiceInfo::new( + MDNS_SERVICE_TYPE, + nodeid, + &format!("{}.local.", nodeid), + "", + 0, // Port not used for DTN discovery + &props_ref[..], + ) + .expect("Failed to create ServiceInfo") + .enable_addr_auto() +} + +async fn handle_mdns_event(event: mdns_sd::ServiceEvent) -> Result<()> { + use mdns_sd::ServiceEvent; + + match event { + ServiceEvent::ServiceResolved(info) => { + trace!("mDNS service resolved: {:?}", info); + handle_mdns_peer_discovered(info).await?; + } + ServiceEvent::ServiceRemoved(_, fullname) => { + debug!("mDNS service removed: {}", fullname); + handle_mdns_peer_removed(&fullname).await?; + } + ServiceEvent::ServiceFound(_, fullname) => { + // Touch peer to keep it alive (mDNS announces service existence) + // Note: This event fires before ServiceResolved, so the peer might not exist yet + debug!("mDNS ServiceFound event (keepalive): {}", fullname); + if let Some(instance_name) = fullname.split('.').next() { + // Extract node name from EID (e.g., "dtn://node1/" -> "node1") + let node_name = if instance_name.starts_with("dtn://") && instance_name.ends_with('/') { + &instance_name[6..instance_name.len()-1] + } else { + instance_name + }; + + // Silently ignore if peer doesn't exist yet - ServiceResolved will add it + if let Ok(_) = peers_touch(node_name) { + debug!("Successfully touched peer: {}", node_name); + } + } + } + ServiceEvent::SearchStarted(_) => { + debug!("mDNS search started"); + } + ServiceEvent::SearchStopped(_) => { + warn!("mDNS search stopped"); + } + } + + Ok(()) +} + +async fn handle_mdns_peer_discovered(info: mdns_sd::ServiceInfo) -> Result<()> { + // Extract EID from TXT records + let eid_str = info.get_property_val_str("eid") + .ok_or_else(|| anyhow::anyhow!("No EID in mDNS service info"))?; + + let eid: bp7::EndpointID = bp7::eid::EndpointID::try_from(eid_str.as_ref()) + .map_err(|e| anyhow::anyhow!("Invalid EID in mDNS service: {:?}", e))?; + + // Don't add ourselves + let our_eid = CONFIG.lock().host_eid.clone(); + if eid == our_eid { + debug!("Ignoring mDNS service from ourselves: {}", eid); + return Ok(()); + } + + // Extract CLAs from TXT records + let mut clas = Vec::new(); + for prop in info.get_properties().iter() { + let key = prop.key(); + if key.starts_with("cla") { + if let Some(val) = info.get_property_val_str(key) { + if let Some((name, port_str)) = val.split_once(':') { + let port = port_str.parse().ok(); + clas.push((name.to_string(), port)); + } + } + } + } + + // Extract custom services + let mut services = HashMap::new(); + for prop in info.get_properties().iter() { + let key = prop.key(); + if key.starts_with("svc") { + if let Ok(tag) = key.strip_prefix("svc").unwrap_or("").parse::() { + if let Some(val) = info.get_property_val_str(key) { + services.insert(tag, val.to_string()); + } + } + } + } + + // Get IP addresses from mDNS response + let addresses: Vec = info.get_addresses().iter().cloned().collect(); + + if addresses.is_empty() { + return Err(anyhow::anyhow!("No addresses in mDNS service info")); + } + + // Log all discovered addresses + let addr_str = addresses.iter() + .map(|a| a.to_string()) + .collect::>() + .join(", "); + + // Prefer IPv4 over IPv6 link-local addresses for better reachability + let addr = addresses.iter() + .find(|ip| ip.is_ipv4()) + .or_else(|| addresses.iter().find(|ip| !ip.to_string().starts_with("fe80::"))) + .or_else(|| addresses.first()) + .ok_or_else(|| anyhow::anyhow!("No valid address in mDNS service info"))? + .clone(); + + debug!("Selected address {} from [{}]", addr, addr_str); + + // Create peer + let peer = DtnPeer::new( + eid.clone(), + addr.into(), + PeerType::Dynamic, + None, // mDNS handles timing + clas, + services, + ); + + let is_new = peers_add(peer); + if let Err(e) = peers_touch(eid.node().unwrap().as_ref()) { + warn!("Failed to touch peer after add: {}", e); + } + + if is_new { + info!("New peer discovered via mDNS: {}", eid); + } else { + debug!("Updated existing peer via mDNS: {} @ [{}]", eid, addr_str); + } + + // Notify routing + routing_notify(RoutingNotifcation::EncounteredPeer(eid)).await?; + + Ok(()) +} + +async fn handle_mdns_peer_removed(fullname: &str) -> Result<()> { + // Extract instance name (nodeid) from fullname + let instance = fullname.split('.').next() + .ok_or_else(|| anyhow::anyhow!("Invalid mDNS fullname"))?; + + info!("Removing peer discovered via mDNS: {}", instance); + peers_remove(instance); + + Ok(()) +} diff --git a/core/dtn7/src/ipnd/mod.rs b/core/dtn7/src/ipnd/mod.rs index 82e766ec5..c45f126eb 100644 --- a/core/dtn7/src/ipnd/mod.rs +++ b/core/dtn7/src/ipnd/mod.rs @@ -1,4 +1,6 @@ #![forbid(unsafe_code)] pub mod beacon; +pub mod mdns_discovery; pub mod neighbour_discovery; pub mod services; +pub mod udp_discovery; diff --git a/core/dtn7/src/ipnd/neighbour_discovery.rs b/core/dtn7/src/ipnd/neighbour_discovery.rs index 1225c96f4..c90470169 100644 --- a/core/dtn7/src/ipnd/neighbour_discovery.rs +++ b/core/dtn7/src/ipnd/neighbour_discovery.rs @@ -1,230 +1,27 @@ -use crate::cla::ConvergenceLayerAgent; -use crate::core::{DtnPeer, PeerType}; -use crate::ipnd::{beacon::Beacon, services::*}; -use crate::routing::RoutingNotifcation; -use crate::{peers_add, routing_notify, CLAS, CONFIG}; -use crate::{peers_touch, DTNCORE}; +use crate::dtnconfig::DiscoveryTransport; +use crate::ipnd::mdns_discovery::spawn_mdns_discovery; +use crate::ipnd::udp_discovery::spawn_udp_discovery; +use crate::CONFIG; use anyhow::Result; -use log::{debug, error, info, trace}; -use socket2::{Domain, Socket, Type}; -use std::collections::HashMap; -use std::io; -use std::net::SocketAddr; -use tokio::net::UdpSocket; -use tokio::time::interval; +use log::info; -async fn receiver(socket: UdpSocket) -> Result<(), io::Error> { - let mut buf: Vec = vec![0; 1024 * 64]; - let nodeid = CONFIG.lock().host_eid.clone(); - loop { - if let Ok((size, peer)) = socket.recv_from(&mut buf).await { - trace!("received {} bytes", size); - let deserialized: Beacon = match serde_cbor::from_slice(&buf[..size]) { - Ok(pkt) => pkt, - Err(e) => { - error!("Deserialization of beacon failed: {}", e); - continue; - } - }; - - // Creates a new peer from received beacon - let dtnpeer = DtnPeer::new( - deserialized.eid().clone(), - peer.ip().into(), - PeerType::Dynamic, - deserialized.beacon_period(), - deserialized.service_block().clas().clone(), - deserialized.service_block().convert_services(), - ); - if dtnpeer.eid == nodeid { - debug!("Received beacon from myself, ignoring"); - continue; - } - if peers_add(dtnpeer) { - info!( - "New peer discovered: {} @ {} (len={})", - deserialized.eid(), - peer, - size - ); - } else { - debug!( - "Beacon from known peer: {} @ {} (len={})", - deserialized.eid(), - peer, - size - ); - // TODO: check if any fields have changed and update not only timestamp - if let Err(e) = peers_touch(deserialized.eid().node().unwrap().as_ref()) { - error!("Failed to touch peer: {}", e); - } - } - trace!("{}", deserialized); - if let Err(err) = routing_notify(RoutingNotifcation::EncounteredPeer( - deserialized.eid().clone(), - )) - .await - { - info!("Error while encountered peer notification: {}", err); - } - } - } -} - -async fn announcer(socket: UdpSocket, _v6: bool) { - let mut task = interval(crate::CONFIG.lock().announcement_interval); - loop { - task.tick().await; - - // Start to build beacon announcement - let eid = CONFIG.lock().host_eid.clone(); - let beacon_period = if !crate::CONFIG.lock().enable_period { - None - } else { - Some(crate::CONFIG.lock().announcement_interval) - }; - let mut pkt = Beacon::with_config(eid, ServiceBlock::new(), beacon_period); - // Get all available clas - (*CLAS.lock()) - .iter() - .for_each(|cla| pkt.add_cla(cla.name(), &Some(cla.port()))); - // Get all available services - DTNCORE - .lock() - .service_list - .iter() - .for_each(|(tag, payload)| pkt.add_custom_service(*tag, payload.clone())); - - //let nodeid = format!("dtn://{}", (*DTNCORE.lock()).nodeid); - //let addr = "127.0.0.1:3003".parse().unwrap(); - - let mut destinations: HashMap = HashMap::new(); - CONFIG - .lock() - .discovery_destinations - .iter() - .for_each(|(key, value)| { - destinations.insert(key.clone().parse().unwrap(), *value); - }); - for (destination, bsn) in destinations { - (*CONFIG.lock()).update_beacon_sequence_number(&destination.to_string()); - pkt.set_beacon_sequence_number(bsn); - - if destination.ip().is_multicast() { - trace!( - "Sending beacon\n{}\nto multicast address {}", - pkt, - destination - ); - } else { - trace!( - "Sending beacon\n{}\nto unicast address {}", - pkt, - destination - ); - } - match socket - .send_to(&serde_cbor::to_vec(&pkt).unwrap(), destination) - .await - { - Ok(amt) => { - debug!("sent announcement beacon (len={}) to {}", amt, destination); - } - Err(err) => error!("Sending announcement failed: {}", err), - } - } - } -} +/// Spawns neighbor discovery based on configured transport pub async fn spawn_neighbour_discovery() -> Result<()> { - let v4 = CONFIG.lock().v4; - let v6 = CONFIG.lock().v6; - let port = CONFIG.lock().discovery_listen_port; - if v4 { - let addr: SocketAddr = format!("0.0.0.0:{}", port).parse()?; - let addr = addr.into(); - let socket = Socket::new(Domain::IPV4, Type::DGRAM, None)?; - socket.set_reuse_address(true)?; - socket.bind(&addr)?; - - //required for tokio - socket.set_nonblocking(true)?; + let transport = CONFIG.lock().discovery_transport; - // DEBUG: setup multicast on loopback to true - socket - .set_multicast_loop_v4(false) - .expect("error activating multicast loop v4"); - socket.set_broadcast(true)?; - for address in CONFIG.lock().discovery_destinations.keys() { - let addr: SocketAddr = address.parse().expect("Error parsing discovery address"); - if addr.is_ipv4() { - if addr.ip().is_multicast() { - socket - .join_multicast_v4( - &addr.ip().to_string().parse()?, - &std::net::Ipv4Addr::new(0, 0, 0, 0), - ) - .expect("error joining multicast v4 group"); - } - info!("Configured discovery destination: {}", addr.ip()); - } + match transport { + DiscoveryTransport::UdpMulticast => { + info!("Using UDP multicast for discovery"); + spawn_udp_discovery().await } - /* - socket - .join_multicast_v4(&"224.0.0.26".parse()?, &std::net::Ipv4Addr::new(0, 0, 0, 0)) - .expect("error joining multicast v4 group"); - */ - - let socket1 = UdpSocket::from_std(socket.try_clone()?.into())?; - let socket2 = UdpSocket::from_std(socket.try_clone()?.into())?; - - info!("Listening on {}", socket1.local_addr()?); - - tokio::spawn(receiver(socket1)); - - tokio::spawn(announcer(socket2, false)); - } - if v6 { - let addr: SocketAddr = format!("[::1]:{}", port).parse()?; - let addr = addr.into(); - let socket = Socket::new(Domain::IPV6, Type::DGRAM, None)?; - socket.set_reuse_address(true)?; - socket.set_only_v6(true)?; - socket.bind(&addr)?; - - //required for tokio - socket.set_nonblocking(true)?; - - // DEBUG: setup multicast on loopback to true - socket - .set_multicast_loop_v6(false) - .expect("error activating multicast loop v6"); - - socket.set_broadcast(true)?; - - for address in CONFIG.lock().discovery_destinations.keys() { - let addr: SocketAddr = address.parse().expect("Error while parsing IPv6 address"); - if addr.is_ipv6() { - if addr.ip().is_multicast() { - socket - .join_multicast_v6(&addr.ip().to_string().parse()?, 0) - .expect("Error joining multicast v6 group"); - } - info!("Configured discovery destination: {}", addr.ip()); - } + DiscoveryTransport::Mdns => { + info!("Using mDNS/DNS-SD for discovery"); + spawn_mdns_discovery().await + } + DiscoveryTransport::Both => { + info!("Using both UDP multicast and mDNS for discovery"); + spawn_udp_discovery().await?; + spawn_mdns_discovery().await } - /* - socket - .join_multicast_v6(&"FF02::300".parse()?, 0) - .expect("error joining multicast v6 group"); - */ - let socket1 = UdpSocket::from_std(socket.try_clone()?.into())?; - let socket2 = UdpSocket::from_std(socket.try_clone()?.into())?; - - info!("Listening on {}", socket1.local_addr()?); - - tokio::spawn(receiver(socket1)); - tokio::spawn(announcer(socket2, true)); } - - Ok(()) } diff --git a/core/dtn7/src/ipnd/udp_discovery.rs b/core/dtn7/src/ipnd/udp_discovery.rs new file mode 100644 index 000000000..33a4a9320 --- /dev/null +++ b/core/dtn7/src/ipnd/udp_discovery.rs @@ -0,0 +1,233 @@ +use crate::cla::ConvergenceLayerAgent; +use crate::core::{DtnPeer, PeerType}; +use crate::ipnd::{beacon::Beacon, services::*}; +use crate::routing::RoutingNotifcation; +use crate::{peers_add, peers_touch, routing_notify, CLAS, CONFIG, DTNCORE}; +use anyhow::Result; +use log::{debug, error, info, trace}; +use socket2::{Domain, Socket, Type}; +use std::collections::HashMap; +use std::io; +use std::net::SocketAddr; +use tokio::net::UdpSocket; +use tokio::time::interval; + +async fn udp_receiver(socket: UdpSocket) -> Result<(), io::Error> { + let mut buf: Vec = vec![0; 1024 * 64]; + let nodeid = CONFIG.lock().host_eid.clone(); + loop { + if let Ok((size, peer)) = socket.recv_from(&mut buf).await { + trace!("received {} bytes", size); + let deserialized: Beacon = match serde_cbor::from_slice(&buf[..size]) { + Ok(pkt) => pkt, + Err(e) => { + error!("Deserialization of beacon failed: {}", e); + continue; + } + }; + + // Creates a new peer from received beacon + let dtnpeer = DtnPeer::new( + deserialized.eid().clone(), + peer.ip().into(), + PeerType::Dynamic, + deserialized.beacon_period(), + deserialized.service_block().clas().clone(), + deserialized.service_block().convert_services(), + ); + if dtnpeer.eid == nodeid { + debug!("Received beacon from myself, ignoring"); + continue; + } + if peers_add(dtnpeer) { + info!( + "New peer discovered: {} @ {} (len={})", + deserialized.eid(), + peer, + size + ); + } else { + debug!( + "Beacon from known peer: {} @ {} (len={})", + deserialized.eid(), + peer, + size + ); + // TODO: check if any fields have changed and update not only timestamp + if let Err(e) = peers_touch(deserialized.eid().node().unwrap().as_ref()) { + error!("Failed to touch peer: {}", e); + } + } + trace!("{}", deserialized); + if let Err(err) = routing_notify(RoutingNotifcation::EncounteredPeer( + deserialized.eid().clone(), + )) + .await + { + info!("Error while encountered peer notification: {}", err); + } + } + } +} + +async fn udp_announcer(socket: UdpSocket, _v6: bool) { + let mut task = interval(crate::CONFIG.lock().announcement_interval); + loop { + task.tick().await; + + // Start to build beacon announcement + let eid = CONFIG.lock().host_eid.clone(); + let beacon_period = if !crate::CONFIG.lock().enable_period { + None + } else { + Some(crate::CONFIG.lock().announcement_interval) + }; + let mut pkt = Beacon::with_config(eid, ServiceBlock::new(), beacon_period); + // Get all available clas + (*CLAS.lock()) + .iter() + .for_each(|cla| pkt.add_cla(cla.name(), &Some(cla.port()))); + // Get all available services + DTNCORE + .lock() + .service_list + .iter() + .for_each(|(tag, payload)| pkt.add_custom_service(*tag, payload.clone())); + + //let nodeid = format!("dtn://{}", (*DTNCORE.lock()).nodeid); + //let addr = "127.0.0.1:3003".parse().unwrap(); + + let mut destinations: HashMap = HashMap::new(); + CONFIG + .lock() + .discovery_destinations + .iter() + .for_each(|(key, value)| { + destinations.insert(key.clone().parse().unwrap(), *value); + }); + for (destination, bsn) in destinations { + (*CONFIG.lock()).update_beacon_sequence_number(&destination.to_string()); + pkt.set_beacon_sequence_number(bsn); + + if destination.ip().is_multicast() { + trace!( + "Sending beacon\n{}\nto multicast address {}", + pkt, + destination + ); + } else { + trace!( + "Sending beacon\n{}\nto unicast address {}", + pkt, + destination + ); + } + match socket + .send_to(&serde_cbor::to_vec(&pkt).unwrap(), destination) + .await + { + Ok(amt) => { + debug!("sent announcement beacon (len={}) to {}", amt, destination); + } + Err(err) => error!("Sending announcement failed: {}", err), + } + } + } +} + +pub async fn spawn_udp_discovery() -> Result<()> { + let v4 = CONFIG.lock().v4; + let v6 = CONFIG.lock().v6; + let port = CONFIG.lock().discovery_listen_port; + + info!("Starting UDP multicast discovery"); + + if v4 { + let addr: SocketAddr = format!("0.0.0.0:{}", port).parse()?; + let addr = addr.into(); + let socket = Socket::new(Domain::IPV4, Type::DGRAM, None)?; + socket.set_reuse_address(true)?; + socket.bind(&addr)?; + + //required for tokio + socket.set_nonblocking(true)?; + + // DEBUG: setup multicast on loopback to true + socket + .set_multicast_loop_v4(false) + .expect("error activating multicast loop v4"); + socket.set_broadcast(true)?; + for address in CONFIG.lock().discovery_destinations.keys() { + let addr: SocketAddr = address.parse().expect("Error parsing discovery address"); + if addr.is_ipv4() { + if addr.ip().is_multicast() { + socket + .join_multicast_v4( + &addr.ip().to_string().parse()?, + &std::net::Ipv4Addr::new(0, 0, 0, 0), + ) + .expect("error joining multicast v4 group"); + } + info!("Configured discovery destination: {}", addr.ip()); + } + } + /* + socket + .join_multicast_v4(&"224.0.0.26".parse()?, &std::net::Ipv4Addr::new(0, 0, 0, 0)) + .expect("error joining multicast v4 group"); + */ + + let socket1 = UdpSocket::from_std(socket.try_clone()?.into())?; + let socket2 = UdpSocket::from_std(socket.try_clone()?.into())?; + + info!("Listening on {}", socket1.local_addr()?); + + tokio::spawn(udp_receiver(socket1)); + + tokio::spawn(udp_announcer(socket2, false)); + } + if v6 { + let addr: SocketAddr = format!("[::1]:{}", port).parse()?; + let addr = addr.into(); + let socket = Socket::new(Domain::IPV6, Type::DGRAM, None)?; + socket.set_reuse_address(true)?; + socket.set_only_v6(true)?; + socket.bind(&addr)?; + + //required for tokio + socket.set_nonblocking(true)?; + + // DEBUG: setup multicast on loopback to true + socket + .set_multicast_loop_v6(false) + .expect("error activating multicast loop v6"); + + socket.set_broadcast(true)?; + + for address in CONFIG.lock().discovery_destinations.keys() { + let addr: SocketAddr = address.parse().expect("Error while parsing IPv6 address"); + if addr.is_ipv6() { + if addr.ip().is_multicast() { + socket + .join_multicast_v6(&addr.ip().to_string().parse()?, 0) + .expect("Error joining multicast v6 group"); + } + info!("Configured discovery destination: {}", addr.ip()); + } + } + /* + socket + .join_multicast_v6(&"FF02::300".parse()?, 0) + .expect("error joining multicast v6 group"); + */ + let socket1 = UdpSocket::from_std(socket.try_clone()?.into())?; + let socket2 = UdpSocket::from_std(socket.try_clone()?.into())?; + + info!("Listening on {}", socket1.local_addr()?); + + tokio::spawn(udp_receiver(socket1)); + tokio::spawn(udp_announcer(socket2, true)); + } + + Ok(()) +} diff --git a/doc/mdns-discovery.md b/doc/mdns-discovery.md new file mode 100644 index 000000000..bec8c970e --- /dev/null +++ b/doc/mdns-discovery.md @@ -0,0 +1,111 @@ +# mDNS Discovery for dtn7-rs + +## Overview + +dtn7-rs supports **mDNS/DNS-SD** (`_dtn._udp.local.`) as an alternative discovery transport alongside UDP multicast IPND. This enables multiple DTN nodes to run on the same host without port conflicts and provides standard service discovery integration. + +## Configuration + +### Config File + +```toml +[discovery] +interval = "2s" +peer-timeout = "20s" + +# Transport options: +# - "udp-multicast" (default): IPND over UDP multicast +# - "mdns": mDNS/DNS-SD service discovery +# - "both": Run both transports simultaneously +transport = "mdns" +``` + +### Command Line + +```bash +# Enable mDNS discovery +dtnd --discovery-transport mdns + +# Use both UDP multicast and mDNS +dtnd --discovery-transport both + +# Default (UDP multicast only) +dtnd --discovery-transport udp-multicast +``` + +## Service Structure + +Each node registers an mDNS service: + +- **Service Type:** `_dtn._udp.local.` +- **Instance Name:** Node ID (e.g., `node1`) +- **Hostname:** `.local.` +- **Port:** 0 (CLAs use their own ports) +- **TXT Records:** + - `eid`: Endpoint ID (e.g., `dtn://node1/`) + - `cla0`, `cla1`, ...: Available CLAs with ports (e.g., `mtcp:16162`) + - `svc`: Custom services (same semantics as IPND) + +Example service: +``` +node1._dtn._udp.local + eid=dtn://node1/ + cla0=mtcp:16162 + cla1=tcp:4556 + svc63=f2f-app-v1.0 +``` + +## Discovery Process + +1. **Registration:** Node registers `_dtn._udp.local.` service on startup +2. **Browsing:** Node browses for services of type `_dtn._udp.local.` +3. **Resolution:** Extract peer EID, CLAs, and IP address from TXT records +4. **Peer Management:** Add discovered peers to global `PEERS` map +5. **Routing:** Notify routing agent via `RoutingNotification::EncounteredPeer` +6. **Refresh:** Re-register service at `interval` to update CLA changes + +## Comparison: UDP Multicast vs mDNS + +| Feature | UDP Multicast | mDNS | +|---------|--------------|------| +| Port conflicts | Yes (only one daemon/host) | No (system manages mDNS) | +| Same-host discovery | No | Yes | +| Standard tooling | Limited | Full (dns-sd, avahi-browse) | +| Name resolution | No | Yes (`.local` hostnames) | +| Legacy compatibility | IPND | New | + +## Platform Requirements + +- **macOS:** Works out of the box (Bonjour) +- **Linux:** Requires Avahi daemon +- **Windows:** Requires Bonjour Service + +## Implementation + +Transport selection happens in `spawn_neighbour_discovery()`: + +```rust +match CONFIG.lock().discovery_transport { + DiscoveryTransport::UdpMulticast => spawn_udp_discovery().await, + DiscoveryTransport::Mdns => spawn_mdns_discovery().await, + DiscoveryTransport::Both => { + spawn_udp_discovery().await?; + spawn_mdns_discovery().await + } +} +``` + +Dependencies: +```toml +mdns-sd = "0.11" +``` + +## Security Considerations + +mDNS discovery inherits IPND security properties: + +- No peer authentication +- No service encryption +- Local network visibility + +For F2F networks, implement application-level security via custom TXT records and bundle payload encryption. diff --git a/examples/dtn7.toml.example b/examples/dtn7.toml.example index 76ef6618e..f1268a1c4 100644 --- a/examples/dtn7.toml.example +++ b/examples/dtn7.toml.example @@ -56,6 +56,12 @@ interval = "2s" peer-timeout = "20s" port = 3003 +# Discovery transport mechanism: +# - "udp-multicast" (default): Legacy IPND over UDP multicast +# - "mdns": Modern mDNS/DNS-SD service discovery (no port conflicts!) +# - "both": Run both UDP multicast and mDNS simultaneously +transport = "udp-multicast" + [convergencylayers] global.tcp.refuse-existing-bundles = true From bc894bdcaa72d012344f3d339d3a386067086ade Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lars=20Baumg=C3=A4rtner?= <1264131+gh0st42@users.noreply.github.com> Date: Tue, 26 May 2026 20:39:16 +0200 Subject: [PATCH 2/2] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- core/dtn7/src/ipnd/udp_discovery.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/dtn7/src/ipnd/udp_discovery.rs b/core/dtn7/src/ipnd/udp_discovery.rs index 33a4a9320..05a370016 100644 --- a/core/dtn7/src/ipnd/udp_discovery.rs +++ b/core/dtn7/src/ipnd/udp_discovery.rs @@ -152,7 +152,7 @@ pub async fn spawn_udp_discovery() -> Result<()> { //required for tokio socket.set_nonblocking(true)?; - // DEBUG: setup multicast on loopback to true + // DEBUG: disable multicast loopback on IPv4 socket .set_multicast_loop_v4(false) .expect("error activating multicast loop v4");