From a331c21c56757aaed070d3a1bced7b7d65ad7110 Mon Sep 17 00:00:00 2001 From: Tien Do Nam Date: Fri, 31 Jul 2026 14:14:05 +0200 Subject: [PATCH] feat: add discovery to core --- packages/core/Cargo.toml | 3 +- packages/core/src/discovery/mod.rs | 454 +++++++++++++++++++++++++++ packages/core/src/discovery/store.rs | 373 ++++++++++++++++++++++ packages/core/src/http/client/mod.rs | 20 ++ packages/core/src/http/client/v2.rs | 15 +- packages/core/src/http/client/v3.rs | 15 +- packages/core/src/lib.rs | 2 + packages/core/tests/discovery.rs | 397 +++++++++++++++++++++++ 8 files changed, 1270 insertions(+), 9 deletions(-) create mode 100644 packages/core/src/discovery/mod.rs create mode 100644 packages/core/src/discovery/store.rs create mode 100644 packages/core/tests/discovery.rs diff --git a/packages/core/Cargo.toml b/packages/core/Cargo.toml index 2da5ac2d..b2169533 100644 --- a/packages/core/Cargo.toml +++ b/packages/core/Cargo.toml @@ -43,11 +43,12 @@ x509-parser = { version = "0.18.1", features = ["verify"], optional = true } [features] default = [] crypto = ["ed25519-dalek", "rcgen", "rsa", "sha2", "tokio-util"] +discovery = ["http", "multicast"] http = ["crypto", "form_urlencoded", "http-body-util", "hyper", "hyper-util", "pem", "percent-encoding", "reqwest", "rustls", "socket2", "tokio-rustls", "tokio-util", "x509-parser"] multicast = ["if-addrs", "socket2", "tokio-util"] webrtc-signaling = ["tokio-tungstenite"] webrtc = ["crypto", "flate2", "dep:webrtc", "webrtc-signaling", "x509-parser"] -full = ["crypto", "http", "multicast", "webrtc"] +full = ["crypto", "discovery", "http", "multicast", "webrtc"] # RSA key generation is bignum-heavy and takes ~10x longer unoptimized; # keep the crypto crates optimized in dev so tests stay fast. diff --git a/packages/core/src/discovery/mod.rs b/packages/core/src/discovery/mod.rs new file mode 100644 index 00000000..7a4022a5 --- /dev/null +++ b/packages/core/src/discovery/mod.rs @@ -0,0 +1,454 @@ +mod store; + +pub use store::{ + DeviceChannel, DeviceLog, DiscoveredDevice, DiscoveredDeviceWithLogs, HttpChannel, +}; + +use crate::http::client::{ClientError, LsHttpClientV2}; +use crate::http::dto::ProtocolType; +use crate::http::dto_v2::{RegisterDtoV2, RegisterResponseDtoV2}; +use crate::model::discovery::{MulticastMessageV2, ProtocolTypeV2}; +use crate::multicast::{ + self, InterfaceFilter, MulticastConfig, MulticastDevice, MulticastEvent, MulticastHandle, +}; +use futures_util::StreamExt; +use std::collections::HashSet; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; +use store::DeviceStore; +use tokio::sync::{mpsc, oneshot}; + +/// The default timeout of the register requests sent by discovery, matching +/// the Flutter app's default discovery timeout. Discovery only talks to LAN +/// peers, which answer quickly or not at all. +pub const DEFAULT_DISCOVERY_TIMEOUT: Duration = Duration::from_millis(500); + +/// Capacity of the internal multicast event channel. +const MULTICAST_CHANNEL_SIZE: usize = 64; + +/// How many hosts a subnet scan probes at once, matching the Flutter app's +/// legacy HTTP discovery. +const SCAN_CONCURRENCY: usize = 50; + +/// This device's TLS identity, sent as client certificate with every register +/// request (client certificates are mandatory in HTTPS mode). +/// +/// Its fingerprint is the one carried in [`MulticastDevice::fingerprint`]. +#[derive(Clone)] +pub struct DeviceIdentity { + /// PEM-encoded certificate. + pub cert_pem: String, + + /// PEM-encoded private key. + pub private_key_pem: String, +} + +/// Configuration of the discovery. +pub struct DiscoveryConfig { + /// The multicast group to join, usually [`multicast::DEFAULT_MULTICAST_GROUP`]. + pub group: Ipv4Addr, + + /// The IPv6 multicast group to additionally join, usually + /// [`multicast::DEFAULT_MULTICAST_GROUP_V6`]. `None` disables IPv6 discovery. + pub group_v6: Option, + + /// The multicast port to bind, usually [`multicast::DEFAULT_PORT`]. + pub port: u16, + + /// Restricts the network interfaces that are used. + pub interface_filter: InterfaceFilter, + + /// The device information announced to the network and sent in register + /// requests. + pub device: MulticastDevice, + + /// The TLS identity used for the register requests. + pub identity: DeviceIdentity, + + /// Timeout of each register request sent by discovery, usually + /// [`DEFAULT_DISCOVERY_TIMEOUT`]. Bounds how long an unresponsive host + /// stalls a subnet scan, so keep it short. + pub timeout: Duration, + + /// Channel on which discovery events are emitted. `None` when the + /// application only polls [`DiscoveryHandle::devices`]. + pub event_tx: Option>, +} + +/// An event emitted by the discovery. Every event is also logged in +/// [`DiscoveredDeviceWithLogs::logs`]; the accumulated state is read from +/// [`DiscoveryHandle::devices`]. +#[derive(Clone, Debug)] +pub enum DiscoveryEvent { + /// A device was confirmed over one of its channels for the first time + /// in this run. + Discovered { + /// The device as it was confirmed, i.e. carrying only the channel + /// the confirmation happened on. + device: DiscoveredDevice, + }, + + /// An already known device was confirmed again: it re-announced itself + /// or was re-discovered, at a known or a new address. + Updated { + /// The device as it was confirmed, like in + /// [`DiscoveryEvent::Discovered`]. + device: DiscoveredDevice, + }, +} + +struct DiscoveryState { + device: MulticastDevice, + identity: DeviceIdentity, + timeout: Duration, + store: DeviceStore, + event_tx: Option>, + + /// The interface addresses a subnet scan is currently running for. + scanning: std::sync::Mutex>, +} + +impl DiscoveryState { + fn register_dto(&self) -> RegisterDtoV2 { + RegisterDtoV2 { + alias: self.device.alias.clone(), + version: self.device.version.clone(), + device_model: self.device.device_model.clone(), + device_type: self.device.device_type.clone(), + fingerprint: self.device.fingerprint.clone(), + port: self.device.port, + protocol: self.device.protocol, + download: self.device.download, + } + } + + /// A client accepting any valid certificate, for peers whose fingerprint + /// is not known before they answer. The fingerprint is then read off the + /// handshake and pinned by connections that transfer data. + fn unpinned_client(&self) -> Result { + LsHttpClientV2::try_new( + &self.identity.private_key_pem, + &self.identity.cert_pem, + None, + Some(self.timeout), + ) + } + + /// Registers with `host:port` and, when a device answers, puts it into + /// the store. Returns the device's stored state after the merge, or + /// `None` when the answer carried this device's own fingerprint. + async fn probe( + &self, + client: &LsHttpClientV2, + host: &str, + port: u16, + protocol: ProtocolTypeV2, + ) -> Result, ClientError> { + let response = client + .register(client_protocol(protocol), host, port, self.register_dto()) + .await?; + + // In HTTPS mode the certificate is the peer's identity; the + // fingerprint claimed in the body only counts without encryption. + let fingerprint = match protocol { + ProtocolTypeV2::Https => response + .cert_fingerprint + .clone() + .ok_or_else(|| anyhow::anyhow!("HTTPS response carried no peer certificate"))?, + ProtocolTypeV2::Http => response.body.fingerprint.clone(), + }; + if fingerprint == self.device.fingerprint { + return Ok(None); + } + + let device = confirmed_device(response.body, host.to_string(), port, protocol, fingerprint); + let (_, merged) = self.found(device).await; + Ok(Some(merged)) + } + + /// Puts a device into the store, logging the confirmation on it, and + /// emits the resulting event. Returns whether the device is new, and + /// its stored state after the merge. + async fn found(&self, device: DiscoveredDevice) -> (bool, DiscoveredDeviceWithLogs) { + let (event, merged) = self.store.upsert(device, SystemTime::now()); + let is_new = matches!(event, DiscoveryEvent::Discovered { .. }); + if let Some(event_tx) = &self.event_tx { + let _ = event_tx.send(event).await; + } + (is_new, merged) + } +} + +/// A handle to a running discovery: the store of discovered devices and the +/// application-initiated operations. +pub struct DiscoveryHandle { + multicast: MulticastHandle, + state: Arc, +} + +impl DiscoveryHandle { + /// Announces this device to the network, which makes every other LocalSend + /// device on it register with this device over HTTP. + /// + /// Devices registering in response arrive at the application as server + /// events, not here: discovery only learns about peers that announce + /// themselves. Feed them back via [`DiscoveryHandle::add_device`]. + /// + /// Returns once the whole announcement burst has been sent, which takes a + /// few seconds, or immediately once discovery has been stopped. + pub async fn announce(&self) { + self.multicast.announce().await; + } + + /// Discovers a device at a known address, e.g. a favorite or a peer that + /// multicast does not reach, by sending it a register request. + /// + /// On success the device is put into the store (and emitted, as + /// `Discovered` or `Updated`), and its full stored state — all known + /// channels and logs — is returned. + /// Returns `None` when the address answered with this device's own + /// fingerprint, i.e. the device discovered itself. + pub async fn discover( + &self, + host: &str, + port: u16, + protocol: ProtocolTypeV2, + ) -> Result, ClientError> { + let client = self.state.unpinned_client()?; + self.state.probe(&client, host, port, protocol).await + } + + /// Scans the `/24` subnet of the local interface address `interface_ip` + /// by sending every other host a register request, + /// for networks that do not carry multicast. + /// + /// At most one scan runs per interface: a call for an address that is + /// still being scanned returns an empty list immediately. + pub async fn scan_subnet( + &self, + interface_ip: Ipv4Addr, + port: u16, + protocol: ProtocolTypeV2, + ) -> Result, ClientError> { + if !self.state.scanning.lock().unwrap().insert(interface_ip) { + return Ok(Vec::new()); + } + let _guard = ScanGuard { + state: &self.state, + interface_ip, + }; + + let client = self.state.unpinned_client()?; + let base = interface_ip.octets(); + + let state = &self.state; + let client = &client; + let found = futures_util::stream::iter( + (0..=255u8) + .map(|host| Ipv4Addr::new(base[0], base[1], base[2], host)) + .filter(|ip| *ip != interface_ip), + ) + .map(|ip| async move { + state + .probe(client, &ip.to_string(), port, protocol) + .await + .ok() + .flatten() + }) + .buffer_unordered(SCAN_CONCURRENCY) + .filter_map(std::future::ready) + .collect() + .await; + + Ok(found) + } + + /// Puts a device confirmed outside of discovery into the store, e.g. one + /// that answered an announcement by registering with this device's HTTP + /// server. The confirmation is emitted as `Discovered` or `Updated`; + /// returns `true` when the device is new. + pub async fn add_device(&self, device: DiscoveredDevice) -> bool { + self.state.found(device).await.0 + } + + /// All discovered devices in discovery order. + pub fn devices(&self) -> Vec { + self.state.store.devices() + } + + pub fn device_by_fingerprint(&self, fingerprint: &str) -> Option { + self.state.store.by_fingerprint(fingerprint) + } + + /// Waits until discovery has terminated and the multicast sockets have + /// been closed, so that the port can be bound again. + /// Must be called after requesting a stop via the stop channel. + pub async fn wait_stopped(&self) { + self.multicast.wait_stopped().await; + } +} + +/// Releases the interface of a finished subnet scan, also when the scan is +/// cancelled by dropping its future. +struct ScanGuard<'a> { + state: &'a DiscoveryState, + interface_ip: Ipv4Addr, +} + +impl Drop for ScanGuard<'_> { + fn drop(&mut self) { + self.state + .scanning + .lock() + .unwrap() + .remove(&self.interface_ip); + } +} + +/// Binds the multicast sockets and starts answering announcements of other +/// devices. Nothing is announced until [`DiscoveryHandle::announce`] is called. +/// +/// Fails when no network interface could be used, e.g. because the port is +/// already bound by another process or because there is no network at all. +pub async fn start( + config: DiscoveryConfig, + stop_rx: oneshot::Receiver<()>, +) -> anyhow::Result { + let (multicast_tx, mut multicast_rx) = mpsc::channel(MULTICAST_CHANNEL_SIZE); + + let multicast = multicast::start( + MulticastConfig { + group: config.group, + group_v6: config.group_v6, + port: config.port, + interface_filter: config.interface_filter, + device: config.device.clone(), + event_tx: multicast_tx, + }, + stop_rx, + ) + .await?; + + let state = Arc::new(DiscoveryState { + device: config.device, + identity: config.identity, + timeout: config.timeout, + store: DeviceStore::new(), + event_tx: config.event_tx, + scanning: std::sync::Mutex::new(HashSet::new()), + }); + + // Ends once discovery is stopped: the multicast side then drops its + // sender and the channel closes. + tokio::spawn({ + let state = state.clone(); + async move { + while let Some(event) = multicast_rx.recv().await { + let MulticastEvent::Discovered { + ip, + scope_id, + message, + } = event; + + // The register request may take a while (up to the timeout), + // so announcements are answered concurrently. + tokio::spawn(answer_announcement(state.clone(), ip, scope_id, message)); + } + } + }); + + Ok(DiscoveryHandle { multicast, state }) +} + +/// Answers an announcement with a register request, as the protocol requires. +/// The device enters the store only once that request succeeded, so that +/// everything in the store is known to be reachable. +async fn answer_announcement( + state: Arc, + ip: IpAddr, + scope_id: Option, + message: MulticastMessageV2, +) { + let host = match scope_id { + Some(scope_id) => format!("{ip}%{scope_id}"), + None => ip.to_string(), + }; + + // Pin the claimed fingerprint, so nothing is sent to a device that does + // not hold the matching certificate. + let expected_fingerprint = match message.protocol { + ProtocolTypeV2::Https => Some(message.fingerprint.clone()), + ProtocolTypeV2::Http => None, + }; + let client = match LsHttpClientV2::try_new( + &state.identity.private_key_pem, + &state.identity.cert_pem, + expected_fingerprint, + Some(state.timeout), + ) { + Ok(client) => client, + Err(err) => { + tracing::error!("Could not create the client to answer {host}: {err:#}"); + return; + } + }; + + let result = client + .register( + client_protocol(message.protocol), + &host, + message.port, + state.register_dto(), + ) + .await; + + match result { + Ok(response) => { + // The pinned certificate identifies the peer, so the fingerprint + // is taken from the announcement, not from the response body. + let device = confirmed_device( + response.body, + host, + message.port, + message.protocol, + message.fingerprint, + ); + state.found(device).await; + } + Err(err) => { + tracing::debug!("Could not register with announcing device {host}: {err:#}"); + } + } +} + +/// Builds the stored device from the register response of a peer confirmed +/// over HTTP. +fn confirmed_device( + response: RegisterResponseDtoV2, + host: String, + port: u16, + protocol: ProtocolTypeV2, + fingerprint: String, +) -> DiscoveredDevice { + DiscoveredDevice { + alias: response.alias, + version: response.version, + device_model: response.device_model, + device_type: response.device_type, + fingerprint, + channels: vec![DeviceChannel::Http(HttpChannel { + host, + port, + protocol, + })], + download: response.download, + } +} + +fn client_protocol(protocol: ProtocolTypeV2) -> ProtocolType { + match protocol { + ProtocolTypeV2::Http => ProtocolType::Http, + ProtocolTypeV2::Https => ProtocolType::Https, + } +} diff --git a/packages/core/src/discovery/store.rs b/packages/core/src/discovery/store.rs new file mode 100644 index 00000000..7a6bba79 --- /dev/null +++ b/packages/core/src/discovery/store.rs @@ -0,0 +1,373 @@ +//! The in-memory store of discovered devices. + +use super::DiscoveryEvent; +use crate::model::discovery::{DeviceType, ProtocolTypeV2}; +use std::sync::Mutex; +use std::time::SystemTime; + +/// How many log entries a device keeps. Every confirmation is logged and +/// chatty peers re-announce for the whole run, so the oldest entries are +/// dropped beyond this. +const MAX_LOGS: usize = 100; + +/// A device discovered on the network, confirmed over one of its channels. +#[derive(Clone, Debug)] +pub struct DiscoveredDevice { + /// The display name of the device. + pub alias: String, + + /// Protocol version (major.minor) implemented by the device. + pub version: String, + + /// Device model (e.g. "Samsung", "Windows"). + pub device_model: Option, + + /// Device type category. + pub device_type: Option, + + /// Fingerprint identifying the device; devices are deduplicated by it. + /// In HTTPS mode the SHA-256 hash of the certificate, otherwise a random string. + pub fingerprint: String, + + /// The channels the device is reachable on, in discovery order. + /// A transfer can jump to another channel when one fails. + pub channels: Vec, + + /// Whether the device's download API is active. + pub download: bool, +} + +/// A [`DiscoveredDevice`] as kept in the store, together with the history of +/// events that affected it. +#[derive(Clone, Debug)] +pub struct DiscoveredDeviceWithLogs { + pub device: DiscoveredDevice, + + /// The events that affected this device, oldest first, at most + /// [`MAX_LOGS`]. Every confirmation is logged, so the last entry is when + /// the device was last seen. + pub logs: Vec, +} + +/// A [`DiscoveryEvent`] that affected a device, with the time it happened. +#[derive(Clone, Debug)] +pub struct DeviceLog { + pub timestamp: SystemTime, + pub event: DiscoveryEvent, +} + +impl DiscoveredDevice { + /// The device's HTTP channels — the addresses it is reachable on — in + /// discovery order. A multi-homed device has one per address it was + /// discovered on. + pub fn http_channels(&self) -> impl Iterator { + // With more channel kinds this becomes a `filter_map`. + self.channels.iter().map(|channel| match channel { + DeviceChannel::Http(http) => http, + }) + } + + /// The device's first HTTP channel, when it has one. + pub fn http(&self) -> Option<&HttpChannel> { + self.http_channels().next() + } +} + +/// A channel a device is reachable on. +/// +/// Only HTTP exists so far; other transports (e.g. WebRTC, Bluetooth) will +/// become further variants. +#[derive(Clone, Debug)] +pub enum DeviceChannel { + /// The device's HTTP server (protocol v2), reachable at one address. + Http(HttpChannel), +} + +impl DeviceChannel { + /// Whether two channels address the same endpoint, so that a + /// re-confirmation updates the known channel instead of adding one. + fn same_endpoint(&self, other: &DeviceChannel) -> bool { + match (self, other) { + (DeviceChannel::Http(own), DeviceChannel::Http(other)) => own.host == other.host, + } + } +} + +/// The address of a device's HTTP server. +#[derive(Clone, Debug)] +pub struct HttpChannel { + /// The host to dial: an IP address, or the scoped form `fe80::1%3` for + /// link-local IPv6 (the HTTP client accepts both). + pub host: String, + + /// The port of the HTTP server. + pub port: u16, + + /// Whether the HTTP server uses TLS. + pub protocol: ProtocolTypeV2, +} + +/// All devices discovered in this run, identified by fingerprint, in +/// discovery order. +pub(super) struct DeviceStore { + devices: Mutex>, +} + +impl DeviceStore { + pub(super) fn new() -> Self { + Self { + devices: Mutex::new(Vec::new()), + } + } + + /// Adds or updates a device, logging the confirmation on it. Returns the + /// event describing what happened — [`DiscoveryEvent::Discovered`] for a + /// new device, [`DiscoveryEvent::Updated`] for a known one — and the + /// stored state after the merge. + /// + /// Channels are merged by endpoint: a rediscovery over a known address + /// updates its channel in place, an unknown address adds one. + pub(super) fn upsert( + &self, + device: DiscoveredDevice, + timestamp: SystemTime, + ) -> (DiscoveryEvent, DiscoveredDeviceWithLogs) { + let mut devices = self.devices.lock().unwrap(); + match devices + .iter_mut() + .find(|known| known.device.fingerprint == device.fingerprint) + { + Some(known) => { + let event = DiscoveryEvent::Updated { + device: device.clone(), + }; + + let mut channels = std::mem::take(&mut known.device.channels); + for channel in device.channels { + match channels.iter_mut().find(|c| c.same_endpoint(&channel)) { + Some(known) => *known = channel, + None => channels.push(channel), + } + } + + known.logs.push(DeviceLog { + timestamp, + event: event.clone(), + }); + if known.logs.len() > MAX_LOGS { + let excess = known.logs.len() - MAX_LOGS; + known.logs.drain(..excess); + } + + known.device = DiscoveredDevice { channels, ..device }; + (event, known.clone()) + } + None => { + let event = DiscoveryEvent::Discovered { + device: device.clone(), + }; + let known = DiscoveredDeviceWithLogs { + device, + logs: vec![DeviceLog { + timestamp, + event: event.clone(), + }], + }; + devices.push(known.clone()); + (event, known) + } + } + } + + /// All discovered devices in discovery order. + pub(super) fn devices(&self) -> Vec { + self.devices.lock().unwrap().clone() + } + + pub(super) fn by_fingerprint(&self, fingerprint: &str) -> Option { + self.devices + .lock() + .unwrap() + .iter() + .find(|known| known.device.fingerprint == fingerprint) + .cloned() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn device(fingerprint: &str, host: &str) -> DiscoveredDevice { + DiscoveredDevice { + alias: format!("Alias of {fingerprint}"), + version: "2.1".to_string(), + device_model: None, + device_type: Some(DeviceType::Desktop), + fingerprint: fingerprint.to_string(), + channels: vec![DeviceChannel::Http(HttpChannel { + host: host.to_string(), + port: 53317, + protocol: ProtocolTypeV2::Https, + })], + download: false, + } + } + + /// The marker telling log entries apart: the host the logged snapshot + /// was confirmed on. + fn log_marker(log: &DeviceLog) -> &str { + let device = match &log.event { + DiscoveryEvent::Discovered { device } | DiscoveryEvent::Updated { device } => device, + }; + device.http().unwrap().host.as_str() + } + + #[test] + fn test_upsert_reports_only_the_first_confirmation_as_discovered() { + let store = DeviceStore::new(); + let now = SystemTime::now(); + + let (event, _) = store.upsert(device("a", "192.168.0.10"), now); + assert!(matches!(event, DiscoveryEvent::Discovered { .. })); + let (event, _) = store.upsert(device("b", "192.168.0.11"), now); + assert!(matches!(event, DiscoveryEvent::Discovered { .. })); + let (event, _) = store.upsert(device("a", "10.0.0.10"), now); + assert!(matches!(event, DiscoveryEvent::Updated { .. })); + } + + #[test] + fn test_upsert_logs_every_confirmation() { + let store = DeviceStore::new(); + + store.upsert(device("a", "192.168.0.10"), SystemTime::now()); + store.upsert(device("a", "fe80::1%3"), SystemTime::now()); + + let known = store.by_fingerprint("a").unwrap(); + assert_eq!(known.logs.len(), 2, "every confirmation must be logged"); + assert!(matches!( + known.logs[0].event, + DiscoveryEvent::Discovered { .. } + )); + assert!(matches!( + known.logs[1].event, + DiscoveryEvent::Updated { .. } + )); + assert_eq!(log_marker(&known.logs[1]), "fe80::1%3"); + } + + #[test] + fn test_oldest_logs_are_dropped_beyond_the_cap() { + let store = DeviceStore::new(); + + for i in 0..MAX_LOGS + 5 { + store.upsert(device("a", &i.to_string()), SystemTime::now()); + } + + let logs = store.by_fingerprint("a").unwrap().logs; + assert_eq!(logs.len(), MAX_LOGS); + assert_eq!(log_marker(&logs[0]), "5", "the oldest entries must go"); + assert_eq!( + log_marker(logs.last().unwrap()), + &(MAX_LOGS + 4).to_string() + ); + } + + #[test] + fn test_upsert_collects_channels_of_a_multi_homed_device() { + let store = DeviceStore::new(); + + store.upsert(device("a", "192.168.0.10"), SystemTime::now()); + store.upsert(device("a", "fe80::1%3"), SystemTime::now()); + + let devices = store.devices(); + assert_eq!(devices.len(), 1); + let hosts: Vec<&str> = devices[0] + .device + .http_channels() + .map(|http| http.host.as_str()) + .collect(); + assert_eq!( + hosts, + ["192.168.0.10", "fe80::1%3"], + "every address the device was confirmed on must be kept" + ); + } + + #[test] + fn test_upsert_updates_channel_of_known_endpoint_in_place() { + let store = DeviceStore::new(); + + store.upsert(device("a", "192.168.0.10"), SystemTime::now()); + store.upsert(device("a", "fe80::1%3"), SystemTime::now()); + + let mut update = device("a", "192.168.0.10"); + match &mut update.channels[0] { + DeviceChannel::Http(http) => http.port = 54000, + } + store.upsert(update, SystemTime::now()); + + let known = store.by_fingerprint("a").unwrap(); + let channels: Vec<(&str, u16)> = known + .device + .http_channels() + .map(|http| (http.host.as_str(), http.port)) + .collect(); + assert_eq!( + channels, + [("192.168.0.10", 54000), ("fe80::1%3", 53317)], + "a known address must be updated in place, not duplicated" + ); + } + + #[test] + fn test_upsert_keeps_channels_missing_from_the_update() { + let store = DeviceStore::new(); + + store.upsert(device("a", "192.168.0.10"), SystemTime::now()); + + let mut update = device("a", "10.0.0.10"); + update.channels.clear(); + store.upsert(update, SystemTime::now()); + + assert_eq!( + store + .by_fingerprint("a") + .unwrap() + .device + .http() + .unwrap() + .host, + "192.168.0.10", + "an update without an HTTP channel must not drop the known one" + ); + } + + #[test] + fn test_devices_keep_discovery_order() { + let store = DeviceStore::new(); + + store.upsert(device("a", "192.168.0.10"), SystemTime::now()); + store.upsert(device("b", "192.168.0.11"), SystemTime::now()); + store.upsert(device("a", "192.168.0.12"), SystemTime::now()); + + let fingerprints: Vec = store + .devices() + .into_iter() + .map(|known| known.device.fingerprint) + .collect(); + assert_eq!(fingerprints, ["a", "b"]); + + assert_eq!( + store + .by_fingerprint("b") + .unwrap() + .device + .http() + .unwrap() + .host, + "192.168.0.11" + ); + assert!(store.by_fingerprint("c").is_none()); + } +} diff --git a/packages/core/src/http/client/mod.rs b/packages/core/src/http/client/mod.rs index 407b27b4..9a43b7a1 100644 --- a/packages/core/src/http/client/mod.rs +++ b/packages/core/src/http/client/mod.rs @@ -94,6 +94,7 @@ impl LsHttpClient { let result = client.register(protocol, ip, port, payload.into()).await?; Ok(ResultWithPublicKey { public_key: result.public_key, + cert_fingerprint: result.cert_fingerprint, body: result.body.into(), }) } @@ -279,6 +280,20 @@ pub(super) fn verify_cert_from_res( Ok(public_key) } +/// The SHA-256 fingerprint (uppercase hex) of the peer certificate the +/// response was received over. This — not any fingerprint claimed in the +/// body — is the peer's identity in HTTPS mode. +pub(super) fn cert_fingerprint_from_res(response: &Response) -> anyhow::Result { + let tls_info_ext = response + .extensions() + .get::() + .ok_or_else(|| anyhow::anyhow!("TLS info not found"))?; + let cert = tls_info_ext + .peer_certificate() + .ok_or_else(|| anyhow::anyhow!("Certificate not found"))?; + Ok(crypto::cert::fingerprint_from_cert_der(cert)) +} + #[derive(Serialize, Deserialize)] struct ErrorResponse { message: String, @@ -290,6 +305,11 @@ pub struct ResultWithPublicKey { /// Only available in HTTPS mode. pub public_key: Option, + /// The SHA-256 fingerprint (uppercase hex) of the peer certificate. + /// Only available in HTTPS mode, where it is the peer's identity and + /// overrules any fingerprint claimed in the body. + pub cert_fingerprint: Option, + /// The response body. pub body: T, } diff --git a/packages/core/src/http/client/v2.rs b/packages/core/src/http/client/v2.rs index a885708e..ea7b976b 100644 --- a/packages/core/src/http/client/v2.rs +++ b/packages/core/src/http/client/v2.rs @@ -96,14 +96,21 @@ impl LsHttpClientV2 { return res.into_error().await; } - let public_key = match protocol { - ProtocolType::Https => Some(super::verify_cert_from_res(&res, None)?), - _ => None, + let (public_key, cert_fingerprint) = match protocol { + ProtocolType::Https => ( + Some(super::verify_cert_from_res(&res, None)?), + Some(super::cert_fingerprint_from_res(&res)?), + ), + _ => (None, None), }; let body = res.json::().await?; - Ok(ResultWithPublicKey { public_key, body }) + Ok(ResultWithPublicKey { + public_key, + cert_fingerprint, + body, + }) } /// Prepares a file upload session with the receiver. diff --git a/packages/core/src/http/client/v3.rs b/packages/core/src/http/client/v3.rs index 86c59939..c03f1fbe 100644 --- a/packages/core/src/http/client/v3.rs +++ b/packages/core/src/http/client/v3.rs @@ -123,14 +123,21 @@ impl LsHttpClientV3 { .send() .await?; - let public_key = match protocol { - ProtocolType::Https => Some(super::verify_cert_from_res(&res, None)?), - _ => None, + let (public_key, cert_fingerprint) = match protocol { + ProtocolType::Https => ( + Some(super::verify_cert_from_res(&res, None)?), + Some(super::cert_fingerprint_from_res(&res)?), + ), + _ => (None, None), }; let body = res.json::().await?; - Ok(ResultWithPublicKey { public_key, body }) + Ok(ResultWithPublicKey { + public_key, + cert_fingerprint, + body, + }) } /// `cancel` is a cancellation token; cancelling it aborts the request with diff --git a/packages/core/src/lib.rs b/packages/core/src/lib.rs index e7996cb6..ae72b099 100644 --- a/packages/core/src/lib.rs +++ b/packages/core/src/lib.rs @@ -1,5 +1,7 @@ #[cfg(feature = "crypto")] pub mod crypto; +#[cfg(feature = "discovery")] +pub mod discovery; #[cfg(feature = "http")] pub mod http; pub mod model; diff --git a/packages/core/tests/discovery.rs b/packages/core/tests/discovery.rs new file mode 100644 index 00000000..b443e24b --- /dev/null +++ b/packages/core/tests/discovery.rs @@ -0,0 +1,397 @@ +#![cfg(feature = "discovery")] + +//! End-to-end discovery tests: real HTTP servers confirm the register +//! requests. Most tests use plain HTTP; the client side of TLS pinning is +//! covered by `v2_tls_pinning.rs`. +//! +//! Binding multicast sockets and delivering multicast traffic depends on the +//! machine, so these tests skip themselves instead of failing when the +//! environment does not cooperate. + +use localsend::crypto::cert::generate_self_signed; +use localsend::discovery::{ + self, DeviceIdentity, DiscoveryConfig, DiscoveryEvent, DiscoveryHandle, +}; +use localsend::http::server::{start_with_port, ServerConfigV2, TlsConfig}; +use localsend::http::state::ClientInfo; +use localsend::model::discovery::{DeviceType, ProtocolTypeV2, PROTOCOL_VERSION_V2}; +use localsend::multicast::MulticastDevice; +use std::net::{Ipv4Addr, Ipv6Addr}; +use std::sync::atomic::{AtomicU16, Ordering}; +use std::time::Duration; +use tokio::sync::{mpsc, oneshot}; + +/// Like the group the protocol uses, but distinct from it (and from the one in +/// `multicast.rs`) so that unrelated instances stay out. +const TEST_GROUP: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 169); + +/// See [TEST_GROUP]. +const TEST_GROUP_V6: Ipv6Addr = Ipv6Addr::new(0xff12, 0, 0, 0, 0, 0, 0xfd3a, 0xe422); + +const RECEIVE_TIMEOUT: Duration = Duration::from_secs(5); + +/// Multicast ports are not reused between tests: a lingering membership of a +/// stopped instance would leak messages into the next test. +static NEXT_MULTICAST_PORT: AtomicU16 = AtomicU16::new(55317); + +/// Returns a free port for an HTTP server. +/// +/// A counter is used instead of binding to port 0 because the OS may hand out +/// the same just-freed ephemeral port to multiple tests running in parallel. +fn free_port() -> u16 { + static PORT_COUNTER: AtomicU16 = AtomicU16::new(41551); + + loop { + let port = PORT_COUNTER.fetch_add(1, Ordering::SeqCst); + if std::net::TcpListener::bind(("127.0.0.1", port)).is_ok() { + return port; + } + } +} + +/// Starts a v2 server on `port` that only exists to answer register requests +/// with the given identity, over TLS when `tls` is given. +async fn start_register_server( + port: u16, + alias: &str, + fingerprint: &str, + tls: Option, +) -> oneshot::Sender<()> { + let _ = tracing_subscriber::fmt().with_test_writer().try_init(); + + // The receiver is dropped: the register endpoint responds either way. + let (event_tx, _) = mpsc::channel(16); + let (stop_tx, stop_rx) = oneshot::channel(); + + start_with_port( + port, + tls, + ClientInfo { + alias: alias.to_string(), + version: PROTOCOL_VERSION_V2.to_string(), + device_model: Some("Rust".to_string()), + device_type: Some(DeviceType::Headless), + token: fingerprint.to_string(), + }, + None, + Some(ServerConfigV2 { + pin: None, + event_tx, + }), + None, + stop_rx, + ) + .await + .expect("Failed to start server"); + + for _ in 0..100 { + if tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .is_ok() + { + return stop_tx; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("Server did not become reachable on port {port}"); +} + +struct TestInstance { + fingerprint: String, + handle: DiscoveryHandle, + events: mpsc::Receiver, + _stop_tx: oneshot::Sender<()>, +} + +impl TestInstance { + /// Waits for the discovery of the device with the given fingerprint, + /// ignoring unrelated devices that may be on the network. + async fn next_discovery( + &mut self, + fingerprint: &str, + ) -> Option { + let deadline = tokio::time::Instant::now() + RECEIVE_TIMEOUT; + loop { + let event = tokio::time::timeout_at(deadline, self.events.recv()) + .await + .ok()??; + let DiscoveryEvent::Discovered { device } = event else { + continue; + }; + if device.fingerprint == fingerprint { + return Some(device); + } + } + } +} + +/// Starts a discovery instance announcing `server_port` as its HTTP port, or +/// returns `None` when this machine has no interface multicast can be bound to. +async fn start_instance( + alias: &str, + multicast_port: u16, + server_port: u16, +) -> Option { + let cert = generate_self_signed().expect("Failed to generate an identity"); + let (event_tx, events) = mpsc::channel(32); + let (stop_tx, stop_rx) = oneshot::channel(); + + let handle = discovery::start( + DiscoveryConfig { + group: TEST_GROUP, + group_v6: Some(TEST_GROUP_V6), + port: multicast_port, + interface_filter: Default::default(), + device: MulticastDevice { + alias: alias.to_string(), + version: PROTOCOL_VERSION_V2.to_string(), + device_model: Some("Rust".to_string()), + device_type: Some(DeviceType::Headless), + fingerprint: cert.fingerprint.clone(), + port: server_port, + protocol: ProtocolTypeV2::Http, + download: false, + }, + identity: DeviceIdentity { + cert_pem: cert.certificate_pem, + private_key_pem: cert.private_key_pem, + }, + timeout: discovery::DEFAULT_DISCOVERY_TIMEOUT, + event_tx: Some(event_tx), + }, + stop_rx, + ) + .await + .ok()?; + + Some(TestInstance { + fingerprint: cert.fingerprint, + handle, + events, + _stop_tx: stop_tx, + }) +} + +fn skip(reason: &str) { + eprintln!("skipping discovery test: {reason}"); +} + +#[tokio::test] +async fn test_targeted_discovery_stores_and_emits_device() { + let multicast_port = NEXT_MULTICAST_PORT.fetch_add(1, Ordering::Relaxed); + let server_port = free_port(); + let _server_stop = + start_register_server(server_port, "Target", "target-fingerprint", None).await; + + let Some(mut instance) = start_instance("Finder", multicast_port, free_port()).await else { + return skip("no network interface available for multicast"); + }; + + let device = instance + .handle + .discover("127.0.0.1", server_port, ProtocolTypeV2::Http) + .await + .expect("Targeted discovery failed") + .expect("The target must not be mistaken for the device itself"); + + assert_eq!(device.device.alias, "Target"); + assert_eq!(device.device.fingerprint, "target-fingerprint"); + let http = device + .device + .http() + .expect("The device must have an HTTP channel"); + assert_eq!(http.host, "127.0.0.1"); + assert_eq!(http.port, server_port); + + let stored = instance + .handle + .device_by_fingerprint("target-fingerprint") + .expect("The discovered device must be stored"); + assert_eq!(stored.device.alias, "Target"); + + let emitted = instance + .next_discovery("target-fingerprint") + .await + .expect("The discovered device must be emitted"); + assert_eq!(emitted.alias, "Target"); + + // Discovering the same device again emits Updated instead of Discovered. + let updated = instance + .handle + .discover("127.0.0.1", server_port, ProtocolTypeV2::Http) + .await + .expect("Targeted discovery failed") + .expect("The target must still be discoverable"); + assert_eq!(instance.handle.devices().len(), 1); + assert_eq!( + updated.logs.len(), + 2, + "every confirmation must be logged on the stored device" + ); + match instance.events.try_recv() { + Ok(DiscoveryEvent::Updated { device }) => { + assert_eq!(device.fingerprint, "target-fingerprint"); + } + other => panic!("expected an Updated event for a known device, got {other:?}"), + } +} + +#[tokio::test] +async fn test_targeted_discovery_does_not_discover_itself() { + let multicast_port = NEXT_MULTICAST_PORT.fetch_add(1, Ordering::Relaxed); + + let Some(instance) = start_instance("Selfish", multicast_port, free_port()).await else { + return skip("no network interface available for multicast"); + }; + + // A server answering with this instance's own fingerprint, as the + // instance's real server would. + let server_port = free_port(); + let _server_stop = + start_register_server(server_port, "Selfish", &instance.fingerprint, None).await; + + let device = instance + .handle + .discover("127.0.0.1", server_port, ProtocolTypeV2::Http) + .await + .expect("Targeted discovery failed"); + + assert!(device.is_none(), "a device must not discover itself"); + assert!(instance.handle.devices().is_empty()); +} + +#[tokio::test] +async fn test_subnet_scan_finds_device_on_loopback() { + let multicast_port = NEXT_MULTICAST_PORT.fetch_add(1, Ordering::Relaxed); + let server_port = free_port(); + let _server_stop = + start_register_server(server_port, "ScanTarget", "scan-fingerprint", None).await; + + let Some(mut instance) = start_instance("Scanner", multicast_port, free_port()).await else { + return skip("no network interface available for multicast"); + }; + + // The scan probes 127.0.0.0/24 except the interface address itself. + // Loopback routes the whole subnet to this host, so depending on the OS + // the server is found on one or many of the addresses. + let found = instance + .handle + .scan_subnet( + Ipv4Addr::new(127, 0, 0, 99), + server_port, + ProtocolTypeV2::Http, + ) + .await + .expect("Subnet scan failed"); + + assert!(!found.is_empty(), "the scan must find the loopback server"); + assert!(found + .iter() + .all(|device| device.device.fingerprint == "scan-fingerprint")); + + let stored = instance + .handle + .device_by_fingerprint("scan-fingerprint") + .expect("The scanned device must be stored"); + assert_eq!(stored.device.alias, "ScanTarget"); + assert!( + stored + .device + .http_channels() + .all(|http| http.host != "127.0.0.99"), + "the interface address itself must not be probed" + ); + + let emitted = instance + .next_discovery("scan-fingerprint") + .await + .expect("The scanned device must be emitted"); + assert_eq!(emitted.alias, "ScanTarget"); +} + +#[tokio::test] +async fn test_targeted_discovery_reads_fingerprint_from_certificate_on_https() { + let multicast_port = NEXT_MULTICAST_PORT.fetch_add(1, Ordering::Relaxed); + + let Some(instance) = start_instance("TlsFinder", multicast_port, free_port()).await else { + return skip("no network interface available for multicast"); + }; + + // A TLS server whose response body claims a fingerprint that does not + // match its certificate. + let server_cert = generate_self_signed().expect("Failed to generate an identity"); + let server_port = free_port(); + let _server_stop = start_register_server( + server_port, + "TlsTarget", + "claimed-fingerprint", + Some(TlsConfig { + cert: server_cert.certificate_pem.clone(), + private_key: server_cert.private_key_pem.clone(), + }), + ) + .await; + + let device = instance + .handle + .discover("127.0.0.1", server_port, ProtocolTypeV2::Https) + .await + .expect("Targeted discovery failed") + .expect("The target must not be mistaken for the device itself"); + + assert_eq!( + device.device.fingerprint, server_cert.fingerprint, + "on HTTPS the identity must be the certificate fingerprint, not the claimed one" + ); + assert!(instance + .handle + .device_by_fingerprint(&server_cert.fingerprint) + .is_some()); + assert!(instance + .handle + .device_by_fingerprint("claimed-fingerprint") + .is_none()); +} + +#[tokio::test] +async fn test_announcement_is_answered_and_device_stored() { + let multicast_port = NEXT_MULTICAST_PORT.fetch_add(1, Ordering::Relaxed); + + // The receiver answers the announcement with a register request to the + // announcer's HTTP server, so the announcer needs a real one. + let Some(mut receiver) = start_instance("Receiver", multicast_port, free_port()).await else { + return skip("no network interface available for multicast"); + }; + let announcer_port = free_port(); + let Some(announcer) = start_instance("Announcer", multicast_port, announcer_port).await else { + return skip("no network interface available for multicast"); + }; + let _server_stop = + start_register_server(announcer_port, "Announcer", &announcer.fingerprint, None).await; + + announcer.handle.announce().await; + + let Some(device) = receiver.next_discovery(&announcer.fingerprint).await else { + return skip("multicast traffic is not delivered on this machine"); + }; + + assert_eq!(device.alias, "Announcer"); + let http = device.http().expect("The device must have an HTTP channel"); + assert_eq!(http.port, announcer_port); + assert_eq!(http.protocol, ProtocolTypeV2::Http); + + assert!( + receiver + .handle + .device_by_fingerprint(&announcer.fingerprint) + .is_some(), + "the announced device must be stored" + ); + assert!( + announcer + .handle + .device_by_fingerprint(&receiver.fingerprint) + .is_none(), + "answering over HTTP must not make the receiver appear on the announcer's side" + ); +}