diff --git a/.config/nextest.toml b/.config/nextest.toml index 7147a61f..945d00c6 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -24,6 +24,10 @@ docker = { max-threads = 4 } # so the group exists only to keep the plain-cargo path calm and stays # serialized out of caution rather than need. keyring = { max-threads = 1 } +# Every mDNS discovery proof binds the loopback multicast group and drives a +# real daemon. Concurrent proofs contend for the loopback, and that stalls a +# link renewal past its bound, so the proofs serialize. +discovery = { max-threads = 1 } [[profile.default.overrides]] filter = 'test(secret_stores::) | test(teardown::) | test(e2e::) | test(revocation::)' @@ -60,3 +64,13 @@ filter = 'package(connetto-file-server) & (test(serve::) | test(sweep::) | test( test-group = 'docker' threads-required = 2 slow-timeout = { period = "120s", terminate-after = 5 } + +[[profile.default.overrides]] +filter = ''' +test(two_nodes_discover_each_other_and_autolink) +| test(a_renewal_re_registers_under_a_new_fingerprint) +| test(a_stopped_discovery_serves_again) +| test(two_clients_discover_each_other_and_autolink) +| test(clients_without_autolink_report_and_link_only_on_the_call) +''' +test-group = 'discovery' diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 77d0c1c6..8bc530d9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -681,6 +681,10 @@ jobs: rust-tests: name: Rust tests ${{ matrix.name }} runs-on: ubuntu-latest + # The discovery proofs fail here, never skip, when loopback multicast is + # missing (R76 slice 3). + env: + CONNETTO_REQUIRE_MULTICAST: "1" timeout-minutes: 60 strategy: fail-fast: false diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index 88859006..6eee8bc9 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -25,6 +25,10 @@ jobs: coverage: name: Coverage ${{ matrix.name }} runs-on: ubuntu-latest + # The discovery proofs fail here, never skip, when loopback multicast is + # missing (R76 slice 3). + env: + CONNETTO_REQUIRE_MULTICAST: "1" timeout-minutes: 90 # Neither a fork's pull request nor a Dependabot run can read # CODECOV_TOKEN, so the upload runs for same-repo branches and main only. diff --git a/Cargo.lock b/Cargo.lock index a8901d66..b650097c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -619,7 +619,7 @@ dependencies = [ "bitflags 2.13.2", "cexpr", "clang-sys", - "itertools 0.10.5", + "itertools 0.13.0", "proc-macro2", "quote", "regex", @@ -1372,14 +1372,17 @@ name = "connetto-peer" version = "0.0.0" dependencies = [ "connetto-core", + "mdns-sd", "parking_lot", "rcgen 0.14.10", + "ring", "rmp-serde", "rustls", "rustls-pki-types", "rustls-webpki", "serde", "serde_bytes", + "socket2", "thiserror 2.0.20", "tokio", "tokio-rustls", @@ -3009,6 +3012,17 @@ dependencies = [ "zlib-rs", ] +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -4239,6 +4253,16 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "if-addrs" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "image" version = "0.25.10" @@ -5177,6 +5201,21 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "mdns-sd" +version = "0.21.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "127d3355da0615643c88c885e29a1ccb29ca4b1ca1f2e44b3caed2e728a796aa" +dependencies = [ + "fastrand", + "flume", + "if-addrs", + "log", + "mio", + "socket-pktinfo", + "socket2", +] + [[package]] name = "memchr" version = "2.8.3" @@ -5281,6 +5320,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" dependencies = [ "libc", + "log", "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.61.2", ] @@ -5623,7 +5663,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http 1.5.0", @@ -6736,8 +6776,8 @@ version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ - "heck 0.4.1", - "itertools 0.10.5", + "heck 0.5.0", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -6756,7 +6796,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.119", @@ -6804,7 +6844,7 @@ version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb5cd191fad9142d096efc97c64715db73db4adb60c38ebdf22f5b54cd6ea712" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "prost", "prost-build", "prost-types", @@ -8313,12 +8353,23 @@ version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.119", ] +[[package]] +name = "socket-pktinfo" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "612942246d0cc239cfd83af1dfd39be47f649208a3524e5e9da651910128e0ac" +dependencies = [ + "libc", + "socket2", + "windows-sys 0.61.2", +] + [[package]] name = "socket2" version = "0.6.5" @@ -8360,6 +8411,9 @@ name = "spin" version = "0.9.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" +dependencies = [ + "lock_api", +] [[package]] name = "spki" diff --git a/crates/connetto-client/Cargo.toml b/crates/connetto-client/Cargo.toml index 57f27843..2e59d703 100644 --- a/crates/connetto-client/Cargo.toml +++ b/crates/connetto-client/Cargo.toml @@ -166,8 +166,9 @@ ring = { version = "0.17", features = ["std"], optional = true } # The binary derives the schema version it declares from the translated # bundle, the same step the server runs. Native only, with the binary. connetto-schema = { workspace = true, optional = true } -# The peer link the device serves and dials (R76). -connetto-peer = { path = "../connetto-peer", optional = true } +# The peer link the device serves and dials (R76). Its discovery feature is +# on, so a peer build always browses and autolinks what it finds (R76 slice 3). +connetto-peer = { path = "../connetto-peer", optional = true, features = ["discovery"] } # The page codec, one per target, both selected here rather than by a feature # because `ReplicaCipher::Encrypted` cannot be honest without it: in a SQLite # with no codec, `PRAGMA key` is an unrecognised pragma that succeeds silently @@ -333,6 +334,9 @@ rcgen = "0.13" rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } tokio-rustls = { version = "0.26", default-features = false, features = ["tls12"] } tokio = { version = "1", features = ["test-util"] } +# The loopback multicast probe the discovery proofs run before they trust the +# interface, shared with the peer crate's own proofs. +connetto-peer = { path = "../connetto-peer", features = ["discovery", "test-support"] } # The fake Secret Service the Linux store tests serve on a peer-to-peer bus. [target.'cfg(target_os = "linux")'.dev-dependencies] diff --git a/crates/connetto-client/src/builder/native.rs b/crates/connetto-client/src/builder/native.rs index f367d931..c5f3a1d8 100644 --- a/crates/connetto-client/src/builder/native.rs +++ b/crates/connetto-client/src/builder/native.rs @@ -450,6 +450,8 @@ struct DeviceSetup { refused: Option, roots: Vec>, #[cfg(feature = "peer")] + peer_autolink: bool, + #[cfg(feature = "peer")] peer_listen: SocketAddr, #[cfg(feature = "peer")] peer_accepted: Vec, @@ -467,6 +469,8 @@ impl Default for DeviceSetup { refused: None, roots: Vec::new(), #[cfg(feature = "peer")] + peer_autolink: true, + #[cfg(feature = "peer")] peer_listen: SocketAddr::from(([0, 0, 0, 0], 0)), #[cfg(feature = "peer")] peer_accepted: AttestationLevel::ALL.to_vec(), @@ -578,6 +582,15 @@ where self } + /// Whether this device dials the peers discovery finds, on by default + /// (R76 decision 12). + #[cfg(feature = "peer")] + #[must_use] + pub fn with_peer_autolink(mut self, autolink: bool) -> Self { + self.device.peer_autolink = autolink; + self + } + /// The attestation levels this device accepts from its peers, all three /// by default (R76 decision 8). #[cfg(feature = "peer")] @@ -725,22 +738,32 @@ where #[cfg(feature = "peer")] let (peer_events, peer_events_rx) = mpsc::unbounded_channel(); #[cfg(feature = "peer")] + let node = connetto_peer::Node::new( + connetto_peer::Trust { + roots: device.roots.clone(), + accepted: device.peer_accepted, + }, + Arc::new(connetto_peer::SystemClock), + peer_events, + ) + .map_err(|connetto_peer::TrustError::Root { index, .. }| { + ClientError::InvalidDeploymentRoot { index } + })?; + #[cfg(feature = "peer")] + let (discovery_events, discovery_events_rx) = mpsc::unbounded_channel(); + #[cfg(feature = "peer")] let peer = Peer { - node: connetto_peer::Node::new( - connetto_peer::Trust { - roots: device.roots.clone(), - accepted: device.peer_accepted, - }, - Arc::new(connetto_peer::SystemClock), - peer_events, - ) - .map_err( - |connetto_peer::TrustError::Root { index, .. }| { - ClientError::InvalidDeploymentRoot { index } - }, - )?, + node: node.clone(), + discovery: connetto_peer::Discovery::new( + node, + device.peer_autolink, + discovery_events, + ), listen: device.peer_listen, events: peer_events_rx, + discovery_events: discovery_events_rx, + #[cfg(all(feature = "peer", target_os = "android"))] + java: device.java.clone(), }; let (enroller, handle) = Enroller::new( keys, diff --git a/crates/connetto-client/src/enrolment/task.rs b/crates/connetto-client/src/enrolment/task.rs index dbc1e4c3..7d2b76a4 100644 --- a/crates/connetto-client/src/enrolment/task.rs +++ b/crates/connetto-client/src/enrolment/task.rs @@ -2,7 +2,7 @@ //! one line of R74's lifecycle table per branch. use core::pin::Pin; -use core::sync::atomic::{AtomicU64, Ordering}; +use core::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use core::time::Duration; use std::sync::Arc; use std::time::SystemTime; @@ -23,7 +23,7 @@ use connetto_core::messages::{ EnrolRequest, FatalErrorReason, RevokeDeviceRequest, SignedList, SyncStatus, }; #[cfg(feature = "peer")] -use connetto_peer::{CloseReason, LinkError, PeerEvent}; +use connetto_peer::{CloseReason, DiscoveryEvent, LinkError, PeerEvent}; use tokio::sync::broadcast::error::RecvError; use tokio::sync::{broadcast, mpsc, oneshot, watch}; @@ -31,6 +31,8 @@ use super::{Answer, Held, KeptList, Standing, TOLERANCE, half_life}; #[cfg(feature = "peer")] use crate::PeerError; use crate::device_key::{ChipError, ChipKeys, KeyRecords, OpenedKey}; +#[cfg(all(feature = "peer", target_os = "android"))] +use crate::multicast::MulticastLock; use crate::{ClientError, ClientEvent}; /// How long one answer is waited for. @@ -187,10 +189,18 @@ enum Command { pub(crate) struct Peer { /// The node presenting this device's identity. pub(crate) node: connetto_peer::Node, + /// The discovery driver, browsing and dialing what it finds (R76). + pub(crate) discovery: connetto_peer::Discovery, /// The address the listener binds. pub(crate) listen: SocketAddr, /// The node's link events, until the task reads them. pub(crate) events: mpsc::UnboundedReceiver, + /// Discovery's events, until the task reads them. + pub(crate) discovery_events: mpsc::UnboundedReceiver, + /// The application's JNI access, for the multicast lock the browse holds + /// (R76). + #[cfg(all(feature = "peer", target_os = "android"))] + pub(crate) java: Option>, } /// The next peer-link event, or nothing when the build has no peer link (R76). @@ -220,6 +230,33 @@ impl PeerEvents { } } +/// The next discovery event, or nothing when the build has no peer link +/// (R76). +#[cfg(feature = "peer")] +type NextDiscoveryEvent = Option; +#[cfg(not(feature = "peer"))] +type NextDiscoveryEvent = core::convert::Infallible; + +/// Discovery's event receiver, empty when the build has no peer link (R76). +struct DiscoveryEvents { + #[cfg(feature = "peer")] + rx: mpsc::UnboundedReceiver, +} + +impl DiscoveryEvents { + /// The next event, or a future that never resolves without the peer link. + async fn next(&mut self) -> NextDiscoveryEvent { + #[cfg(feature = "peer")] + { + self.rx.recv().await + } + #[cfg(not(feature = "peer"))] + { + core::future::pending().await + } + } +} + /// The task's half the builder hands to the client. pub(crate) struct Enroller { keys: Arc, @@ -232,6 +269,10 @@ pub(crate) struct Enroller { commands: mpsc::UnboundedReceiver, published: watch::Sender>, home: watch::Sender>, + /// Whether the local clock puts the held certificate outside its window + /// (decision 29), shared with the handle so a dial reads the standing the + /// task does (R76). + clock_off: Arc, #[cfg(feature = "peer")] peer: Option, } @@ -243,6 +284,10 @@ pub(crate) struct EnrolHandle { published: watch::Receiver>, home: watch::Receiver>, #[cfg(feature = "peer")] + /// The task's clock-off latch, its dials refusing by the window while it + /// is set (R76). + clock_off: Arc, + #[cfg(feature = "peer")] peer: connetto_peer::Node, } @@ -272,6 +317,7 @@ impl Enroller { let (sender, commands) = mpsc::unbounded_channel(); let (published, observed) = watch::channel(held.as_ref().map(|held| held.leaf.clone())); let (home, homed) = watch::channel(None); + let clock_off = Arc::new(AtomicBool::new(false)); #[cfg(feature = "peer")] let peer_node = peer.node.clone(); ( @@ -286,6 +332,7 @@ impl Enroller { commands, published, home, + clock_off: Arc::clone(&clock_off), #[cfg(feature = "peer")] peer: Some(peer), }, @@ -295,6 +342,8 @@ impl Enroller { published: observed, home: homed, #[cfg(feature = "peer")] + clock_off, + #[cfg(feature = "peer")] peer: peer_node, }, ) @@ -362,6 +411,9 @@ impl EnrolHandle { let held = self.published.borrow().clone(); match Standing::of_leaf(held.as_ref(), SystemTime::now()) { Standing::NoKey => Err(PeerError::NoIdentity), + Standing::Expired if self.clock_off.load(Ordering::Acquire) => { + Err(PeerError::ClockOutsideWindow) + } Standing::Expired => { let _ = self.commands.send(Command::CertificateExpired); Err(PeerError::CertificateExpired) @@ -457,9 +509,6 @@ struct Run { kept: HashMap, KeptList>, /// A certificate the root withdrew, whose key enrols again. withdrawn: Option, - /// Whether the local clock puts the held certificate outside its window - /// (decision 29). - clock_off: bool, /// Whether the deployment refused the device's attestation level, so it /// asks again only on its next connection (decision 33). attestation_refused: bool, @@ -473,6 +522,18 @@ struct Run { peer_standing: Standing, /// The peer link's events, until the task ends them (R76). peer_events: PeerEvents, + /// The discovery driver, behind the peer link (R76). + #[cfg(feature = "peer")] + discovery: connetto_peer::Discovery, + /// Discovery's events, until the task ends them (R76). + discovery_events: DiscoveryEvents, + /// The application's JNI access, for the multicast lock the browse holds + /// (R76). + #[cfg(all(feature = "peer", target_os = "android"))] + java: Option>, + /// The multicast lock the browse holds, until the standing leaves (R76). + #[cfg(all(feature = "peer", target_os = "android"))] + multicast: Option, } /// A certificate withdrawn because the root revoked its issuer. @@ -683,7 +744,7 @@ impl Run { match standing { Standing::ClockOff => self.clock_outside(false), Standing::Expired => self.clock_outside(true), - _ => self.clock_off = false, + _ => self.enroller.clock_off.store(false, Ordering::Release), } #[cfg(feature = "peer")] self.peer_transition(standing).await; @@ -824,9 +885,11 @@ impl Run { self.clock_outside(false); Standing::ClockOff } - Standing::Expired if self.clock_off => Standing::ClockOff, + Standing::Expired if self.enroller.clock_off.load(Ordering::Acquire) => { + Standing::ClockOff + } standing => { - self.clock_off = false; + self.enroller.clock_off.store(false, Ordering::Release); standing } } @@ -834,8 +897,7 @@ impl Run { /// Enter `ClockOff`, raising `ClockOutsideWindow` once (decision 29). fn clock_outside(&mut self, ahead: bool) { - if !self.clock_off { - self.clock_off = true; + if !self.enroller.clock_off.swap(true, Ordering::Release) { self.link.emit(ClientEvent::ClockOutsideWindow { ahead }); } } @@ -889,17 +951,33 @@ impl Run { return; } }; + let fingerprint = connetto_peer::Fingerprint::of(&certificate); let identity = connetto_peer::Identity { certificate, issuer, key, }; - if let Err(err) = self.peer.serve(self.peer_listen, identity) { - tracing::warn!(error = %err, "the peer listener could not bind"); - self.link.emit(ClientEvent::PeerListenFailed { - address: self.peer_listen, - error: err.to_string(), - }); + match self.peer.serve(self.peer_listen, identity) { + Ok(bound) => { + #[cfg(all(feature = "peer", target_os = "android"))] + if let Some(java) = self.java.clone() { + match MulticastLock::acquire(java) { + Ok(lock) => self.multicast = Some(lock), + Err(err) => tracing::warn!( + error = %err, + "the multicast lock will not hold, so the browse holds none" + ), + } + } + self.discovery.serve(bound.port(), fingerprint); + } + Err(err) => { + tracing::warn!(error = %err, "the peer listener could not bind"); + self.link.emit(ClientEvent::PeerListenFailed { + address: self.peer_listen, + error: err.to_string(), + }); + } } } @@ -908,6 +986,11 @@ impl Run { #[cfg(feature = "peer")] fn peer_stop(&mut self, reason: CloseReason) { self.peer_standing = Standing::NoKey; + #[cfg(all(feature = "peer", target_os = "android"))] + { + self.multicast = None; + } + self.discovery.stop(); self.peer.stop(reason); } @@ -972,7 +1055,6 @@ pub(crate) async fn run(link: L, mut enroller: Enroller) { key: None, kept, withdrawn: None, - clock_off: false, attestation_refused: false, #[cfg(feature = "peer")] peer: peer.node, @@ -982,8 +1064,20 @@ pub(crate) async fn run(link: L, mut enroller: Enroller) { peer_standing: Standing::NoKey, #[cfg(feature = "peer")] peer_events: PeerEvents { rx: peer.events }, + #[cfg(feature = "peer")] + discovery: peer.discovery, + #[cfg(feature = "peer")] + discovery_events: DiscoveryEvents { + rx: peer.discovery_events, + }, + #[cfg(all(feature = "peer", target_os = "android"))] + java: peer.java, + #[cfg(all(feature = "peer", target_os = "android"))] + multicast: None, #[cfg(not(feature = "peer"))] peer_events: PeerEvents {}, + #[cfg(not(feature = "peer"))] + discovery_events: DiscoveryEvents {}, }; // Opened at once, so a lost key is noticed before any connection. if let Err(err) = run.key().await { @@ -1046,24 +1140,42 @@ pub(crate) async fn run(link: L, mut enroller: Enroller) { } peer = run.peer_events.next() => { #[cfg(feature = "peer")] - match peer { - Some(PeerEvent::Linked { peer }) => { - run.link.emit(ClientEvent::PeerLinked { peer }); - } - Some(PeerEvent::Unlinked { peer, reason }) => { - run.link.emit(ClientEvent::PeerUnlinked { peer, reason }); - } - Some(PeerEvent::ListReceived { list, signer }) => { - run.take_lists(vec![SignedList { list, signer }]).await; - if run.withdrawn.is_some() { - due = run.link.connected().await; + if let Some(event) = peer { + run.discovery.on_node_event(&event); + match event { + PeerEvent::Linked { peer } => { + run.link.emit(ClientEvent::PeerLinked { peer }); + } + PeerEvent::Unlinked { peer, reason } => { + run.link.emit(ClientEvent::PeerUnlinked { peer, reason }); + } + PeerEvent::ListReceived { list, signer } => { + run.take_lists(vec![SignedList { list, signer }]).await; + if run.withdrawn.is_some() { + due = run.link.connected().await; + } } } - None => {} } #[cfg(not(feature = "peer"))] match peer {} }, + discovery = run.discovery_events.next() => { + #[cfg(feature = "peer")] + if let Some(event) = discovery { + match event { + DiscoveryEvent::Found { address, fingerprint } => { + run.link + .emit(ClientEvent::PeerFound { address, fingerprint }); + } + DiscoveryEvent::Gone { fingerprint } => { + run.link.emit(ClientEvent::PeerGone { fingerprint }); + } + } + } + #[cfg(not(feature = "peer"))] + match discovery {} + }, () = tokio::time::sleep(look) => { let _ = run.stand_by().await; due = run.link.connected().await; @@ -1073,5 +1185,5 @@ pub(crate) async fn run(link: L, mut enroller: Enroller) { } // The task is done, so the peer link ends with it (R76). #[cfg(feature = "peer")] - run.peer.stop(CloseReason::Closed); + run.peer_stop(CloseReason::Closed); } diff --git a/crates/connetto-client/src/enrolment/tests.rs b/crates/connetto-client/src/enrolment/tests.rs index 90faea48..bd37598a 100644 --- a/crates/connetto-client/src/enrolment/tests.rs +++ b/crates/connetto-client/src/enrolment/tests.rs @@ -33,6 +33,8 @@ use crate::ClientEvent; #[cfg(feature = "peer")] use crate::PeerError; use crate::device_key::{KeyRecords, OpenedKey, open_software_key}; +#[cfg(feature = "peer")] +use connetto_peer::test_support::loopback_multicast; const HOUR: Duration = Duration::from_hours(1); const MINUTE: Duration = Duration::from_mins(1); @@ -423,6 +425,9 @@ struct FakeLink { ask: Ask, /// A grant answer waits on it first, so a test holds the renewal open. gate: Option>, + /// A granted leaf's `not_before`, so a test's clock can stand past + /// `not_after` when the grant lands. + grant_from: Option, } impl Link for FakeLink { @@ -495,7 +500,7 @@ impl Link for FakeLink { .issue( &request, "alice", - SystemTime::now(), + self.grant_from.unwrap_or_else(SystemTime::now), 12 * HOUR, serial, AttestationLevel::Unproven, @@ -658,6 +663,7 @@ async fn a_refused_attestation_is_raised_once_and_asks_again_only_on_the_next_co certificate: authority.certificate, ask: Ask::Refused(EnrolRefusal::AttestationRequired), gate: None, + grant_from: None, }; let (enroller, _handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -740,6 +746,7 @@ async fn a_renewal_sends_no_attestation() { certificate: authority.certificate, ask: Ask::Grant, gate: None, + grant_from: None, }; let (enroller, _handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -792,6 +799,7 @@ async fn a_reissue_refused_as_revoked_deletes_the_key_and_reports_revoked() { certificate: authority.certificate, ask: Ask::Refused(EnrolRefusal::Revoked), gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -847,6 +855,7 @@ async fn a_reissue_refused_over_the_ceiling_reports_the_ceiling() { certificate: authority.certificate, ask: Ask::Refused(EnrolRefusal::OverCeiling { ceiling_secs: 600 }), gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -895,6 +904,7 @@ async fn a_reissue_refused_for_another_reason_reports_that_reason() { certificate: authority.certificate, ask: Ask::Refused(EnrolRefusal::ChallengeExpired), gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -942,6 +952,7 @@ async fn a_reissue_with_no_answer_is_offline() { certificate: authority.certificate, ask: Ask::Silent, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -991,6 +1002,7 @@ async fn a_reissue_lost_on_the_wire_is_offline() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1038,6 +1050,7 @@ async fn a_reissue_violating_the_protocol_is_a_device_error() { certificate: authority.certificate, ask: Ask::Violated, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1065,25 +1078,50 @@ async fn a_reissue_violating_the_protocol_is_a_device_error() { run.await.expect("the run ends"); } -/// A peer node the run drives, on the suite's root and a loopback port (R76). +/// A loopback port the discovery's mDNS daemon binds, free at the probe. #[cfg(feature = "peer")] -fn peer(authority: &Authority) -> Peer { +fn fresh_mdns_port() -> u16 { + std::net::UdpSocket::bind("127.0.0.1:0") + .expect("the loopback binds") + .local_addr() + .expect("the bound address") + .port() +} + +/// A peer node the run drives, on the suite's root and a loopback port (R76), +/// whose discovery autolinks `autolink` on the mDNS `port`. +#[cfg(feature = "peer")] +fn peer_with(authority: &Authority, autolink: bool, port: u16) -> Peer { let (tx, rx) = mpsc::unbounded_channel(); + let node = connetto_peer::Node::new( + connetto_peer::Trust { + roots: vec![authority.root.clone()], + accepted: AttestationLevel::ALL.to_vec(), + }, + Arc::new(connetto_peer::SystemClock), + tx, + ) + .expect("the roots hold keys"); + let (discovery_events, discovery_events_rx) = mpsc::unbounded_channel(); Peer { - node: connetto_peer::Node::new( - connetto_peer::Trust { - roots: vec![authority.root.clone()], - accepted: AttestationLevel::ALL.to_vec(), - }, - Arc::new(connetto_peer::SystemClock), - tx, - ) - .expect("the roots hold keys"), + node: node.clone(), + discovery: connetto_peer::Discovery::new(node, autolink, discovery_events) + .with_mdns_port(port) + .loopback_only(), listen: "127.0.0.1:0".parse().expect("a loopback address"), events: rx, + discovery_events: discovery_events_rx, + #[cfg(all(feature = "peer", target_os = "android"))] + java: None, } } +/// A peer node the run drives, on the suite's root and a loopback port (R76). +#[cfg(feature = "peer")] +fn peer(authority: &Authority) -> Peer { + peer_with(authority, true, fresh_mdns_port()) +} + /// A second node holding its own identity under the same root, served on a /// loopback port, to link against the task's node (R76). #[cfg(feature = "peer")] @@ -1158,6 +1196,7 @@ async fn a_fresh_device_serves_its_peer_listener_at_open() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1200,6 +1239,7 @@ async fn an_expired_device_refuses_to_link_and_raises_the_event() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1260,6 +1300,7 @@ async fn a_clock_off_device_refuses_to_link_outside_its_window() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1309,6 +1350,79 @@ async fn a_clock_off_device_refuses_to_link_outside_its_window() { run.await.expect("the run ends"); } +/// A grant the local clock already puts past `not_after` lands the task in +/// the window refusal, so a dial is refused by the window, not the expiry, +/// and no expiry event joins the task's own (R76 proof 3). +#[cfg(feature = "peer")] +#[tokio::test] +async fn a_grant_past_its_window_refuses_the_dial_as_a_window() { + let (_, der) = attesting_key(); + let now = SystemTime::now(); + let authority = authority(now).await; + let state = Arc::new(Mutex::new(FakeState::default())); + let (events, _events) = broadcast::channel(64); + let (end_tx, end_rx) = watch::channel(()); + #[cfg(feature = "peer")] + let peer = peer(&authority); + // The server just granted it, so a window that does not hold it now says + // the local clock runs ahead of `not_after`. + let link = FakeLink { + state: Arc::clone(&state), + events, + end: end_rx, + issuer: Arc::new(authority.issuer), + certificate: authority.certificate, + ask: Ask::Grant, + gate: None, + grant_from: Some(now - 13 * HOUR), + }; + let (enroller, handle) = Enroller::new( + Arc::new(RunKeys { der }), + Some(HOUR), + Vec::new(), + vec![authority.root.clone()], + None, + Vec::new(), + inbox(), + peer, + ); + let run = tokio::spawn(run(link, enroller)); + + poll_until(|| !state.lock().stored.is_empty()).await; + let err = handle + .link_peer("127.0.0.1:0".parse().expect("loopback")) + .await + .expect_err("the window refuses the dial"); + assert!( + matches!(err, PeerError::ClockOutsideWindow), + "the refusal names the window, got {err:?}" + ); + // The window event is the task's own, raised once at the grant. + poll_until(|| { + state + .lock() + .emitted + .iter() + .any(|event| matches!(event, ClientEvent::ClockOutsideWindow { ahead: true })) + }) + .await; + assert!( + state + .lock() + .emitted + .iter() + .all(|event| !matches!(event, ClientEvent::CertificateExpired)), + "a window refusal is not an expiry" + ); + assert!( + handle.peer_address().is_none(), + "nothing serves a window-out identity" + ); + + end_tx.send(()).expect("the run ends"); + run.await.expect("the run ends"); +} + /// The refusal of a device that holds no certificate (R76 proof 4). #[cfg(feature = "peer")] #[tokio::test] @@ -1328,6 +1442,7 @@ async fn a_certificate_less_device_refuses_to_link() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1391,6 +1506,7 @@ async fn a_look_at_the_expiry_closes_the_live_peer_links() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1485,6 +1601,7 @@ async fn a_granted_renewal_keeps_the_peer_port_and_the_live_link() { certificate: authority.certificate, ask: Ask::Grant, gate: Some(gate.clone()), + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1587,6 +1704,7 @@ async fn a_peer_list_revoking_the_own_serial_revokes_the_device() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1713,6 +1831,7 @@ async fn a_stale_peer_list_changes_nothing() { certificate: authority.certificate, ask: Ask::Lost, gate: None, + grant_from: None, }; let (enroller, handle) = Enroller::new( Arc::new(RunKeys { der }), @@ -1757,3 +1876,225 @@ async fn a_stale_peer_list_changes_nothing() { end_tx.send(()).expect("the run ends"); run.await.expect("the run ends"); } + +/// Whether `state` has reported the instance `fingerprint` found (R76). +#[cfg(feature = "peer")] +fn reports_found(state: &Arc>, fingerprint: connetto_peer::Fingerprint) -> bool { + state.lock().emitted.iter().any(|event| { + matches!( + event, + ClientEvent::PeerFound { + fingerprint: found, + .. + } if *found == fingerprint + ) + }) +} + +/// Whether `state` has reported a peer link (R76). +#[cfg(feature = "peer")] +fn reports_linked(state: &Arc>) -> bool { + state + .lock() + .emitted + .iter() + .any(|event| matches!(event, ClientEvent::PeerLinked { .. })) +} + +/// Two clients on loopback discover each other through mDNS and autolink with +/// no `link_peer` call, each reporting the other found (R76 proof 4). +#[cfg(feature = "peer")] +#[tokio::test] +async fn two_clients_discover_each_other_and_autolink() { + if !loopback_multicast() { + eprintln!("the host will not multicast on the loopback, so the discovery proof skips"); + return; + } + eprintln!("R76-PROOF-RAN two_clients_discover_each_other_and_autolink"); + let (key_a, der_a) = attesting_key(); + let (key_b, der_b) = attesting_key(); + let now = whole_second(); + let authority = authority(now).await; + let held_a = issue_for(&authority, &key_a, now, 12 * HOUR, [5; 16]); + let held_b = issue_for(&authority, &key_b, now, 12 * HOUR, [6; 16]); + let fp_a = connetto_peer::Fingerprint::of(&held_a.certificate); + let fp_b = connetto_peer::Fingerprint::of(&held_b.certificate); + let port = fresh_mdns_port(); + let peer_a = peer_with(&authority, true, port); + let peer_b = peer_with(&authority, true, port); + let issuer = Arc::new(authority.issuer); + let state_a = Arc::new(Mutex::new(FakeState::default())); + let (events_a, _events_a) = broadcast::channel(64); + let (end_tx, end_rx) = watch::channel(()); + let state_b = Arc::new(Mutex::new(FakeState::default())); + let (events_b, _events_b) = broadcast::channel(64); + let (fin_tx, fin_rx) = watch::channel(()); + let link_a = FakeLink { + state: Arc::clone(&state_a), + events: events_a, + end: end_rx, + issuer: Arc::clone(&issuer), + certificate: authority.certificate.clone(), + ask: Ask::Lost, + gate: None, + grant_from: None, + }; + let link_b = FakeLink { + state: Arc::clone(&state_b), + events: events_b, + end: fin_rx, + issuer: Arc::clone(&issuer), + certificate: authority.certificate, + ask: Ask::Lost, + gate: None, + grant_from: None, + }; + let (enroller_a, _handle_a) = Enroller::new( + Arc::new(RunKeys { der: der_a }), + Some(HOUR), + Vec::new(), + vec![authority.root.clone()], + Some(held_a), + Vec::new(), + inbox(), + peer_a, + ); + let (enroller_b, _handle_b) = Enroller::new( + Arc::new(RunKeys { der: der_b }), + Some(HOUR), + Vec::new(), + vec![authority.root.clone()], + Some(held_b), + Vec::new(), + inbox(), + peer_b, + ); + let run_a = tokio::spawn(run(link_a, enroller_a)); + let run_b = tokio::spawn(run(link_b, enroller_b)); + + poll_until(|| reports_found(&state_a, fp_b)).await; + poll_until(|| reports_found(&state_b, fp_a)).await; + poll_until(|| reports_linked(&state_a)).await; + poll_until(|| reports_linked(&state_b)).await; + + end_tx.send(()).expect("the run ends"); + fin_tx.send(()).expect("the run ends"); + run_a.await.expect("the run ends"); + run_b.await.expect("the run ends"); +} + +/// Two clients on loopback with autolink off report each other found and link +/// only the one that calls `link_peer` (R76 proof 4). +#[cfg(feature = "peer")] +#[tokio::test] +#[expect( + clippy::too_many_lines, + reason = "the two reports and the call's one link are one scenario, in the order discovery crosses them" +)] +async fn clients_without_autolink_report_and_link_only_on_the_call() { + if !loopback_multicast() { + eprintln!("the host will not multicast on the loopback, so the discovery proof skips"); + return; + } + eprintln!("R76-PROOF-RAN clients_without_autolink_report_and_link_only_on_the_call"); + let (key_a, der_a) = attesting_key(); + let (key_b, der_b) = attesting_key(); + let now = whole_second(); + let authority = authority(now).await; + let held_a = issue_for(&authority, &key_a, now, 12 * HOUR, [5; 16]); + let held_b = issue_for(&authority, &key_b, now, 12 * HOUR, [6; 16]); + let fp_a = connetto_peer::Fingerprint::of(&held_a.certificate); + let fp_b = connetto_peer::Fingerprint::of(&held_b.certificate); + let port = fresh_mdns_port(); + let peer_a = peer_with(&authority, false, port); + let peer_b = peer_with(&authority, false, port); + let issuer = Arc::new(authority.issuer); + let state_a = Arc::new(Mutex::new(FakeState::default())); + let (events_a, _events_a) = broadcast::channel(64); + let (end_tx, end_rx) = watch::channel(()); + let state_b = Arc::new(Mutex::new(FakeState::default())); + let (events_b, _events_b) = broadcast::channel(64); + let (fin_tx, fin_rx) = watch::channel(()); + let link_a = FakeLink { + state: Arc::clone(&state_a), + events: events_a, + end: end_rx, + issuer: Arc::clone(&issuer), + certificate: authority.certificate.clone(), + ask: Ask::Lost, + gate: None, + grant_from: None, + }; + let link_b = FakeLink { + state: Arc::clone(&state_b), + events: events_b, + end: fin_rx, + issuer: Arc::clone(&issuer), + certificate: authority.certificate, + ask: Ask::Lost, + gate: None, + grant_from: None, + }; + let (enroller_a, handle_a) = Enroller::new( + Arc::new(RunKeys { der: der_a }), + Some(HOUR), + Vec::new(), + vec![authority.root.clone()], + Some(held_a), + Vec::new(), + inbox(), + peer_a, + ); + let (enroller_b, _handle_b) = Enroller::new( + Arc::new(RunKeys { der: der_b }), + Some(HOUR), + Vec::new(), + vec![authority.root.clone()], + Some(held_b), + Vec::new(), + inbox(), + peer_b, + ); + let run_a = tokio::spawn(run(link_a, enroller_a)); + let run_b = tokio::spawn(run(link_b, enroller_b)); + + poll_until(|| reports_found(&state_a, fp_b)).await; + poll_until(|| reports_found(&state_b, fp_a)).await; + // Seconds past the find, and nothing dials itself. + tokio::time::sleep(Duration::from_secs(3)).await; + { + let (a, b) = (state_a.lock(), state_b.lock()); + assert!( + a.emitted + .iter() + .all(|event| !matches!(event, ClientEvent::PeerLinked { .. })), + "autolink off dials nothing" + ); + assert!( + b.emitted + .iter() + .all(|event| !matches!(event, ClientEvent::PeerLinked { .. })), + "autolink off dials nothing" + ); + } + let address = state_a + .lock() + .emitted + .iter() + .find_map(|event| match event { + ClientEvent::PeerFound { + address, + fingerprint, + } if *fingerprint == fp_b => Some(*address), + _ => None, + }) + .expect("the found report carries the other's address"); + handle_a.link_peer(address).await.expect("the call links"); + poll_until(|| reports_linked(&state_a)).await; + poll_until(|| reports_linked(&state_b)).await; + + end_tx.send(()).expect("the run ends"); + fin_tx.send(()).expect("the run ends"); + run_a.await.expect("the run ends"); + run_b.await.expect("the run ends"); +} diff --git a/crates/connetto-client/src/lib.rs b/crates/connetto-client/src/lib.rs index 8b52f71f..5dc57ad4 100644 --- a/crates/connetto-client/src/lib.rs +++ b/crates/connetto-client/src/lib.rs @@ -93,6 +93,8 @@ pub mod harden; mod keyring; pub mod live; mod memory_stores; +#[cfg(all(feature = "peer", target_os = "android"))] +mod multicast; pub mod reconnect; pub mod replica; mod subscriptions; @@ -1247,6 +1249,20 @@ pub enum ClientEvent { /// Why the bind failed. error: String, }, + /// Discovery found a peer instance at `address` (R76). + #[cfg(feature = "peer")] + PeerFound { + /// The resolved address to dial. + address: SocketAddr, + /// The instance's leaf fingerprint. + fingerprint: connetto_peer::Fingerprint, + }, + /// Discovery lost a peer instance (R76). + #[cfg(feature = "peer")] + PeerGone { + /// The instance's leaf fingerprint. + fingerprint: connetto_peer::Fingerprint, + }, } /// A primary-key column value carried on a mutation event. diff --git a/crates/connetto-client/src/multicast.rs b/crates/connetto-client/src/multicast.rs new file mode 100644 index 00000000..ab060912 --- /dev/null +++ b/crates/connetto-client/src/multicast.rs @@ -0,0 +1,108 @@ +//! The Wi-Fi multicast lock the peer discovery's browse holds on Android +//! (R76), through the application's JNI access. Receiving the browse's +//! mDNS traffic needs the lock, so the browse holds it for its standing. + +use std::sync::Arc; + +use jni::JNIEnv; +use jni::objects::{GlobalRef, JValue}; + +use crate::device_key::JavaAccess; + +/// The tag the lock announces to the Wi-Fi manager. +const LOCK_TAG: &str = "connetto-peer"; + +/// A failure reaching the Wi-Fi multicast lock through the Java VM. +#[derive(Debug, thiserror::Error)] +pub(crate) enum MulticastFailure { + /// A refusal from the Java side, kept as the exception's text, else the JNI error's. + #[error("the multicast lock will not hold: {0}")] + Java(String), + /// The process has no reachable Java VM. + #[error("the multicast lock will not hold: {0}")] + NoVm(String), + /// The application's JNI access returned without running the body. + #[error("the multicast lock will not hold: the JNI access ran nothing")] + RanNothing, +} + +/// The held `WifiManager.MulticastLock`, over the application's JNI access. +pub(crate) struct MulticastLock { + java: Arc, + lock: GlobalRef, +} + +impl MulticastLock { + /// Take the Wi-Fi service's multicast lock and acquire it, through `java`. + /// + /// # Errors + /// + /// [`MulticastFailure`] when the VM is unreachable or the service refuses. + pub(crate) fn acquire(java: Arc) -> Result { + let mut body = Some(|env: &mut JNIEnv| -> jni::errors::Result { + let application = env + .call_static_method( + "android/app/ActivityThread", + "currentApplication", + "()Landroid/app/Application;", + &[], + )? + .l()?; + let wifi = env.new_string("wifi")?; + let service = env + .call_method( + &application, + "getSystemService", + "(Ljava/lang/String;)Ljava/lang/Object;", + &[JValue::Object(&wifi)], + )? + .l()?; + let tag = env.new_string(LOCK_TAG)?; + let lock = env + .call_method( + &service, + "createMulticastLock", + "(Ljava/lang/String;)Landroid/net/wifi/WifiManager$MulticastLock;", + &[JValue::Object(&tag)], + )? + .l()?; + env.call_method(&lock, "acquire", "()V", &[])?; + env.new_global_ref(&lock) + }); + let mut outcome = None; + java.with_env(&mut |env| { + let Some(body) = body.take() else { return }; + outcome = Some(body(env).map_err(|err| exception_text(env, &err))); + }) + .map_err(|err| MulticastFailure::NoVm(err.to_string()))?; + let lock = outcome.unwrap_or_else(|| Err(MulticastFailure::RanNothing))?; + Ok(Self { java, lock }) + } + + /// Let the Wi-Fi service have the multicast again. + pub(crate) fn release(&self) { + let _ = self.java.with_env(&mut |env| { + let _ = env.call_method(&self.lock, "release", "()V", &[]); + }); + } +} + +impl Drop for MulticastLock { + fn drop(&mut self) { + self.release(); + } +} + +/// The pending Java exception's text, cleared, else the JNI error's. +fn exception_text(env: &mut JNIEnv, err: &jni::errors::Error) -> MulticastFailure { + let pending = env.exception_occurred().ok().filter(|ex| !ex.is_null()); + let _ = env.exception_clear(); + let text = pending + .and_then(|ex| { + env.call_method(&ex, "toString", "()Ljava/lang/String;", &[]) + .ok() + }) + .and_then(|value| value.l().ok()) + .and_then(|text| env.get_string(&text.into()).ok().map(String::from)); + MulticastFailure::Java(text.unwrap_or_else(|| err.to_string())) +} diff --git a/crates/connetto-client/tests/it/enrolment.rs b/crates/connetto-client/tests/it/enrolment.rs index a7f7a78a..6672b291 100644 --- a/crates/connetto-client/tests/it/enrolment.rs +++ b/crates/connetto-client/tests/it/enrolment.rs @@ -136,7 +136,7 @@ async fn enrolment_phase() { }; match phase.as_str() { "first" => first_connect().await, - "renewal" => renewal().await, + "renewal" => Box::pin(renewal()).await, "reissue" => reissue().await, "cap" => cap().await, "revoke" => Box::pin(revoke()).await, @@ -146,7 +146,7 @@ async fn enrolment_phase() { "forget" => forget().await, "report" => Box::pin(report()).await, "learned" => Box::pin(learned_at_connect()).await, - "untrusted" => untrusted().await, + "untrusted" => Box::pin(untrusted()).await, "no-roots" => no_roots().await, "rotation" => Box::pin(rotation_withdraws()).await, "clock-behind" => Box::pin(clock_behind()).await, @@ -1869,17 +1869,17 @@ async fn peer_report() { matches!(refused, connetto_client::PeerError::Link(_)), "the refusal is a dial refusal, got {refused:?}" ); - let until = Instant::now() + Duration::from_secs(3); - let mut linked_c = false; - while Instant::now() < until { - if let Ok(event) = b_events.try_recv() { - linked_c |= matches!( + assert!( + next_event(&mut b_events, Duration::from_secs(3), |event| { + matches!( event, ClientEvent::PeerLinked { peer } if peer.key() == c_cert.identity().key() - ); - } - } - assert!(!linked_c, "b never links c"); + ) + }) + .await + .is_none(), + "b never links c" + ); a.close().await; b2.close().await; diff --git a/crates/connetto-peer/Cargo.toml b/crates/connetto-peer/Cargo.toml index 7425a51f..bf078786 100644 --- a/crates/connetto-peer/Cargo.toml +++ b/crates/connetto-peer/Cargo.toml @@ -15,6 +15,15 @@ homepage.workspace = true [lints] workspace = true +[features] +default = [] +# The mDNS advertisement and browse beside the policy that dials what they +# find, and the leaf fingerprint the TXT record carries (R76 slice 3). +discovery = ["dep:mdns-sd", "dep:ring"] +# The loopback multicast probe the discovery proofs share with +# `connetto-client`'s suite, for test builds only. +test-support = ["dep:socket2"] + [dependencies] connetto-core = { workspace = true, features = ["device-cert"] } # TLS 1.3 only, over the ring provider: the device key signs @@ -41,6 +50,10 @@ serde.workspace = true serde_bytes.workspace = true rmp-serde.workspace = true thiserror.workspace = true +# The advertisement and the fingerprint hash, behind the discovery feature. +mdns-sd = { version = "0.21", optional = true } +ring = { version = "0.17", optional = true } +socket2 = { version = "0.6", optional = true } tracing = { version = "0.1", default-features = false, features = ["std"] } [dev-dependencies] @@ -49,4 +62,7 @@ tracing = { version = "0.1", default-features = false, features = ["std"] } rcgen = { version = "0.14.10", default-features = false, features = ["ring", "x509-parser"] } # `time::pause` drives the 15/45 s liveness clocks in the proofs. tokio = { version = "1", default-features = false, features = ["test-util"] } +# The shared probe in `test_support`, which the crate's own proofs build +# without the feature. +socket2 = "0.6" uuid = { version = "1", default-features = false } diff --git a/crates/connetto-peer/src/discovery.rs b/crates/connetto-peer/src/discovery.rs new file mode 100644 index 00000000..ec6ad0f4 --- /dev/null +++ b/crates/connetto-peer/src/discovery.rs @@ -0,0 +1,611 @@ +//! The discovery driver, the discovery table's advertisement, browse and +//! dials behind the crate's `discovery` feature. + +use mdns_sd::{ + DaemonEvent, IfKind, MDNS_PORT, ResolvedService, ServiceDaemon, ServiceEvent, ServiceInfo, +}; +use parking_lot::Mutex; +use ring::rand::{SecureRandom, SystemRandom}; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Instant; +use tokio::sync::mpsc; +use tracing::warn; + +use connetto_core::device_cert::DeviceIdentity; + +use crate::CloseReason; +use crate::LinkError; +use crate::Node; +use crate::PeerEvent; +use crate::fingerprint::Fingerprint; +use crate::frame::PROTOCOL_VERSION; +use crate::policy::{Action, DialOutcome, Policy}; + +/// The service type discovery announces under. +const SERVICE_TYPE: &str = "_connetto-peer._tcp.local."; +/// The TXT key the frame version rides under. +const TXT_VERSION: &str = "v"; +/// The TXT key the leaf fingerprint rides under. +const TXT_FINGERPRINT: &str = "fp"; + +/// What discovery tells the device about the instances it sees. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DiscoveryEvent { + /// An instance was found or resolved. + Found { + /// The resolved address to dial. + address: SocketAddr, + /// The instance's leaf fingerprint. + fingerprint: Fingerprint, + }, + /// An instance was removed. + Gone { + /// The instance's leaf fingerprint. + fingerprint: Fingerprint, + }, +} + +/// The discovery driver, one device's side of the browse. +pub struct Discovery { + /// The node the found instances dial. + node: Node, + /// Whether the device dials what it finds. + autolink: bool, + /// The commands the driver obeys. + tx: mpsc::UnboundedSender, + /// The runner's command intake, until the first serve takes it. + rx: Mutex>>, + /// What the runner tells the device. + events: mpsc::UnboundedSender, + /// The driver, once the first serve has spawned the runner. + running: AtomicBool, + /// The mDNS port the daemon binds. + mdns_port: u16, + /// The daemon restricted to the loopback, where the proofs run. + loopback_only: bool, +} + +impl std::fmt::Debug for Discovery { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Discovery").finish_non_exhaustive() + } +} + +/// A command for the discovery's runner. +enum Command { + /// The standing serves at `port` under `own`, or a renewal moved either. + Serve { port: u16, own: Fingerprint }, + /// The standing left serving, so the driver stops and forgets. + Stop, + /// A node link event, mapped to an instance by the peer's leaf. + NodeEvent { + /// The peer the link reached or fell from. + peer: DeviceIdentity, + /// The close reason, the link fell, else the link reached. + reason: Option, + }, + /// A dial the runner started ended. + DialOutcome { + /// The dialled instance. + fp: Fingerprint, + /// How the dial ended. + outcome: DialOutcome, + }, + /// A retry timer woke, so the due instances dial. + RetryDue, +} + +impl Discovery { + /// Drive discovery for `node`, dialing what it finds when `autolink`, + /// telling `events` what it learns. + pub fn new(node: Node, autolink: bool, events: mpsc::UnboundedSender) -> Self { + let (tx, rx) = mpsc::unbounded_channel(); + Self { + node, + autolink, + tx, + rx: Mutex::new(Some(rx)), + events, + running: AtomicBool::new(false), + mdns_port: MDNS_PORT, + loopback_only: false, + } + } + + /// The mDNS port the daemon binds, to clear the system daemon's in the + /// proofs. + #[must_use] + pub fn with_mdns_port(mut self, port: u16) -> Self { + self.mdns_port = port; + self + } + + /// Restrict the daemon to the loopback interface, where the proofs run. + #[must_use] + pub fn loopback_only(mut self) -> Self { + self.loopback_only = true; + self + } + + /// Start the advertisement and the browse at the node's bound `port` + /// under `own`, or re-register when a renewal moved either. + pub fn serve(&self, port: u16, own: Fingerprint) { + let started = self.running.fetch_or(true, Ordering::AcqRel); + if !started { + let rx = self.rx.lock().take(); + match (rx, tokio::runtime::Handle::try_current()) { + (Some(rx), Ok(handle)) => { + handle.spawn(run( + self.node.clone(), + self.autolink, + self.mdns_port, + self.loopback_only, + rx, + self.tx.downgrade(), + self.events.clone(), + )); + } + (Some(rx), Err(_)) => { + self.running.store(false, Ordering::Release); + *self.rx.lock() = Some(rx); + warn!("no runtime drives the discovery, so it waits"); + return; + } + (None, _) => return, + } + } + let _ = self.tx.send(Command::Serve { port, own }); + } + + /// Stop the advertisement and the browse and forget every instance. + pub fn stop(&self) { + let _ = self.tx.send(Command::Stop); + } + + /// Hand a node link event to the policy, which maps it by the peer's + /// leaf. + pub fn on_node_event(&self, event: &PeerEvent) { + let command = match event { + PeerEvent::Linked { peer } => Command::NodeEvent { + peer: peer.clone(), + reason: None, + }, + PeerEvent::Unlinked { peer, reason } => Command::NodeEvent { + peer: peer.clone(), + reason: Some(*reason), + }, + PeerEvent::ListReceived { .. } => return, + }; + let _ = self.tx.send(command); + } +} + +/// The driver until the standing leaves, the one policy per standing. +/// +/// The policy outlives a local address change, which re-browses under the +/// same standing, and is dropped when the standing leaves. The commands are +/// held weak, so the runner ends when the last driver reference is gone. +async fn run( + node: Node, + autolink: bool, + mdns_port: u16, + loopback_only: bool, + mut commands: mpsc::UnboundedReceiver, + commands_tx: mpsc::WeakUnboundedSender, + events: mpsc::UnboundedSender, +) { + loop { + let (port, own) = match commands.recv().await { + Some(Command::Serve { port, own }) => (port, own), + // Nothing serves, so the events nothing serves wait. + Some(Command::NodeEvent { .. } | Command::DialOutcome { .. } | Command::RetryDue) => { + continue; + } + // A stop while idle is the standing's own, and the driver waits + // for its next serve. + Some(Command::Stop) => continue, + // The last driver reference is gone, so the runner ends with it. + None => return, + }; + // A new standing, a new policy, until the standing leaves. + let mut policy = Policy::new(own, autolink); + let mut current = (port, own); + loop { + let re_browse = serve_round( + &node, + mdns_port, + loopback_only, + &mut commands, + &commands_tx, + &events, + &mut policy, + &mut current, + ) + .await; + if !re_browse { + break; + } + // The local addresses changed, so the same standing and the same + // policy browse again. + } + } +} + +/// One round of the browse, until the standing leaves or a re-browse is due. +/// +/// Answers `true` when the standing holds and a re-browse is due. +#[expect( + clippy::too_many_arguments, + reason = "each input is a distinct channel or config" +)] +#[expect(clippy::too_many_lines, reason = "the select! loop is one unit")] +async fn serve_round( + node: &Node, + mdns_port: u16, + loopback_only: bool, + commands: &mut mpsc::UnboundedReceiver, + commands_tx: &mpsc::WeakUnboundedSender, + events: &mpsc::UnboundedSender, + policy: &mut Policy, + current: &mut (u16, Fingerprint), +) -> bool { + let (port, own) = *current; + let daemon = match ServiceDaemon::new_with_port(mdns_port) { + Ok(daemon) => daemon, + Err(err) => { + warn!(%err, "the mDNS daemon will not start, so discovery waits"); + return false; + } + }; + if loopback_only { + let loopback = (|| { + daemon.disable_interface(IfKind::All)?; + daemon.enable_interface(IfKind::LoopbackV4) + })(); + if let Err(err) = loopback { + warn!(%err, "the loopback will not select, so discovery waits"); + return false; + } + } + let browse = match daemon.browse(SERVICE_TYPE) { + Ok(browse) => browse, + Err(err) => { + warn!(%err, "the browse will not start, so discovery waits"); + return false; + } + }; + let monitor = match daemon.monitor() { + Ok(monitor) => monitor, + Err(err) => { + warn!(%err, "the monitor will not start, so discovery waits"); + return false; + } + }; + let mut names: HashMap = HashMap::new(); + let mut registered: Option = None; + if let Some(fullname) = register(&daemon, own, port) { + registered = Some(fullname); + } else { + warn!("the advertisement will not register, so the browse alone serves"); + } + let re_browse = { + loop { + tokio::select! { + command = commands.recv() => { + let Some(command) = command else { + break false; + }; + match command { + Command::Serve { port, own } => { + // A renewal re-registers under a new name and + // fingerprint, and the advertisement follows the + // port. + if (port, own) != *current { + re_register(&daemon, &mut registered, policy, port, own); + *current = (port, own); + } + } + Command::Stop => break false, + Command::NodeEvent { peer, reason } => { + let Some(fp) = node.peer_fingerprint(&peer) else { + continue; + }; + match reason { + None => policy.linked(fp), + Some(reason) => policy.unlinked(Instant::now(), fp, reason), + }; + arm_retry(commands_tx, policy, fp); + } + // The dial's truth is the link's, so a late outcome + // lands on the state the machine holds. + Command::DialOutcome { fp, outcome } => { + let now = Instant::now(); + policy.dial_result(now, fp, outcome); + arm_retry(commands_tx, policy, fp); + // The ended dial frees its place, and the due + // dials take their turn. + for queued in policy.due(now) { + spawn_dial(node, commands_tx, queued, policy.addresses(&queued)); + } + } + Command::RetryDue => { + let now = Instant::now(); + for fp in policy.due(now) { + spawn_dial(node, commands_tx, fp, policy.addresses(&fp)); + } + } + } + } + event = browse.recv_async() => { + let Ok(event) = event else { + warn!("the browse channel closed, so the browse ends"); + break false; + }; + match event { + ServiceEvent::ServiceResolved(resolved) => { + let Some(fp) = fingerprint_of(&resolved) else { + continue; + }; + let name = resolved.fullname.clone(); + names.insert(name.clone(), fp); + let addresses: Vec = resolved + .addresses + .iter() + .map(|ip| SocketAddr::new(ip.to_ip_addr(), resolved.port)) + .collect(); + let mut actions = Vec::new(); + for address in addresses { + actions.append(&mut policy.found(&name, fp, address)); + } + apply(&actions, fp, node, commands_tx, events, policy); + } + ServiceEvent::ServiceRemoved(_service_type, fullname) => { + let Some(fp) = names.remove(&fullname) else { + continue; + }; + // The peer may have re-advertised under a fresh + // name before this goodbye, so only the last + // name's removal reports the peer gone. + let last_name = !names.values().any(|still| *still == fp); + let actions = if last_name { + policy.removed(fp) + } else { + Vec::new() + }; + apply(&actions, fp, node, commands_tx, events, policy); + } + // The search's own bookkeeping, which the browse owns. + _ => {} + } + } + event = monitor.recv_async() => { + match event { + // The host's addresses moved, so the same standing + // browses again, the policy holding. + Ok(DaemonEvent::IpAdd(_) | DaemonEvent::IpDel(_)) => { + policy.addresses_changed(); + break true; + } + Ok(DaemonEvent::Error(err)) => { + warn!(%err, "the mDNS daemon met an error"); + } + Ok(_) => {} + Err(_) => { + warn!("the monitor channel closed, so the browse ends"); + break false; + } + } + } + } + } + }; + if let Some(fullname) = registered { + let _ = daemon.unregister(&fullname); + } + let _ = daemon.shutdown(); + re_browse +} + +/// Register the advertisement under a fresh name at `port` carrying `own`. +fn register(daemon: &ServiceDaemon, own: Fingerprint, port: u16) -> Option { + let name = random_name(); + // The SRV target is the instance's own hostname, the one name + // `mdns-sd` accepts for a `.local.` host. + let host = format!("{name}.local."); + let version = PROTOCOL_VERSION.to_string(); + let fingerprint = own.to_string(); + let props = [ + (TXT_VERSION, version.as_str()), + (TXT_FINGERPRINT, fingerprint.as_str()), + ]; + let info = ServiceInfo::new(SERVICE_TYPE, &name, &host, (), port, &props[..]) + .expect("the service names hold") + .enable_addr_auto(); + let fullname = info.get_fullname().to_string(); + daemon.register(info).ok().map(|()| fullname) +} + +/// Drop the old registration and register the renewal, moving the policy's +/// own fingerprint with it. +fn re_register( + daemon: &ServiceDaemon, + registered: &mut Option, + policy: &mut Policy, + port: u16, + own: Fingerprint, +) { + if let Some(fullname) = registered.take() { + let _ = daemon.unregister(fullname.as_str()); + } + policy.set_own(own); + if let Some(fullname) = register(daemon, own, port) { + *registered = Some(fullname); + } else { + warn!("the renewed advertisement will not register"); + } +} + +/// The actions a policy move left, told to the device and dialled. +fn apply( + actions: &[Action], + fp: Fingerprint, + node: &Node, + commands_tx: &mpsc::WeakUnboundedSender, + events: &mpsc::UnboundedSender, + policy: &Policy, +) { + for action in actions { + match action { + Action::Found(address) => { + let _ = events.send(DiscoveryEvent::Found { + address: *address, + fingerprint: fp, + }); + } + Action::Gone => { + let _ = events.send(DiscoveryEvent::Gone { fingerprint: fp }); + } + Action::Dial => { + spawn_dial(node, commands_tx, fp, policy.addresses(&fp)); + } + // The browse owns its own restart, the monitor's branch. + Action::Browse => {} + } + } +} + +/// The timer that wakes the runner when the instance's retry is due. A +/// replacement deadline leaves the old timer behind, and the late wake finds +/// the dial already in flight. +fn arm_retry(commands_tx: &mpsc::WeakUnboundedSender, policy: &Policy, fp: Fingerprint) { + if let Some(at) = policy.retry_at(&fp) { + let commands_tx = commands_tx.clone(); + let at = tokio::time::Instant::from_std(at); + tokio::spawn(async move { + tokio::time::sleep_until(at).await; + // The driver is gone, so the timer it served is done. + let Some(commands_tx) = commands_tx.upgrade() else { + return; + }; + let _ = commands_tx.send(Command::RetryDue); + }); + } +} + +/// A dial the policy started, over the instance's addresses `IPv4` first. +fn spawn_dial( + node: &Node, + commands_tx: &mpsc::WeakUnboundedSender, + fp: Fingerprint, + addresses: Vec, +) { + let node = node.clone(); + let commands_tx = commands_tx.clone(); + tokio::spawn(async move { + let outcome = dial(&node, addresses).await; + // The driver is gone, so its dial's result lands nowhere. + let Some(commands_tx) = commands_tx.upgrade() else { + return; + }; + let _ = commands_tx.send(Command::DialOutcome { fp, outcome }); + }); +} + +/// Dial the instance's addresses, `IPv4` first, until one answers. +async fn dial(node: &Node, addresses: Vec) -> DialOutcome { + let mut ordered: Vec = Vec::new(); + let mut rest: Vec = Vec::new(); + for address in &addresses { + if address.is_ipv4() { + ordered.push(*address); + } else { + rest.push(*address); + } + } + ordered.extend(rest); + for address in ordered { + let linked = match node.link(address).await { + Ok(_) => true, + Err(LinkError::Unreachable(_) | LinkError::Timeout | LinkError::NotServing) => false, + Err(_) => return DialOutcome::Refused, + }; + if linked { + return DialOutcome::Linked; + } + } + DialOutcome::Unreachable +} + +/// The random instance name a registration announces under. +fn random_name() -> String { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let random = SystemRandom::new(); + let mut bytes = [0u8; 8]; + random.fill(&mut bytes).expect("the system random fills"); + bytes + .iter() + .fold(String::with_capacity(16), |mut name, byte| { + name.push(char::from(HEX[usize::from(*byte >> 4)])); + name.push(char::from(HEX[usize::from(*byte & 0xf)])); + name + }) +} + +/// The fingerprint a hex TXT value carries, when it carries one. +fn fingerprint_text(text: &str) -> Option { + // A fingerprint is exactly 64 hex characters, and a shorter value is a + // different record, not a truncated one. + if text.len() != 64 { + return None; + } + let mut bytes = [0u8; 32]; + for (byte, pair) in bytes.iter_mut().zip(text.as_bytes().chunks(2)) { + *byte = u8::from_str_radix(std::str::from_utf8(pair).ok()?, 16).ok()?; + } + Some(Fingerprint::new(bytes)) +} + +/// The fingerprint a resolved instance carries, when it carries one. +fn fingerprint_of(resolved: &ResolvedService) -> Option { + let value = resolved + .txt_properties + .get_property_val(TXT_FINGERPRINT) + .and_then(|value| value)?; + let text = std::str::from_utf8(value).ok()?; + fingerprint_text(text) +} + +#[cfg(test)] +mod tests { + use super::fingerprint_text; + use crate::fingerprint::Fingerprint; + + /// The TXT fingerprint is exactly 64 hex characters, or nothing. + #[test] + fn a_fingerprint_is_exactly_sixty_four_hex_characters() { + let fingerprint = Fingerprint::new([0xab; 32]); + let hex = fingerprint.to_string(); + assert_eq!( + fingerprint_text(&hex), + Some(fingerprint), + "the full value holds" + ); + assert_eq!( + fingerprint_text(&hex[..62]), + None, + "a short value is refused" + ); + assert_eq!( + fingerprint_text(&hex[..63]), + None, + "an odd length is refused" + ); + assert_eq!( + fingerprint_text(&format!("zz{}", &hex[2..])), + None, + "a non-hex value is refused" + ); + } +} diff --git a/crates/connetto-peer/src/discovery_proofs.rs b/crates/connetto-peer/src/discovery_proofs.rs new file mode 100644 index 00000000..2efb554f --- /dev/null +++ b/crates/connetto-peer/src/discovery_proofs.rs @@ -0,0 +1,321 @@ +//! The R76 slice 3 loopback proofs, every advertisement and browse running +//! on the loopback. +//! +//! Each proof first probes that loopback multicast works, and skips itself +//! when it does not, so a host without loopback multicast never reads as a +//! discovery failure. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use connetto_core::device_cert::{AttestationLevel, DeviceIdentity}; +use tokio::sync::mpsc; +use tokio::time; + +use crate::Discovery; +use crate::DiscoveryEvent; +use crate::Fingerprint; +use crate::PeerEvent; +use crate::test_support::loopback_multicast; +use crate::tests::{Deployment, Peer, events, node, serve, whole_second}; + +/// A loopback port the proof's mDNS daemons bind, clear of the system +/// daemon's and of every other proof's. +fn fresh_mdns_port() -> u16 { + std::net::UdpSocket::bind("127.0.0.1:0") + .expect("the loopback binds") + .local_addr() + .expect("the bound address") + .port() +} +const DAY: Duration = Duration::from_hours(24); +const WAIT: Duration = Duration::from_secs(30); + +/// The node's link events handed to the discovery and teed to `out`, until +/// the node's channel ends. +async fn forward( + discovery: Arc, + mut node_events: mpsc::UnboundedReceiver, + out: mpsc::UnboundedSender, +) { + while let Some(event) = node_events.recv().await { + discovery.on_node_event(&event); + let _ = out.send(event); + } +} + +/// Wait for the instance `fingerprint` to be found, answering its address. +async fn await_found( + rx: &mut mpsc::UnboundedReceiver, + fingerprint: Fingerprint, +) -> SocketAddr { + loop { + let event = time::timeout(WAIT, rx.recv()) + .await + .expect("discovery reports within the bound") + .expect("the channel stays open"); + match event { + DiscoveryEvent::Found { + address, + fingerprint: found, + } if found == fingerprint => { + return address; + } + _ => {} + } + } +} + +/// Wait for the link to `peer` to land. +async fn await_linked(rx: &mut mpsc::UnboundedReceiver, peer: &DeviceIdentity) { + loop { + let event = time::timeout(WAIT, rx.recv()) + .await + .expect("the link lands within the bound") + .expect("the channel stays open"); + match event { + PeerEvent::Linked { peer: linked } if &linked == peer => return, + _ => {} + } + } +} + +/// Mint one deployment, one issuer, and the two devices a proof links. +fn pair(now: std::time::SystemTime) -> (Deployment, Peer, Peer) { + let deployment = Deployment::new(1, now); + let issuer = deployment.add_issuer(now, [1; 16]); + let alice = deployment.device( + &issuer, + "alice", + now, + DAY, + [1; 16], + AttestationLevel::Unproven, + ); + let bob = deployment.device( + &issuer, + "bob", + now, + DAY, + [2; 16], + AttestationLevel::Unproven, + ); + (deployment, alice, bob) +} + +/// A node and its discovery, served on loopback, the events teed out. +fn standing( + deployment: &Deployment, + peer: &crate::tests::Peer, + autolink: bool, + port: u16, +) -> ( + crate::Node, + Arc, + SocketAddr, + mpsc::UnboundedReceiver, + mpsc::UnboundedReceiver, +) { + let (tx, rx) = events(); + let node = node(deployment.root_der(), tx); + let (discovery_tx, discovery_rx) = mpsc::unbounded_channel(); + let discovery = Arc::new( + Discovery::new(node.clone(), autolink, discovery_tx) + .with_mdns_port(port) + .loopback_only(), + ); + let addr = serve(&node, peer); + let (tee_tx, tee_rx) = mpsc::unbounded_channel(); + let forwarder = discovery.clone(); + tokio::spawn(forward(forwarder, rx, tee_tx)); + (node, discovery, addr, discovery_rx, tee_rx) +} + +/// Proof 1. +#[tokio::test] +async fn two_nodes_discover_each_other_and_autolink() { + if !loopback_multicast() { + eprintln!("the loopback multicast probe failed, so the proof skips"); + return; + } + eprintln!("R76-PROOF-RAN two_nodes_discover_each_other_and_autolink"); + let (deployment, alice, bob) = pair(whole_second()); + let port = fresh_mdns_port(); + let (alice_node, da, a_addr, mut da_rx, mut a_events) = + standing(&deployment, &alice, true, port); + let (bob_node, db, b_addr, mut db_rx, mut b_events) = standing(&deployment, &bob, true, port); + + let alice_fp = Fingerprint::of(&alice.leaf); + let bob_fp = Fingerprint::of(&bob.leaf); + da.serve(a_addr.port(), alice_fp); + db.serve(b_addr.port(), bob_fp); + + // Each side reports the other at its bound address, and the dial lands. + assert_eq!(await_found(&mut da_rx, bob_fp).await, b_addr); + assert_eq!(await_found(&mut db_rx, alice_fp).await, a_addr); + await_linked(&mut a_events, &bob.identity).await; + await_linked(&mut b_events, &alice.identity).await; + + // The node maps the live link to the instance's fingerprint. + assert_eq!(alice_node.peer_fingerprint(&bob.identity), Some(bob_fp)); + assert_eq!(bob_node.peer_fingerprint(&alice.identity), Some(alice_fp)); + + da.stop(); + db.stop(); +} + +/// Proof 3. +#[tokio::test] +async fn a_renewal_re_registers_under_a_new_fingerprint() { + if !loopback_multicast() { + eprintln!("the loopback multicast probe failed, so the proof skips"); + return; + } + eprintln!("R76-PROOF-RAN a_renewal_re_registers_under_a_new_fingerprint"); + let now = whole_second(); + let deployment = Deployment::new(1, now); + let issuer = deployment.add_issuer(now, [1; 16]); + let alice = deployment.device( + &issuer, + "alice", + now, + DAY, + [1; 16], + AttestationLevel::Unproven, + ); + let bob = deployment.device( + &issuer, + "bob", + now, + DAY, + [2; 16], + AttestationLevel::Unproven, + ); + let renewed = deployment.reissue(&issuer, &alice, now, DAY, [3; 16]); + + let port = fresh_mdns_port(); + let (alice_node, da, a_addr, mut da_rx, mut a_events) = + standing(&deployment, &alice, true, port); + let (bob_node, db, b_addr, mut db_rx, mut b_events) = standing(&deployment, &bob, true, port); + + let alice_fp = Fingerprint::of(&alice.leaf); + let bob_fp = Fingerprint::of(&bob.leaf); + let renewed_fp = Fingerprint::of(&renewed.leaf); + da.serve(a_addr.port(), alice_fp); + db.serve(b_addr.port(), bob_fp); + + // They find each other and link under the old fingerprint. + await_found(&mut da_rx, bob_fp).await; + await_found(&mut db_rx, alice_fp).await; + await_linked(&mut a_events, &bob.identity).await; + await_linked(&mut b_events, &alice.identity).await; + + // The renewal keeps the port and the link, and the advertisement moves + // to the new fingerprint. + alice_node + .serve(a_addr, renewed.identity()) + .expect("the renewal serves"); + da.serve(a_addr.port(), renewed_fp); + + // Bob's browse sees the old instance gone and the new one found, in + // either order, and the link's renewal reaches his node. + let mut gone = false; + let mut found = None; + while !gone || found.is_none() { + let event = time::timeout(WAIT, db_rx.recv()) + .await + .expect("discovery reports within the bound") + .expect("the channel stays open"); + match event { + DiscoveryEvent::Gone { fingerprint: g } if g == alice_fp => gone = true, + DiscoveryEvent::Found { + address, + fingerprint: f, + } if f == renewed_fp => found = Some(address), + _ => {} + } + } + assert_eq!(found, Some(a_addr)); + let deadline = Instant::now() + WAIT; + loop { + if bob_node.peer_fingerprint(&alice.identity) == Some(renewed_fp) { + break; + } + assert!( + Instant::now() < deadline, + "the renewal reaches the link within the bound" + ); + time::sleep(Duration::from_millis(50)).await; + } + + da.stop(); + db.stop(); +} + +/// Proof 5. +#[tokio::test] +async fn dropping_the_driver_ends_its_runner() { + let (deployment, alice, _bob) = pair(whole_second()); + let (alice_events, _) = events(); + let alice_node = node(deployment.root_der(), alice_events); + let a_addr = serve(&alice_node, &alice); + let (da_tx, mut da_rx) = mpsc::unbounded_channel(); + let da = Discovery::new(alice_node, false, da_tx) + .with_mdns_port(fresh_mdns_port()) + .loopback_only(); + let fp = Fingerprint::of(&alice.leaf); + da.serve(a_addr.port(), fp); + // Dropping the driver ends its runner, and the event channel closes with + // it. + drop(da); + let closed = time::timeout(Duration::from_secs(5), da_rx.recv()).await; + assert!( + matches!(closed, Ok(None)), + "dropping the driver ends its runner" + ); +} + +/// Proof 6. +#[tokio::test] +async fn a_stopped_discovery_serves_again() { + if !loopback_multicast() { + eprintln!("the loopback multicast probe failed, so the proof skips"); + return; + } + eprintln!("R76-PROOF-RAN a_stopped_discovery_serves_again"); + let (deployment, alice, bob) = pair(whole_second()); + let port = fresh_mdns_port(); + + // The advertiser, without the tee, so the proof can drop it. + let (alice_events, _) = events(); + let alice_node = node(deployment.root_der(), alice_events); + let a_addr = serve(&alice_node, &alice); + let (da_tx, _da_rx) = mpsc::unbounded_channel(); + let da = Discovery::new(alice_node.clone(), false, da_tx) + .with_mdns_port(port) + .loopback_only(); + + // The browser. + let (bob_events, _) = events(); + let bob_node = node(deployment.root_der(), bob_events); + let b_addr = serve(&bob_node, &bob); + let (db_tx, mut db_rx) = mpsc::unbounded_channel(); + let db = Discovery::new(bob_node.clone(), false, db_tx) + .with_mdns_port(port) + .loopback_only(); + + let alice_fp = Fingerprint::of(&alice.leaf); + let bob_fp = Fingerprint::of(&bob.leaf); + da.serve(a_addr.port(), alice_fp); + db.serve(b_addr.port(), bob_fp); + await_found(&mut db_rx, alice_fp).await; + da.stop(); + + // A serve after a stop advertises again, under a fresh name. + da.serve(a_addr.port(), alice_fp); + await_found(&mut db_rx, alice_fp).await; + + da.stop(); + db.stop(); +} diff --git a/crates/connetto-peer/src/fingerprint.rs b/crates/connetto-peer/src/fingerprint.rs new file mode 100644 index 00000000..5a442a4d --- /dev/null +++ b/crates/connetto-peer/src/fingerprint.rs @@ -0,0 +1,64 @@ +//! The leaf fingerprint the discovery TXT record and events carry. + +use std::fmt; + +/// The SHA-256 of a presented leaf, the only thing discovery names about a +/// device. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct Fingerprint([u8; 32]); + +impl Fingerprint { + /// Wrap `bytes`. + pub const fn new(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + /// Hash `leaf_der`, the presented leaf. + /// + /// # Panics + /// + /// Never, a SHA-256 digest is 32 bytes. + pub fn of(leaf_der: &[u8]) -> Self { + let digest = ring::digest::digest(&ring::digest::SHA256, leaf_der); + Self( + digest + .as_ref() + .try_into() + .expect("a sha-256 digest is 32 bytes"), + ) + } + + /// The 32 bytes. + pub const fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } +} + +impl fmt::Display for Fingerprint { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + for byte in self.0 { + write!(f, "{byte:02x}")?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::Fingerprint; + + #[test] + fn the_hash_of_a_leaf_is_its_lower_hex_display() { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let fingerprint = Fingerprint::of(b"the leaf"); + let hex = fingerprint + .as_bytes() + .iter() + .fold(String::new(), |mut out, byte| { + out.push(char::from(HEX[usize::from(*byte >> 4)])); + out.push(char::from(HEX[usize::from(*byte & 0xf)])); + out + }); + assert_eq!(fingerprint.to_string(), hex); + } +} diff --git a/crates/connetto-peer/src/lib.rs b/crates/connetto-peer/src/lib.rs index ccb33a3e..0399fc9e 100644 --- a/crates/connetto-peer/src/lib.rs +++ b/crates/connetto-peer/src/lib.rs @@ -17,11 +17,29 @@ mod node; mod signer; mod verify; +#[cfg(feature = "discovery")] +mod discovery; +#[cfg(feature = "discovery")] +mod fingerprint; +#[cfg(feature = "discovery")] +mod policy; + #[cfg(test)] mod tests; +#[cfg(all(test, feature = "discovery"))] +mod discovery_proofs; + +#[cfg(any(test, feature = "test-support"))] +pub mod test_support; + pub use error::{CloseReason, LinkError, Refusal, TrustError}; pub use event::PeerEvent; pub use frame::PeerFrame; pub use identity::{Clock, Identity, SystemClock, Trust}; pub use node::{Liveness, Node}; + +#[cfg(feature = "discovery")] +pub use discovery::{Discovery, DiscoveryEvent}; +#[cfg(feature = "discovery")] +pub use fingerprint::Fingerprint; diff --git a/crates/connetto-peer/src/link.rs b/crates/connetto-peer/src/link.rs index 9288a4f6..99a137f8 100644 --- a/crates/connetto-peer/src/link.rs +++ b/crates/connetto-peer/src/link.rs @@ -221,6 +221,12 @@ fn renew( { slot.leaf = leaf.to_vec(); slot.issuer = issuer.to_vec(); + // The renewed chain changes the leaf discovery fingerprints. + #[cfg(feature = "discovery")] + state + .peer_fingerprints + .lock() + .insert(key, crate::fingerprint::Fingerprint::of(leaf)); } Ok(()) } diff --git a/crates/connetto-peer/src/node.rs b/crates/connetto-peer/src/node.rs index 72d3081e..fb93d432 100644 --- a/crates/connetto-peer/src/node.rs +++ b/crates/connetto-peer/src/node.rs @@ -29,6 +29,9 @@ use crate::identity::{Clock, Identity, Trust}; use crate::signer::{IdentityClientCert, IdentityServerCert}; use crate::verify::{Crl, KeptList, PeerVerifier}; +#[cfg(feature = "discovery")] +use crate::fingerprint::Fingerprint; + /// The time a dial gives its connect. const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); /// The time a dial gives the handshake. @@ -41,6 +44,9 @@ pub(crate) const WRITE_BOUND: Duration = Duration::from_secs(10); pub(crate) const PING_EVERY: Duration = Duration::from_secs(15); /// The silence that takes a link down. pub(crate) const SILENCE_LIMIT: Duration = Duration::from_secs(45); +/// The inbound handshakes the listener runs at once, the unauthenticated +/// network's bound on its sockets. +const MAX_INBOUND_HANDSHAKES: usize = 16; const DIAL_NAME: &str = "connetto-peer"; /// The links' liveness bounds, the ping pace and the silence that takes a @@ -85,6 +91,10 @@ pub(crate) struct NodeState { pub(crate) events: mpsc::UnboundedSender, pub(crate) ping_every: Duration, pub(crate) silence_limit: Duration, + /// The leaf fingerprints the links carry, per peer key, for discovery's + /// mapping of the node's events to its instances. + #[cfg(feature = "discovery")] + pub(crate) peer_fingerprints: Mutex>, } /// One entry per peer, under the node's single links lock. @@ -235,6 +245,8 @@ impl Node { events, ping_every: liveness.ping_every, silence_limit: liveness.silence_limit, + #[cfg(feature = "discovery")] + peer_fingerprints: Mutex::new(HashMap::new()), }); Ok(Self { _lifetime: Arc::new(Lifetime(Arc::clone(&state))), @@ -287,9 +299,13 @@ impl Node { /// Hand the node's current chain to the live links, which verify it as at /// a handshake and move their expiry deadline. fn renew_links(&self) { - let identity_slot = self.state.identity.read(); - let held = identity_slot.as_ref().expect("serving holds an identity"); + // The links lock first, then the identity, the order a registration + // takes too, so a link registering during the swap is never missed. let links = self.state.links.lock(); + let identity_slot = self.state.identity.read(); + let Some(held) = identity_slot.as_ref() else { + return; + }; for slot_state in links.values() { if let SlotState::Live(slot) = slot_state { let _ = slot.command.send(LinkCommand::Certificate { @@ -311,6 +327,18 @@ impl Node { self.state.listener.read().as_ref().map(|slot| slot.addr) } + /// The fingerprint the live link to `peer` carries, for discovery's + /// mapping of the node's events to its instances. + #[cfg(feature = "discovery")] + #[must_use] + pub fn peer_fingerprint(&self, peer: &DeviceIdentity) -> Option { + self.state + .peer_fingerprints + .lock() + .get(&peer.key()) + .copied() + } + /// Dial `addr`, answering the peer's identity once the link is live. /// /// # Errors @@ -378,6 +406,7 @@ impl Node { numbers: peer_numbers, dialer: key_id(&*identity.key), dial, + presented: identity.certificate.clone(), }, tls, )?; @@ -602,16 +631,15 @@ fn map_tls(err: io::Error) -> LinkError { } /// A frame failure as the dial's typed error, surfacing the peer's TLS -/// alert when it reaches the dial on the first read. +/// alert when it reaches the dial on the first read, and a plain I/O +/// failure as `Unreachable`. fn map_frame(err: FrameError) -> LinkError { let message = err.to_string(); match err { - FrameError::Io(io_err) => { - if let Ok(rustls_err) = io_err.downcast::() { - return map_rustls(rustls_err); - } - LinkError::Protocol(message) - } + FrameError::Io(io_err) => match io_err.downcast::() { + Ok(rustls_err) => map_rustls(rustls_err), + Err(io_err) => LinkError::Unreachable(io_err), + }, _ => LinkError::Protocol(message), } } @@ -622,6 +650,7 @@ async fn accept_loop( listener: TcpListener, mut stop: watch::Receiver, ) { + let permits = Arc::new(tokio::sync::Semaphore::new(MAX_INBOUND_HANDSHAKES)); loop { tokio::select! { _ = stop.changed() => return, @@ -630,16 +659,26 @@ async fn accept_loop( Ok(pair) => pair, Err(err) => { debug!(%err, "the accept failed"); + // A descriptor shortage stays ready, so the loop + // backs off before the next poll. + tokio::time::sleep(Duration::from_millis(100)).await; continue; } }; + let Ok(permit) = Arc::clone(&permits).try_acquire_owned() else { + drop(tcp); + continue; + }; let identity = state.identity.read().clone(); let Some(identity) = identity else { drop(tcp); continue; }; let state = Arc::clone(&state); - tokio::spawn(inbound(state, tcp, identity)); + tokio::spawn(async move { + let _permit = permit; + inbound(state, tcp, identity).await; + }); } } } @@ -706,6 +745,7 @@ async fn inbound(state: Arc, tcp: TcpStream, identity: Identity) { numbers: peer_numbers, dialer, dial, + presented: identity.certificate.clone(), }, tls, ); @@ -739,12 +779,13 @@ async fn exchange_hello( let frame = tokio::time::timeout(HELLO_TIMEOUT, read_frame(io)) .await .map_err(|_| LinkError::Timeout)? - .map_err(map_frame)?; - let Some(PeerFrame::Hello { + .map_err(map_frame)? + .ok_or_else(|| LinkError::Unreachable(io::ErrorKind::UnexpectedEof.into()))?; + let PeerFrame::Hello { version, dial: peer_dial, numbers, - }) = frame + } = frame else { return Err(LinkError::Protocol( "the peer's first frame is not a hello".into(), @@ -799,6 +840,22 @@ struct Handshaken { dialer: KeyId, /// The dialer's dial counter. dial: u64, + /// The leaf this node presented at the handshake, DER. + presented: Vec, +} + +/// Send `slot` the presented chain when a renewal swapped it since the +/// handshake presented `presented`, the chain the link would have had from +/// the swap. Called under the links lock, the order `renew_links` takes. +fn catch_up_renewal(state: &NodeState, slot: &LinkSlot, presented: &[u8]) { + if let Some(held) = state.identity.read().as_ref() + && held.certificate != presented + { + let _ = slot.command.send(LinkCommand::Certificate { + leaf: held.certificate.clone(), + issuer: held.issuer.clone(), + }); + } } /// Register a completed link under the duplicate rule, spawning its frame @@ -821,6 +878,7 @@ fn register_link( numbers, dialer, dial, + presented, } = handshaken; let peer = &peer; let peer_key = peer.key(); @@ -856,6 +914,10 @@ fn register_link( } }; if keep_new { + // Discovery maps the node's events to instances by the peer's + // leaf, which the slot records. + #[cfg(feature = "discovery")] + let fingerprint = Fingerprint::of(&leaf); // Close the losing link, if the new one keeps the slot. Its frame // task sends the `Close` frame and exits on the command. if let Some(SlotState::Live(existing)) = links.get(&peer_key) { @@ -874,7 +936,10 @@ fn register_link( command: cmd_tx, task: None, }; + catch_up_renewal(state, &slot, &presented); links.insert(peer_key, SlotState::Live(slot)); + #[cfg(feature = "discovery")] + state.peer_fingerprints.lock().insert(peer_key, fingerprint); if fresh { state .events diff --git a/crates/connetto-peer/src/policy.rs b/crates/connetto-peer/src/policy.rs new file mode 100644 index 00000000..39bf798e --- /dev/null +++ b/crates/connetto-peer/src/policy.rs @@ -0,0 +1,924 @@ +//! The discovery policy, the discovery table as a pure state machine on an +//! injected clock. + +use std::collections::{BTreeMap, BTreeSet}; +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +use crate::CloseReason; +use crate::fingerprint::Fingerprint; + +/// The backoff a fresh retry starts at. +pub(crate) const FIRST_WAIT: Duration = Duration::from_secs(5); +/// The longest backoff a retry waits. +pub(crate) const MAX_WAIT: Duration = Duration::from_secs(300); +/// The dials the policy lets run at once, the unauthenticated network's +/// bound on its sockets. +pub(crate) const MAX_DIALS: usize = 8; +/// The instances the policy tracks at once, the unauthenticated network's +/// bound on its memory and its events. +pub(crate) const MAX_INSTANCES: usize = 256; + +/// The one state an instance is in, beside the device's standing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum InstanceState { + /// Found, no live link yet. + New, + /// A live link carries the instance's fingerprint. + Linked, + /// A retry is pending. + Waiting, + /// A refusal, until the instance's fingerprint changes. + Barred, +} + +/// One found instance, under its fingerprint. +#[derive(Debug)] +pub(crate) struct Instance { + /// The mDNS name the instance advertises under, fresh at every + /// registration. + name: String, + state: InstanceState, + /// The resolved addresses, dialled `IPv4` first. + addresses: std::collections::BTreeSet, + /// The addresses already reported as found. + reported: std::collections::BTreeSet, + /// The current backoff step, while a retry is pending. + wait: Duration, + /// When the pending retry is due, on a monotonic clock. + retry_at: Option, + /// The instance's place in the find order, for the bound's eviction. + seq: u64, +} + +/// What a dial ends with. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DialOutcome { + /// One of the instance's addresses answered and linked. + Linked, + /// No address answered, so the retry backs off. + Unreachable, + /// An address or the link refused, so the instance is barred. + Refused, +} + +/// The actions a policy move leaves the device to take. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum Action { + /// Report the found instance at its address. + Found(SocketAddr), + /// Report the instance gone. + Gone, + /// Dial the instance's addresses, `IPv4` first. + Dial, + /// Browse again, the local addresses changed. + Browse, +} + +/// The policy against the discovery table, pure on its inputs and the clock +/// its caller hands it. +#[derive(Debug)] +pub(crate) struct Policy { + /// The device's own fingerprint, the instances it ignores. + own: Fingerprint, + /// Whether the device dials what it finds. + autolink: bool, + instances: BTreeMap, + /// The next instance's place in the find order. + next_seq: u64, + /// The dials running at once, keyed by fingerprint, freed only at the + /// dial's outcome, whatever its instance did meanwhile. + in_flight: BTreeSet, +} + +/// A close reason that bars the instance until its fingerprint changes. +pub(crate) fn bars(reason: CloseReason) -> bool { + matches!( + reason, + CloseReason::PeerRevoked | CloseReason::PeerExpired | CloseReason::Protocol + ) +} + +impl Policy { + /// The policy for the standing that just became serving, the device's own + /// fingerprint `own`, dialing what it finds when `autolink`. + pub(crate) fn new(own: Fingerprint, autolink: bool) -> Self { + Self { + own, + autolink, + instances: BTreeMap::new(), + next_seq: 0, + in_flight: BTreeSet::new(), + } + } + + /// A renewal moved the device's own fingerprint, which the browse ignores. + pub(crate) fn set_own(&mut self, own: Fingerprint) { + self.own = own; + } + + /// The instance `name`, carrying `fp`, was found or resolved at `addr`. + pub(crate) fn found(&mut self, name: &str, fp: Fingerprint, addr: SocketAddr) -> Vec { + if fp == self.own { + return Vec::new(); + } + let mut actions = Vec::new(); + let fresh = !self.instances.contains_key(&fp); + // A find beyond the bound evicts the oldest instance the link does + // not carry, without an event, or takes nothing when every tracked + // instance carries a link. + if fresh && self.instances.len() >= MAX_INSTANCES { + let Some((oldest, _evicted)) = self + .instances + .iter() + .filter(|(_, entry)| entry.state != InstanceState::Linked) + .min_by_key(|(_, entry)| entry.seq) + else { + return actions; + }; + let oldest = *oldest; + self.instances.remove(&oldest); + } + let seq = self.next_seq; + if fresh { + self.next_seq += 1; + } + let capacity = MAX_DIALS.saturating_sub(self.in_flight.len()); + let entry = self.instances.entry(fp).or_insert_with(|| Instance { + name: name.to_string(), + state: InstanceState::New, + addresses: BTreeSet::new(), + reported: BTreeSet::new(), + wait: FIRST_WAIT, + retry_at: None, + seq, + }); + // A re-advertisement under a fresh name reports the peer again. + if entry.name != name { + entry.name = name.to_string(); + entry.reported.clear(); + } + // The address reports once, and the instance dials once from new, + // while the bound of running dials holds room. + if entry.reported.insert(addr) { + actions.push(Action::Found(addr)); + } + entry.addresses.insert(addr); + if entry.state == InstanceState::New + && self.autolink + && capacity > 0 + && !self.in_flight.contains(&fp) + { + self.in_flight.insert(fp); + actions.push(Action::Dial); + } + actions + } + + /// The instance `fp` was removed. + pub(crate) fn removed(&mut self, fp: Fingerprint) -> Vec { + match self.instances.remove(&fp) { + Some(_) => vec![Action::Gone], + None => Vec::new(), + } + } + + /// A live link now carries `fp`, by dial or inbound. + pub(crate) fn linked(&mut self, fp: Fingerprint) -> Vec { + if let Some(entry) = self.instances.get_mut(&fp) { + entry.state = InstanceState::Linked; + entry.retry_at = None; + entry.wait = FIRST_WAIT; + } + Vec::new() + } + + /// The last live link to the peer behind `fp` closed with `reason`. + pub(crate) fn unlinked( + &mut self, + now: Instant, + fp: Fingerprint, + reason: CloseReason, + ) -> Vec { + if let Some(entry) = self.instances.get_mut(&fp) { + if bars(reason) { + entry.state = InstanceState::Barred; + entry.retry_at = None; + } else if entry.state == InstanceState::Linked { + entry.state = InstanceState::Waiting; + entry.wait = FIRST_WAIT; + entry.retry_at = if self.autolink { + Some(now + FIRST_WAIT) + } else { + None + }; + } + } + Vec::new() + } + + /// A dial for `fp` ended with `outcome`. + pub(crate) fn dial_result( + &mut self, + now: Instant, + fp: Fingerprint, + outcome: DialOutcome, + ) -> Vec { + if let Some(entry) = self.instances.get_mut(&fp) { + match outcome { + DialOutcome::Linked => { + if entry.state != InstanceState::Barred { + entry.state = InstanceState::Linked; + entry.retry_at = None; + entry.wait = FIRST_WAIT; + } + } + DialOutcome::Unreachable => { + if matches!(entry.state, InstanceState::New | InstanceState::Waiting) { + entry.state = InstanceState::Waiting; + if self.autolink { + entry.retry_at = Some(now + entry.wait); + } + entry.wait = (entry.wait * 2).min(MAX_WAIT); + } + } + DialOutcome::Refused => { + if matches!(entry.state, InstanceState::New | InstanceState::Waiting) { + entry.state = InstanceState::Barred; + entry.retry_at = None; + } + } + } + } + // The outcome frees the dial's place, whatever its instance did. + self.in_flight.remove(&fp); + Vec::new() + } + + /// The local addresses changed, so the browse restarts. + pub(crate) fn addresses_changed(&mut self) -> Vec { + for entry in self.instances.values_mut() { + if entry.state == InstanceState::Waiting { + entry.state = InstanceState::New; + entry.retry_at = None; + entry.wait = FIRST_WAIT; + } + } + vec![Action::Browse] + } + + /// The instances whose dial starts now, the bound of running dials held. + pub(crate) fn due(&mut self, now: Instant) -> Vec { + let capacity = MAX_DIALS.saturating_sub(self.in_flight.len()); + if capacity == 0 { + return Vec::new(); + } + // The due instances in the order they were found, the oldest first. + let mut due = self + .instances + .iter() + .filter(|(fp, entry)| { + if self.in_flight.contains(fp) { + return false; + } + // A fresh found dials when its turn comes, the bound allowing. + match entry.state { + InstanceState::New => self.autolink, + InstanceState::Waiting => entry.retry_at.is_some_and(|at| at <= now), + InstanceState::Linked | InstanceState::Barred => false, + } + }) + .map(|(fp, entry)| (entry.seq, *fp)) + .collect::>(); + due.sort_unstable_by_key(|(seq, _)| *seq); + let mut started = Vec::new(); + for (_, fp) in due.into_iter().take(capacity) { + self.in_flight.insert(fp); + started.push(fp); + } + started + } + + /// The next retry deadline, for the proofs. + #[cfg(test)] + pub(crate) fn next_retry(&self) -> Option { + self.instances + .values() + .filter_map(|entry| entry.retry_at) + .min() + } + + /// The instance's state, for the proofs. + #[cfg(test)] + pub(crate) fn state(&self, fp: &Fingerprint) -> Option { + self.instances.get(fp).map(|entry| entry.state) + } + + /// The instance's resolved addresses, for a dial the proofs spawn. + pub(crate) fn addresses(&self, fp: &Fingerprint) -> Vec { + self.instances + .get(fp) + .map(|entry| entry.addresses.iter().copied().collect()) + .unwrap_or_default() + } + + /// The instance's pending retry deadline, for the runner's timer. + pub(crate) fn retry_at(&self, fp: &Fingerprint) -> Option { + self.instances.get(fp).and_then(|entry| entry.retry_at) + } +} + +#[cfg(test)] +mod tests { + use std::net::SocketAddr; + use std::sync::LazyLock; + use std::time::{Duration, Instant}; + + use crate::CloseReason; + + use super::{ + Action, DialOutcome, FIRST_WAIT, InstanceState, MAX_DIALS, MAX_INSTANCES, MAX_WAIT, Policy, + }; + use crate::fingerprint::Fingerprint; + + const OWN: Fingerprint = Fingerprint::new([0; 32]); + const PEER: Fingerprint = Fingerprint::new([1; 32]); + const OTHER: Fingerprint = Fingerprint::new([2; 32]); + const RENEWED: Fingerprint = Fingerprint::new([3; 32]); + const ADDR: SocketAddr = + SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 443); + const ADDR2: SocketAddr = + SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 444); + static BASE: LazyLock = LazyLock::new(Instant::now); + + /// The one fixed clock the proofs stand on. + fn now() -> Instant { + *BASE + } + + fn autolink() -> Policy { + Policy::new(OWN, true) + } + + /// Proof 1. + #[test] + fn the_own_fingerprint_is_ignored() { + let mut policy = autolink(); + assert!(policy.found("n", OWN, ADDR).is_empty(), "a self is ignored"); + assert_eq!(policy.state(&OWN), None); + // A renewal moves the own fingerprint, and the new one is ignored too. + policy.set_own(RENEWED); + assert_eq!(policy.found("n", RENEWED, ADDR), Vec::new()); + assert!(policy.removed(OWN).is_empty(), "a self never reports gone"); + } + + /// Proof 2. + #[test] + fn a_found_instance_is_reported_and_dialled() { + let mut policy = autolink(); + let actions = policy.found("n", PEER, ADDR); + assert_eq!( + actions, + vec![Action::Found(ADDR), Action::Dial], + "a found instance reports and dials" + ); + assert_eq!(policy.state(&PEER), Some(InstanceState::New)); + // A repeated resolve of the same address reports nothing and dials + // nothing, and a second address reports only itself. + assert_eq!(policy.found("n", PEER, ADDR), Vec::::new()); + assert_eq!(policy.found("n", PEER, ADDR2), vec![Action::Found(ADDR2)]); + assert_eq!( + policy.addresses(&PEER), + vec![ADDR, ADDR2], + "every resolved address is kept" + ); + } + + /// Proof 3. + #[test] + fn a_linked_instance_is_not_redialled() { + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + assert_eq!(policy.state(&PEER), Some(InstanceState::Linked)); + // The browse keeps resolving it, and nothing dials. + assert_eq!(policy.found("n", PEER, ADDR), Vec::::new()); + assert_eq!( + policy.found("n", PEER, ADDR2), + vec![Action::Found(ADDR2)], + "a new address reports while linked" + ); + } + + /// Proof 4. + #[test] + fn an_unreachable_dial_backs_off_to_the_cap() { + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + // The first dial ends, and the retry is due after the first wait. + let mut due_at = now(); + policy.dial_result(due_at, PEER, DialOutcome::Unreachable); + for wait in [5, 10, 20, 40, 80, 160, 300, 300] { + let wait = Duration::from_secs(wait); + let next = (wait * 2).min(MAX_WAIT); + due_at += wait; + let due = policy.due(due_at); + assert_eq!(due, vec![PEER], "the retry is due"); + policy.dial_result(due_at, PEER, DialOutcome::Unreachable); + assert_eq!(policy.state(&PEER), Some(InstanceState::Waiting)); + assert_eq!(policy.next_retry(), Some(due_at + next)); + } + assert_eq!(policy.state(&PEER), Some(InstanceState::Waiting)); + } + + /// Proof 5. + #[test] + fn a_refused_instance_is_barred_until_its_fingerprint_changes() { + let mut policy = autolink(); + // A dial the peer refuses. + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Refused); + assert_eq!(policy.state(&PEER), Some(InstanceState::Barred)); + assert!(!policy.found("n", PEER, ADDR).contains(&Action::Found(ADDR))); + assert!( + !policy.found("n", PEER, ADDR2).contains(&Action::Dial), + "a barred instance never dials" + ); + // A dial of the renewed fingerprint starts clean. + let actions = policy.found("n", RENEWED, ADDR); + assert_eq!(actions, vec![Action::Found(ADDR), Action::Dial]); + // A live link the kept list revokes, the other bar reasons included. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + for reason in [ + CloseReason::PeerRevoked, + CloseReason::PeerExpired, + CloseReason::Protocol, + ] { + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + policy.unlinked(now(), PEER, reason); + assert_eq!( + policy.state(&PEER), + Some(InstanceState::Barred), + "{reason:?} bars" + ); + assert!( + policy.due(now() + Duration::from_secs(3600)).is_empty(), + "a bar never retries" + ); + } + } + + /// Proof 6. + #[test] + fn a_lost_link_redials_after_five_seconds() { + for reason in [ + CloseReason::PeerLost, + CloseReason::Closed, + CloseReason::Duplicate, + ] { + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + policy.unlinked(now(), PEER, reason); + assert_eq!( + policy.state(&PEER), + Some(InstanceState::Waiting), + "{reason:?} redials" + ); + assert_eq!(policy.next_retry(), Some(now() + Duration::from_secs(5))); + assert_eq!(policy.due(now()), Vec::::new()); + assert_eq!(policy.due(now() + Duration::from_secs(5)), vec![PEER]); + } + } + + /// Proof 7. + #[test] + fn a_removal_drops_the_retry_and_keeps_the_live_link() { + // The retry, dropped with the entry. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + assert_eq!(policy.removed(PEER), vec![Action::Gone]); + assert_eq!(policy.state(&PEER), None); + assert!(policy.next_retry().is_none()); + // The live link: the entry goes, the link is the node's, untouched. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + assert_eq!(policy.removed(PEER), vec![Action::Gone]); + assert_eq!(policy.state(&PEER), None); + // A removal nothing found reports nothing. + assert_eq!(policy.removed(OTHER), Vec::new()); + } + + /// Proof 8. + #[test] + fn a_local_address_change_resets_the_waiting() { + // The waiting entry, dropped to new with its retry. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + assert_eq!(policy.addresses_changed(), vec![Action::Browse]); + assert_eq!(policy.state(&PEER), Some(InstanceState::New)); + assert!(policy.next_retry().is_none()); + // The linked entry, the link stays. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + policy.addresses_changed(); + assert_eq!(policy.state(&PEER), Some(InstanceState::Linked)); + // The barred entry, the bar stays. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Refused); + policy.addresses_changed(); + assert_eq!(policy.state(&PEER), Some(InstanceState::Barred)); + } + + /// Proof 9. + #[test] + fn an_inbound_link_moves_every_state_to_linked() { + for start in [ + (InstanceState::New, false), + (InstanceState::Waiting, true), + (InstanceState::Barred, true), + ] { + let (state, wait) = start; + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + match state { + InstanceState::New | InstanceState::Linked => {} + InstanceState::Waiting => { + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + } + InstanceState::Barred => { + policy.dial_result(now(), PEER, DialOutcome::Refused); + } + } + assert_eq!(policy.state(&PEER), Some(state)); + policy.linked(PEER); + assert_eq!( + policy.state(&PEER), + Some(InstanceState::Linked), + "an inbound link lands from {state:?}" + ); + if wait { + assert!( + policy.next_retry().is_none(), + "the retry is dropped with the link" + ); + } + } + // An inbound link from a peer the browse never found changes nothing. + let mut policy = autolink(); + assert_eq!(policy.linked(OTHER), Vec::new()); + assert_eq!(policy.state(&OTHER), None); + } + + /// Proof 10. + #[test] + fn leaving_the_serving_forgets_everything() { + // The standing leaves, the runner drops the policy, and the instances + // go with it. The proof holds the whole table at once and drops it. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + policy.found("n", OTHER, ADDR2); + policy.set_own(OWN); + drop(policy); + } + + /// Proof 11. + #[test] + fn autolink_off_reports_but_never_dials() { + let mut policy = Policy::new(OWN, false); + assert_eq!( + policy.found("n", PEER, ADDR), + vec![Action::Found(ADDR)], + "the found peer reports" + ); + assert_eq!(policy.state(&PEER), Some(InstanceState::New)); + // An inbound link still lands, and its close waits without a retry. + policy.linked(PEER); + policy.unlinked(now(), PEER, CloseReason::PeerLost); + assert_eq!(policy.state(&PEER), Some(InstanceState::Waiting)); + assert!(policy.next_retry().is_none(), "no retry is armed"); + assert_eq!(policy.due(now() + Duration::from_secs(3600)), Vec::new()); + // The dial results the table removes never arrive, and the browse + // keeps reporting new addresses. + assert_eq!(policy.found("n", PEER, ADDR2), vec![Action::Found(ADDR2)]); + } + + /// The table's found row, column by column. + #[test] + fn the_found_row_holds_in_every_column() { + // New, dial. + let mut policy = autolink(); + assert_eq!( + policy.found("n", PEER, ADDR), + vec![Action::Found(ADDR), Action::Dial] + ); + // Linked, no dial. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + assert_eq!(policy.found("n", PEER, ADDR2), vec![Action::Found(ADDR2)]); + // Waiting, keep the retry. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + let retry = policy.next_retry(); + assert_eq!(policy.found("n", PEER, ADDR2), vec![Action::Found(ADDR2)]); + assert_eq!(policy.next_retry(), retry, "the retry is kept"); + // Barred, no dial. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Refused); + assert_eq!(policy.found("n", PEER, ADDR2), vec![Action::Found(ADDR2)]); + } + + /// A re-advertisement under a fresh name, same fingerprint and address, + /// reports the peer again, and a repeat of the same name does not. + #[test] + fn a_fresh_name_readvertised_reports_the_peer_again() { + let mut policy = autolink(); + assert_eq!( + policy.found("old", PEER, ADDR), + vec![Action::Found(ADDR), Action::Dial] + ); + // The same name resolves again, nothing reports. + assert_eq!(policy.found("old", PEER, ADDR), Vec::::new()); + // A fresh name, same fingerprint and address, reports again while + // the first dial is still in flight, so no second dial starts. + assert_eq!(policy.found("fresh", PEER, ADDR), vec![Action::Found(ADDR)]); + // A linked peer re-advertised reports without a dial. + let mut policy = autolink(); + policy.found("old", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + assert_eq!(policy.found("fresh", PEER, ADDR), vec![Action::Found(ADDR)]); + assert_eq!(policy.state(&PEER), Some(InstanceState::Linked)); + } + + /// The table's dial rows, the refused dial barred and the dial that + /// lands linked. + #[test] + fn the_dial_rows_hold() { + // A dial from a new instance lands linked. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + assert_eq!(policy.state(&PEER), Some(InstanceState::Linked)); + // A dial from the waiting, the backoff doubling. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + let due = now() + Duration::from_secs(5); + let _ = policy.due(due); + policy.dial_result(due, PEER, DialOutcome::Linked); + assert_eq!(policy.state(&PEER), Some(InstanceState::Linked)); + assert!(policy.next_retry().is_none()); + } + + /// A dial result the link already answered is the link's truth. + #[test] + fn a_dial_result_never_demotes_a_live_link() { + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + // The other addresses gave up, or the peer refused them, after the + // inbound link landed. + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + assert_eq!(policy.state(&PEER), Some(InstanceState::Linked)); + policy.dial_result(now(), PEER, DialOutcome::Refused); + assert_eq!(policy.state(&PEER), Some(InstanceState::Linked)); + // A bar, set by the close, is sticky under the late dial result. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Linked); + policy.unlinked(now(), PEER, CloseReason::PeerRevoked); + policy.dial_result(now(), PEER, DialOutcome::Linked); + assert_eq!(policy.state(&PEER), Some(InstanceState::Barred)); + } + + /// A fingerprint at `i`, for the bound's proofs. + fn fp_at(i: u16) -> Fingerprint { + let mut bytes = [0u8; 32]; + bytes[0] = u8::try_from(i >> 8).expect("the high half of a u16 holds a byte"); + bytes[1] = u8::try_from(i & 0xff).expect("the low half of a u16 holds a byte"); + Fingerprint::new(bytes) + } + + /// A loopback address at `i`, for the bound's proofs. + fn addr_at(i: u16) -> SocketAddr { + SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 443 + i) + } + + /// Decision 13. A due dial waits its turn while the bound of running + /// dials is full. + #[test] + fn a_due_dial_waits_its_turn_behind_the_running_dials() { + let mut policy = autolink(); + let bound = u16::try_from(MAX_DIALS).expect("the dial bound holds a u16"); + for i in 1..=bound { + assert!( + policy + .found("n", fp_at(i), addr_at(i)) + .contains(&Action::Dial), + "a running dial starts" + ); + } + let queued = fp_at(bound + 1); + assert_eq!( + policy.found("n", queued, addr_at(bound + 1)), + vec![Action::Found(addr_at(bound + 1))], + "the dial behind the bound is queued" + ); + assert!( + policy.due(now()).is_empty(), + "the bound holds while it is full" + ); + // A running dial ends, and the queued one takes its turn. + policy.dial_result(now(), fp_at(1), DialOutcome::Linked); + assert_eq!(policy.due(now()), vec![queued]); + } + + /// Decision 13. A queued retry takes its turn when a running dial ends. + #[test] + fn a_queued_retry_takes_its_turn_when_a_running_dial_ends() { + let mut policy = autolink(); + // The retry's deadline passes while the bound of dials runs. + policy.found("n", fp_at(1), addr_at(1)); + policy.dial_result(now(), fp_at(1), DialOutcome::Unreachable); + let bound = u16::try_from(MAX_DIALS).expect("the dial bound holds a u16"); + for i in 2..=bound + 1 { + policy.found("n", fp_at(i), addr_at(i)); + } + let due_at = now() + FIRST_WAIT; + assert!( + policy.due(due_at).is_empty(), + "the bound holds while it is full" + ); + // A running dial ends, and the queued retry takes its turn. + policy.dial_result(due_at, fp_at(2), DialOutcome::Linked); + assert_eq!(policy.due(due_at), vec![fp_at(1)]); + } + + /// Decision 13. A queued dial starts when a running dial ends, and a + /// removal, a link, an unlink or an address change meanwhile never frees + /// the slot before the outcome, nor starts a second dial. + #[test] + fn a_running_dial_holds_its_slot_until_its_outcome() { + let bound = u16::try_from(MAX_DIALS).expect("the dial bound holds a u16"); + let queue = |policy: &mut Policy| { + assert_eq!( + policy.found("n", fp_at(bound + 1), addr_at(bound + 1)), + vec![Action::Found(addr_at(bound + 1))], + "the running dial holds its slot" + ); + assert!(policy.due(now()).is_empty(), "the bound still holds"); + }; + let settle = |policy: &mut Policy| { + policy.dial_result(now(), fp_at(3), DialOutcome::Linked); + assert_eq!(policy.due(now()), vec![fp_at(bound + 1)]); + }; + // The instance is removed while its dial runs. + let mut policy = autolink(); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + assert_eq!(policy.removed(fp_at(3)), vec![Action::Gone]); + queue(&mut policy); + settle(&mut policy); + // An inbound link lands while the dial runs. + let mut policy = autolink(); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + policy.linked(fp_at(3)); + assert_eq!(policy.state(&fp_at(3)), Some(InstanceState::Linked)); + queue(&mut policy); + settle(&mut policy); + // A barred close lands while the dial runs. + let mut policy = autolink(); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + policy.linked(fp_at(3)); + policy.unlinked(now(), fp_at(3), CloseReason::PeerRevoked); + assert_eq!(policy.state(&fp_at(3)), Some(InstanceState::Barred)); + queue(&mut policy); + settle(&mut policy); + // The local addresses change while a waiting retry's dial runs. + let mut policy = autolink(); + policy.found("n", PEER, ADDR); + policy.dial_result(now(), PEER, DialOutcome::Unreachable); + let due_at = now() + FIRST_WAIT; + assert_eq!(policy.due(due_at), vec![PEER], "the retry dials"); + assert_eq!(policy.addresses_changed(), vec![Action::Browse]); + assert_eq!(policy.state(&PEER), Some(InstanceState::New)); + assert_eq!( + policy.found("n", PEER, ADDR2), + vec![Action::Found(ADDR2)], + "no second dial for a running one" + ); + assert!( + policy.due(due_at).is_empty(), + "the running dial holds its slot" + ); + policy.dial_result(due_at, PEER, DialOutcome::Unreachable); + assert_eq!(policy.state(&PEER), Some(InstanceState::Waiting)); + assert_eq!(policy.next_retry(), Some(due_at + FIRST_WAIT)); + } + + /// Decision 13. Queued dials start oldest first, by the find's order, + /// whatever the fingerprints order. + #[test] + fn queued_dials_start_oldest_first_not_by_fingerprint() { + let mut policy = autolink(); + let bound = u16::try_from(MAX_DIALS).expect("the dial bound holds a u16"); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + // The older instance carries the higher fingerprint. + let older = fp_at(bound + 2); + let younger = fp_at(bound + 1); + policy.found("n", older, addr_at(bound + 2)); + policy.found("n", younger, addr_at(bound + 1)); + assert!(policy.due(now()).is_empty(), "the bound holds while full"); + policy.dial_result(now(), fp_at(1), DialOutcome::Linked); + assert_eq!( + policy.due(now()), + vec![older], + "the older instance starts, not the lower fingerprint" + ); + } + + /// Decision 13. A find beyond the bound evicts the oldest instance not + /// linked, without an event. + #[test] + fn a_find_beyond_the_bound_evicts_the_oldest_not_linked() { + let bound = u16::try_from(MAX_INSTANCES).expect("the instance bound holds a u16"); + let beyond = fp_at(bound + 1); + // The oldest tracked instance, not linked, is evicted. + let mut policy = autolink(); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + assert_eq!( + policy.found("n", beyond, addr_at(bound + 1)), + vec![Action::Found(addr_at(bound + 1))], + "the find reports, without an eviction event" + ); + assert_eq!(policy.state(&beyond), Some(InstanceState::New)); + assert_eq!(policy.state(&fp_at(1)), None, "the oldest is evicted"); + // The evicted dial ends, and the next queued instance takes its place. + policy.dial_result(now(), fp_at(1), DialOutcome::Unreachable); + assert_eq!(policy.due(now()), vec![fp_at(9)]); + // A linked instance is kept, and its successor is evicted. + let mut policy = autolink(); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + policy.dial_result(now(), fp_at(1), DialOutcome::Linked); + assert_eq!( + policy.found("n", beyond, addr_at(bound + 1)), + vec![Action::Found(addr_at(bound + 1)), Action::Dial], + "the evicted dial still holds one place" + ); + assert_eq!(policy.state(&fp_at(1)), Some(InstanceState::Linked)); + assert_eq!( + policy.state(&fp_at(2)), + None, + "the oldest not linked is evicted" + ); + // A bound full of linked instances takes no find. + let mut policy = autolink(); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + for i in 1..=bound { + policy.dial_result(now(), fp_at(i), DialOutcome::Linked); + } + assert!( + policy.found("n", beyond, addr_at(bound + 1)).is_empty(), + "a bound full of links takes no find" + ); + assert_eq!(policy.state(&beyond), None); + assert_eq!(policy.state(&fp_at(1)), Some(InstanceState::Linked)); + // A find of a tracked instance takes no eviction. + let mut policy = autolink(); + for i in 1..=bound { + policy.found("n", fp_at(i), addr_at(i)); + } + assert_eq!( + policy.found("n", fp_at(1), addr_at(bound + 1)), + vec![Action::Found(addr_at(bound + 1))], + "a tracked instance re-found takes no eviction" + ); + assert_eq!(policy.state(&fp_at(2)), Some(InstanceState::New)); + } +} diff --git a/crates/connetto-peer/src/test_support.rs b/crates/connetto-peer/src/test_support.rs new file mode 100644 index 00000000..b790b421 --- /dev/null +++ b/crates/connetto-peer/src/test_support.rs @@ -0,0 +1,93 @@ +//! The loopback multicast probe the discovery proofs run before they trust +//! the interface, here and in `connetto-client`'s suite (R76 slice 3). +//! +//! A proof whose probe fails skips itself, so a host without loopback +//! multicast never reads as a discovery failure. Setting +//! `CONNETTO_REQUIRE_MULTICAST` turns a failed probe into a panic, which CI +//! sets so a broken probe never hides every proof behind a skip. + +use std::net::{Ipv4Addr, SocketAddrV4}; +use std::time::Duration; + +use socket2::{Domain, SockAddr, Socket, Type}; + +/// A loopback multicast round-trip, answered when the host lets it through. +/// +/// Binds a datagram socket to every address, joins the mDNS group on the +/// loopback, and asks a second socket, pointed at the loopback, to multicast +/// to it. +/// +/// # Panics +/// +/// When the probe fails while `CONNETTO_REQUIRE_MULTICAST` is set. +#[must_use] +pub fn loopback_multicast() -> bool { + let works = probe_loopback_multicast(); + assert!( + works || std::env::var_os("CONNETTO_REQUIRE_MULTICAST").is_none(), + "CONNETTO_REQUIRE_MULTICAST is set and the loopback multicast probe failed" + ); + works +} + +/// Whether a datagram sent to the mDNS group over the loopback comes back. +fn probe_loopback_multicast() -> bool { + let group = Ipv4Addr::new(224, 0, 0, 251); + let lo = Ipv4Addr::LOCALHOST; + let receiver = match Socket::new(Domain::IPV4, Type::DGRAM, None) { + Ok(socket) => socket, + Err(err) => { + eprintln!("the mDNS probe will not open a socket: {err}"); + return false; + } + }; + // Bound to every address, since a socket bound to the loopback's unicast + // address never receives a datagram sent to the group. + let bind = SockAddr::from(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0)); + if let Err(err) = receiver.bind(&bind) { + eprintln!("the mDNS probe will not bind the loopback: {err}"); + return false; + } + let Some(addr) = receiver + .local_addr() + .ok() + .and_then(|addr| addr.as_socket_ipv4()) + else { + eprintln!("the mDNS probe will not read its bound address"); + return false; + }; + let port = addr.port(); + if let Err(err) = receiver.join_multicast_v4(&group, &lo) { + eprintln!("the mDNS probe will not join the group on the loopback: {err}"); + return false; + } + let _ = receiver.set_read_timeout(Some(Duration::from_secs(3))); + let sender = match Socket::new(Domain::IPV4, Type::DGRAM, None) { + Ok(socket) => socket, + Err(err) => { + eprintln!("the mDNS probe will not open a sender: {err}"); + return false; + } + }; + if let Err(err) = sender.set_multicast_if_v4(&lo) { + eprintln!("the mDNS probe will not point the sender at the loopback: {err}"); + return false; + } + let target = SockAddr::from(SocketAddrV4::new(group, port)); + if let Err(err) = sender.send_to(b"connetto", &target) { + eprintln!("the mDNS probe will not multicast to the loopback: {err}"); + return false; + } + let mut buf = vec![std::mem::MaybeUninit::uninit(); 32]; + match receiver.recv(buf.as_mut_slice()) { + Ok(n) if n > 0 => true, + Ok(_) => { + eprintln!("the mDNS probe multicast no datagram to the loopback"); + false + } + Err(err) => { + eprintln!("the mDNS probe heard nothing on the loopback: {err}"); + false + } + } +} diff --git a/crates/connetto-peer/src/tests.rs b/crates/connetto-peer/src/tests.rs index 9a3ef79d..6c056265 100644 --- a/crates/connetto-peer/src/tests.rs +++ b/crates/connetto-peer/src/tests.rs @@ -54,7 +54,7 @@ impl DeviceKey for Device { } /// An issuer beside the certificate and key it signs with. -struct Issuer { +pub(crate) struct Issuer { /// The issuer certificate, DER. certificate: Vec, /// The signing key, for the lists it signs. @@ -65,13 +65,13 @@ struct Issuer { /// A deployment a test mints: one root and the issuers and devices it /// signs. -struct Deployment { +pub(crate) struct Deployment { root: RootCa, } impl Deployment { /// Mint a root for deployment `id`, valid around `now`. - fn new(id: u128, now: SystemTime) -> Self { + pub(crate) fn new(id: u128, now: SystemTime) -> Self { let root = RootCa::create( DeploymentId::from_uuid(uuid::Uuid::from_u128(id)), now - 10 * DAY, @@ -82,12 +82,12 @@ impl Deployment { } /// The root's DER, what a device is built with. - fn root_der(&self) -> Vec { + pub(crate) fn root_der(&self) -> Vec { self.root.certificate().to_vec() } /// Sign one more issuer, valid around `now`. - fn add_issuer(&self, now: SystemTime, serial: [u8; 16]) -> Issuer { + pub(crate) fn add_issuer(&self, now: SystemTime, serial: [u8; 16]) -> Issuer { let key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).expect("the issuer key"); let holder = Device(Arc::new(key)); let stored = rustls_pki_types::PrivatePkcs8KeyDer::from(holder.0.serialize_der()); @@ -108,7 +108,7 @@ impl Deployment { /// A device certificate under `issuer`, valid from `not_before` for `lifetime`. #[expect(clippy::unused_self, reason = "the issuer carries its own signer")] - fn device( + pub(crate) fn device( &self, issuer: &Issuer, account: &str, @@ -140,7 +140,7 @@ impl Deployment { /// Reissue the device `peer` under `issuer`, a new serial and window. #[expect(clippy::unused_self, reason = "the issuer carries its own signer")] - fn reissue( + pub(crate) fn reissue( &self, issuer: &Issuer, peer: &Peer, @@ -201,24 +201,24 @@ impl Deployment { } /// A device certificate beside the key that signs for it. -struct Peer { +pub(crate) struct Peer { /// The key, held in memory. key: Device, /// What the certificate names. - identity: DeviceIdentity, + pub(crate) identity: DeviceIdentity, /// The serial a list names the leaf by. serial: Vec, /// The attestation level the leaf records. level: AttestationLevel, /// The leaf, DER. - leaf: Vec, + pub(crate) leaf: Vec, /// The issuer certificate, DER. issuer: Vec, } impl Peer { /// The presented identity the node serves and dials with. - fn identity(&self) -> Identity { + pub(crate) fn identity(&self) -> Identity { Identity { certificate: self.leaf.clone(), issuer: self.issuer.clone(), @@ -269,7 +269,7 @@ fn node_on( } /// A node on one deployment root and the system clock. -fn node(root_der: Vec, events: mpsc::UnboundedSender) -> Node { +pub(crate) fn node(root_der: Vec, events: mpsc::UnboundedSender) -> Node { node_on(vec![root_der], all_levels(), Arc::new(SystemClock), events) } @@ -293,7 +293,7 @@ fn node_liveness( } /// An event channel pair. -fn events() -> ( +pub(crate) fn events() -> ( mpsc::UnboundedSender, mpsc::UnboundedReceiver, ) { @@ -301,7 +301,7 @@ fn events() -> ( } /// The whole second the wall clock stands on, as X.509 carries time. -fn whole_second() -> SystemTime { +pub(crate) fn whole_second() -> SystemTime { let secs = SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) .expect("after the epoch") @@ -310,7 +310,7 @@ fn whole_second() -> SystemTime { } /// Serve `peer` on loopback, answering the bound address. -fn serve(node: &Node, peer: &Peer) -> SocketAddr { +pub(crate) fn serve(node: &Node, peer: &Peer) -> SocketAddr { node.serve(SocketAddr::from(([127, 0, 0, 1], 0)), peer.identity()) .expect("the listener binds") } @@ -1622,6 +1622,135 @@ async fn a_dial_nobody_answers_or_without_an_identity_fails_typed() { )); } +/// A peer that closes the link before its hello is unreachable, not a +/// protocol break. +#[tokio::test] +async fn a_peer_that_closes_before_its_hello_is_unreachable() { + let now = whole_second(); + let deployment = Deployment::new(20, now); + let issuer = deployment.add_issuer(now, [1; 16]); + let peer = deployment.device( + &issuer, + "peer", + now, + DAY, + [1; 16], + AttestationLevel::Unproven, + ); + let dialer = deployment.device( + &issuer, + "dialer", + now, + DAY, + [2; 16], + AttestationLevel::Unproven, + ); + // A raw server on the peer's identity, completing the handshake and + // closing before the peer's hello. + let crls = Arc::new(parking_lot::RwLock::new(Arc::from( + Vec::::new().into_boxed_slice(), + ))); + let config = crate::node::server_config_for( + &trust_root(deployment.root_der()), + Arc::new(SystemClock), + Some(peer.key_id()), + crls, + &peer.identity(), + ); + let acceptor = tokio_rustls::TlsAcceptor::from(config); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the loopback binds"); + let addr = listener.local_addr().expect("its address"); + tokio::spawn(async move { + let (tcp, _) = listener.accept().await.expect("the dialer connects"); + let tls = acceptor.accept(tcp).await.expect("the handshake completes"); + drop(tls); + }); + let (d_tx, _d_rx) = events(); + let dialer_node = node(deployment.root_der(), d_tx); + let _served = serve(&dialer_node, &dialer); + let err = dialer_node + .link(addr) + .await + .expect_err("the hello never comes"); + assert!( + matches!(err, LinkError::Unreachable(_)), + "a closed peer is unreachable, got {err:?}" + ); +} + +/// Decision 13. The listener runs at most 16 inbound handshakes at once and +/// closes a connection beyond that at accept, and a real peer links after +/// the stalled ones give up their sockets. +#[tokio::test] +async fn a_seventeenth_inbound_connection_is_closed_at_accept() { + use tokio::io::AsyncReadExt; + + let now = whole_second(); + let deployment = Deployment::new(21, now); + let issuer = deployment.add_issuer(now, [1; 16]); + let host = deployment.device( + &issuer, + "host", + now, + DAY, + [1; 16], + AttestationLevel::Unproven, + ); + let peer = deployment.device( + &issuer, + "peer", + now, + DAY, + [2; 16], + AttestationLevel::Unproven, + ); + let (h_tx, mut h_rx) = events(); + let host_node = node(deployment.root_der(), h_tx); + let host_addr = serve(&host_node, &host); + + // Sixteen connections that send nothing, each holding an inbound + // handshake for the ten-second bound. + let mut stalled = Vec::with_capacity(16); + for _ in 0..16 { + stalled.push( + tokio::net::TcpStream::connect(host_addr) + .await + .expect("it connects"), + ); + } + // The seventeenth gets closed at accept, at once. + let mut seventeenth = tokio::net::TcpStream::connect(host_addr) + .await + .expect("it connects"); + let mut eof = [0u8; 8]; + assert_eq!( + time::timeout(Duration::from_secs(5), seventeenth.read(&mut eof)) + .await + .expect("the close is readable within the bound") + .expect("a close ends the read"), + 0, + "the seventeenth connection is closed at accept" + ); + // The stalled handshakes give up their sockets, and a real peer links. + drop(stalled); + // A bound for the stalled handshakes to notice the close and free their + // places. + time::sleep(Duration::from_secs(1)).await; + let _raw = raw_linked(deployment.root_der(), &peer, host_addr).await; + let event = await_event(&mut h_rx, Duration::from_secs(10), |event| { + matches!(event, PeerEvent::Linked { .. }) + }) + .await; + assert_eq!( + event, + PeerEvent::Linked { + peer: peer.identity.clone() + } + ); +} + #[tokio::test] async fn a_peer_speaking_another_version_is_refused_both_ways() { let now = whole_second(); diff --git a/docs/architecture/19-device-to-device.md b/docs/architecture/19-device-to-device.md index 6fdebda3..7bab51fc 100644 --- a/docs/architecture/19-device-to-device.md +++ b/docs/architecture/19-device-to-device.md @@ -48,7 +48,9 @@ The lifetime is application-requested per case under a server ceiling and refuse **Decided (R76).** The data plane is the local network, with discovery by gateway-probe first (a hotspot host IS the gateway, so clients need no multicast to find it) and mDNS as the general case under the service type `_connetto-peer._tcp.local`, then mutual TLS over plain TCP with both sides verifying against the built-in roots and the revocation lists. It is for native targets only, phones, tablets, laptops and desktops alike, since browsers cannot take the listening role on any radio path and keep the server path. The TXT record carries the protocol version and a certificate fingerprint only, never an account identity, and discovery is treated as fully public with authentication happening only at the handshake. -**Built (R76, 2026-10-07), the link.** The `connetto-peer` crate holds the link, and `connetto-client`'s `peer` feature drives it from the enrolment task. A device holding a valid certificate listens at the address `NativeDurable::with_peer_listener` names, `0.0.0.0:0` by default, and dials with `NativeClient::link_peer`. Both sides run TLS 1.3 signed by their device keys, verify each other against the deployment roots, the kept revocation lists, the five-minute window and the accepted attestation levels, and then exchange the lists the other lacks through the same intake the server's pushes use, so a list learned from a peer revokes exactly as one pushed by the server. A link pings every 15 seconds, closes after 45 seconds of silence or when the peer's certificate expires, takes a renewed certificate in a `Certificate` frame, and two devices dialling each other keep one link. A device whose certificate expires, whose clock leaves the window, or whose certificate is withdrawn or revoked closes every link and stops listening. Discovery, hosting and the Bluetooth exchange are the next slices. +**Built (R76, 2026-10-07), the link.** The `connetto-peer` crate holds the link, and `connetto-client`'s `peer` feature drives it from the enrolment task. A device holding a valid certificate listens at the address `NativeDurable::with_peer_listener` names, `0.0.0.0:0` by default, and dials with `NativeClient::link_peer`. Both sides run TLS 1.3 signed by their device keys, verify each other against the deployment roots, the kept revocation lists, the five-minute window and the accepted attestation levels, and then exchange the lists the other lacks through the same intake the server's pushes use, so a list learned from a peer revokes exactly as one pushed by the server. A link pings every 15 seconds, closes after 45 seconds of silence or when the peer's certificate expires, takes a renewed certificate in a `Certificate` frame, and two devices dialling each other keep one link. A device whose certificate expires, whose clock leaves the window, or whose certificate is withdrawn or revoked closes every link and stops listening. Hosting and the Bluetooth exchange are the next slices. + +**Built (R76, 2026-10-07), discovery.** While a device serves, it advertises `_connetto-peer._tcp.local` over mDNS under a random name, with only the frame version and the SHA-256 of its current certificate in the TXT record, and browses for the others, reporting each as `ClientEvent::PeerFound` and `PeerGone`. Unless the application turns autolink off with `NativeDurable::with_peer_autolink(false)`, the client dials what it finds, retrying an unreachable peer from 5 seconds up to 5 minutes and never redialling a refused, revoked or expired one until it advertises a renewed certificate. At most 8 dials run at once and at most 256 found devices are tracked, so fake advertisements cannot exhaust the device. Android holds a multicast lock while browsing. A device that joins a hotspot through the offer or the text fallback dials the host directly at the gateway, which arrives with the hotspot slice. **Decided (R76, 2026-10-02), making a shared network and joining it.** Android, Windows and Linux start a hotspot from the application (Android's local-only hotspot, a Wi-Fi Direct access point on Windows, NetworkManager's hotspot on Linux). macOS cannot host from an application, and an iPhone or iPad hosts only through Personal Hotspot switched on by hand, whose password the user types into the application once. A host announces itself with a public Bluetooth beacon carrying the protocol version and its certificate fingerprint. A joining device connects to it over Bluetooth, both run the same mutual TLS as the Wi-Fi link with the same checks and exchange revocation lists, and only then does the host send the hotspot's name and password, encrypted for that device, as Apple's Wi-Fi password sharing proves identity before sending a password. With Bluetooth off, the host shows a `WIFI:` QR code and the same details as text. Every platform then asks its own system to join. diff --git a/examples/dioxus-desktop-demo/android/src/main/AndroidManifest.xml b/examples/dioxus-desktop-demo/android/src/main/AndroidManifest.xml index 12a09389..16d6e527 100644 --- a/examples/dioxus-desktop-demo/android/src/main/AndroidManifest.xml +++ b/examples/dioxus-desktop-demo/android/src/main/AndroidManifest.xml @@ -3,7 +3,8 @@ Merged into the demo's generated manifest. It turns off cloud backup and device-to-device transfer, because the replica and the credential records are encrypted with a Keystore key that never leaves the phone, so a copy on - another device cannot be read. + another device cannot be read. It also declares the permission the peer + discovery's multicast lock needs. This module is a stand-in awaiting the proper home in Dioxus. dx 0.7.10 parses but never applies `[android.raw] application_attrs` and @@ -12,6 +13,7 @@ and its declaration in src/main.rs are deleted. --> + diff --git a/plans/master-implementation-plan.md b/plans/master-implementation-plan.md index 0b6787b3..94ba80a1 100644 --- a/plans/master-implementation-plan.md +++ b/plans/master-implementation-plan.md @@ -273,7 +273,7 @@ Execution order and nothing else. Status, blockers, landing dates and what each | R72 clock discipline (X6) | NOT STARTED | nothing | no | | R74 device identity and certificates | IN PROGRESS, merged (2026-10-06, #130), and step 7's peer half with proofs 3, 4 and 6's peer half built by R76's slices 1 and 2 (2026-10-07) | an iPhone run for proof 5's Apple half, the IANA PEN before a real deployment | no | | R75 the per-device applied frontier | NOT STARTED | R74. Touches the R2 watermark contract and the R56 import | no | -| R76 the peer link | IN PROGRESS, sliced into pull requests 2026-10-06 (its section's **Slices**), the Android slices first, slices 1 and 2 built 2026-10-07 | nothing for slices 1 to 7, other machines for slice 8 and the home run | no | +| R76 the peer link | IN PROGRESS, sliced into pull requests 2026-10-06 (its section's **Slices**), the Android slices first, slices 1 to 3 built 2026-10-07 | nothing for slices 1 to 7, other machines for slice 8 and the home run | no | | R77 the peer exchange and provisional tier | NOT STARTED | R75 and R76. The retraction's reason rides on R57 step 8's fix | no | | R78 courier recovery | NOT STARTED | R75 and R77 | no | | R79 media over the peer link | NOT STARTED | R77, and R64 to R67 for the chunk machinery | no | @@ -5279,7 +5279,7 @@ A write applied through any path is refused by every other path, proven three wa ## R76: the peer link -**Status.** IN PROGRESS, its hosting, joining, Bluetooth and proof questions decided with the maintainer on 2026-10-02 and the link's on 2026-10-06 (below). Slices 1 and 2, the `connetto-peer` crate and the client's peer link, are built (2026-10-07). +**Status.** IN PROGRESS, its hosting, joining, Bluetooth and proof questions decided with the maintainer on 2026-10-02 and the link's on 2026-10-06 (below). Slices 1 to 3, the `connetto-peer` crate, the client's peer link and discovery, are built (2026-10-07). **Blocked on** R74. @@ -5302,6 +5302,9 @@ The local data plane R25 decided, discovery by gateway-probe first (a hotspot ho 8. **A peer refuses a certificate whose attestation level is outside its own accepted set.** `NativeDurable::with_peer_accepted_attestation` takes the set, all three levels by default as on the server (R74 decision 33), and a refused peer closes with `AttestationRefused`. A set narrowed in an application update then takes effect at once. Relying on the server's enrolment gate alone was rejected, since certificates issued before an operator narrows the set stay valid for up to the ceiling. 9. **A certificate's expiry reaches every live link on time, at both ends** (2026-10-07). The device's task wakes at its own `not_after` plus the five-minute tolerance, and when a `ClockOff` certificate's window opens, so the table's standing rows apply when the standing changes. Each link closes at its peer's `not_after` plus the tolerance with `PeerExpired`, which a modified peer that never closes its own links cannot escape. Both are monotonic sleeps up to a wall-clock deadline, checked again against the wall clock when they fire, and a jumped clock is caught by the next look. Applying either only at the hourly look was rejected, since a link would outlive an expiry by up to an hour, and so was leaving a peer's expiry to that peer. 10. **A renewed certificate travels to every live link in a `Certificate` frame** (2026-10-07). After a grant the device sends `Certificate { leaf, issuer }` on every link, and the far end verifies the chain exactly as at a handshake, requires the same device key the handshake proved, and moves that link's expiry deadline. A chain the verifier refuses closes the link, with `PeerRevoked` for a revoked chain, `PeerExpired` for an expired one, and `Protocol` for any other refusal or another key. Ending every link at the old certificate's expiry was rejected, since each link would break once per certificate lifetime while both devices stayed valid. +11. **The gateway probe dials the port the join handed over** (2026-10-07). A device that joined a hotspot through the Bluetooth `HotspotOffer`, the text fallback or the join API learns the host's peer port with the network and dials the gateway at that port at once, so decision 6's system-picked port stays. Every other network relies on mDNS, whose SRV record carries the port. A fixed default port was rejected, since a second client on the host machine would be reachable only by mDNS and decision 6 would change, and so was dropping the probe, which leaves a hotspot whose multicast is filtered with no path. +12. **Discovery reports every peer it finds, and the client dials them itself unless the application turns that off** (2026-10-07). While the device serves, discovery advertises and browses, and reports `ClientEvent::PeerFound { address, fingerprint }` and `PeerGone { fingerprint }`. `NativeDurable::with_peer_autolink(bool)`, on by default, lets the client dial what it finds under the table below, which keeps every application from rebuilding the backoff and refusal rules. With autolink off the application dials with `link_peer`, for a picker the user confirms, for linking only while a screen is open, or for not revealing the device to every certified device nearby. Inbound links are accepted either way, since the listener verifies every peer. Reporting peers and leaving every dial to the application was rejected, since each application would rebuild the retry rules and a careless one would hammer refused peers. +13. **Discovery bounds what an unauthenticated network can make it do** (2026-10-07). At most 8 dials run at once and a due dial waits its turn, so every row still dials, and at most 256 instances are tracked, a find beyond that evicting the oldest instance not Linked without an event. A dial holds its slot until its outcome arrives, whatever happens to its instance meanwhile, and queued dials start oldest first. The listener runs at most 16 inbound handshakes at once and closes a connection beyond that at accept, and a failed accept waits 100 ms before the next. mDNS lets any process on the network advertise, and a few hundred fake instances each holding a dial's socket for the 10-second handshake bound would exhaust an Android application's file descriptors, its server connection included. Leaving the instance table unbounded was rejected, since memory and `PeerFound` events would still grow with fakes until their TTL runs out. ### Design @@ -5310,6 +5313,30 @@ The local data plane R25 decided, discovery by gateway-probe first (a hotspot ho - **Hosting permissions.** Android needs `CHANGE_WIFI_STATE`, `NEARBY_WIFI_DEVICES`, `BLUETOOTH_ADVERTISE` and `BLUETOOTH_CONNECT`. pippo's NetworkManager answers `no` to `wifi.share.protected` outside a desktop session, measured, so a refused hotspot maps to a typed error naming the permission. A host keeps its hotspot credentials in memory only, for the life of the hotspot. - **Revocation lists on the link.** Each side sends its CRL Numbers per issuer as the first frames inside the established TLS session and then any list the other lacks (R74). A live link whose peer the new list revokes closes with a typed reason. - **Discovery networks.** emi's eduroam put the two A35s in separate subnets, measured, so they reach each other by direct connection and never by mDNS. Devices on such a network meet through a hotspot. + +- **Discovery.** `mdns-sd` advertises `_connetto-peer._tcp.local` under a random instance name drawn at each registration, at the bound port, with a TXT record of `v`, the frame version, and `fp`, the hex SHA-256 of the presented leaf, so the instance and fingerprint change at every renewal and nothing in it names an account. A found instance is dialled at each resolved address, IPv4 first, until one answers. On Android the client holds a `WifiManager.MulticastLock` while it browses, taken through the application's `JavaAccess`. + +Discovery against the device's standing and the found instance's state, beside the link's table above. `Serving` is the link's `Serving`, `Idle` is every other standing. An instance is `New` when found, `Linked` while a live link carries its fingerprint, `Waiting` while a retry is pending, and `Barred` after a refusal until its fingerprint changes. Autolink off removes every dial and retry from the table and keeps the rest. + +| Event | Idle | Serving, instance New | Serving, Linked | Serving, Waiting | Serving, Barred | +| --- | --- | --- | --- | --- | --- | +| Standing becomes Serving | register the advertisement, start browsing | not applicable | not applicable | not applicable | not applicable | +| Standing leaves Serving | not applicable | unregister, stop browsing, forget every instance | same | same, drop the retry | same | +| Renewal changes the leaf | not applicable | register again under a new name and fingerprint | same | same | same | +| Instance found carrying the device's own fingerprint | not applicable | ignore it | not applicable | not applicable | not applicable | +| Instance found or resolved | not applicable | emit `PeerFound`, dial | emit `PeerFound`, no dial | emit `PeerFound`, keep the retry | emit `PeerFound`, no dial | +| Dial links | not applicable | becomes Linked | not applicable | becomes Linked | not applicable | +| Dial unreachable or timed out | not applicable | becomes Waiting, retry after 5 s | not applicable | retry after twice the last wait, at most 5 min | not applicable | +| Dial refused, or link closed with `PeerRevoked`, `PeerExpired` or `Protocol` | not applicable | becomes Barred | becomes Barred | becomes Barred | not applicable | +| Link closed with `PeerLost`, `Closed` or `Duplicate` | not applicable | not applicable | becomes Waiting, retry after 5 s | not applicable | not applicable | +| Instance found under a new fingerprint for a Barred peer | not applicable | not applicable | not applicable | not applicable | becomes New, dial | +| Instance removed | not applicable | emit `PeerGone`, forget it | emit `PeerGone`, forget it, the link stays | emit `PeerGone`, drop the retry | emit `PeerGone`, forget it | +| Local addresses change | not applicable | browse again | the link stays | drop the retry, become New | stay Barred | +| Inbound link from a found peer | not applicable | becomes Linked | not applicable | becomes Linked, drop the retry | becomes Linked, since the verifier accepted it | +| Instance found while 256 are tracked | not applicable | evict the oldest instance not Linked without an event, then as found | same | same | same | +| A dial is due while 8 dials run | not applicable | queue it, dial when a running dial ends | not applicable | queue the retry the same way | not applicable | +| Inbound connection while 16 handshakes run | close it at accept | close it at accept | same | same | same | +| Autolink turned off at build | browse and report only | no dial | no change | no retry | no change | - **The link.** TLS 1.3 alone over TCP with rustls and ring, each side presenting its leaf and issuer and signing through the device key, which signs ECDSA P-256 SHA-256 in DER exactly as TLS 1.3's `ecdsa_secp256r1_sha256` needs. Both verifiers wrap webpki over the deployment roots and the kept lists, checking the whole chain against them with unknown status allowed and list expiry ignored (R74), at the local time and, when that fails on validity alone, five minutes either side. They then require the R74 certificate profile, an attestation level in the accepted set, and a device key other than the device's own. The dialled name is ignored, since a device's identity is its URI. TCP connect and the handshake are bounded at 10 seconds each. - **Frames.** A 4-byte big-endian length, then a MessagePack `PeerFrame` as `connetto-core::codec` encodes the server protocol, at most 4 MiB, a longer one closing the link with `Protocol`. `Hello { version, dial, numbers }` goes first from each side, `dial` the dialler's counter for decision 7 and `numbers` the CRL Number kept per issuer key, then `List` carries each list the other side's numbers lack, `Ping` and `Pong` keep it alive, `Certificate { leaf, issuer }` carries a renewed chain (decision 10), and `Close { reason }` precedes a deliberate close. A received reason other than `Duplicate` is reported as `Closed`, since it names the sender's view. A received list goes to the client's one intake (R74 decision 22), and every list the intake keeps, from the server or from a peer, comes back to the link, which forwards it to every link whose numbers lack it and closes every link it revokes. - **Events.** `ClientEvent::PeerLinked { peer }` once a link survives decision 7, `PeerUnlinked { peer, reason }` when it closes, `CertificateExpired` on a refused `link_peer`, and `PeerListenFailed` when the listener cannot bind. A refused inbound handshake is logged and raises nothing, since any scanner on the network could cause one. @@ -5357,7 +5384,7 @@ One pull request each, in this order, each green and useful on its own (2026-10- 1. **The `connetto-peer` link on loopback** (step 2, decisions 5, 7 and 8). **Built (R76, 2026-10-07)** as `crates/connetto-peer`, proven by its seventeen loopback tests. The crate alone, its verifiers, frames, list forwarding, ping and duplicate rule, taking its certificate, key, roots, lists and clock as inputs. Proven in CI on loopback with certificates minted in the test, which carries R74's proof 4 and proof 6's refusal with a clock set outside the window. 2. **The client's peer link** (R74 step 7's peer half, decision 6). **Built (R76, 2026-10-07)** behind `connetto-client`'s `peer` feature, with decisions 9 and 10 in `connetto-peer`, proven by the enrolment unit tests and by `a_reported_device_is_refused_by_a_peer_that_learned_it_from_a_peer` against the real server. The `peer` feature, the builder setters, the table above against the enrolment task's standing, the events, and received lists through the one intake. Proven in CI against the real server, which carries R74's proof 3, three clients and a server stopped before the reported device dials. -3. **Discovery** (step 1). The gateway probe and `mdns-sd` under `_connetto-peer._tcp.local`, the TXT record carrying the protocol version and the certificate fingerprint only, each found peer handed to slice 2's link. Proven on loopback in CI and between the two A35s on one network. +3. **Discovery** (step 1, decisions 12 and 13). **Built (R76, 2026-10-07)** in `connetto-peer`'s `discovery` feature, which `connetto-client`'s `peer` feature turns on. `mdns-sd` under `_connetto-peer._tcp.local`, the TXT record carrying the frame version and the leaf fingerprint only, `PeerFound` and `PeerGone`, autolink under the discovery table, and Android's `MulticastLock`. Proven on loopback in CI. The gateway probe of decision 11 arrives with slice 4's join API, which knows the gateway and the port, and the A35 leg of discovery runs over slice 4's hotspot, since emi's eduroam separates the phones. 4. **Android hosting and joining** (step 3's Android part and step 6's Wi-Fi permissions). `startLocalOnlyHotspot` from the application through a platform plugin on the `connetto-auth-session` pattern, the typed refusal naming a missing permission, joining through the platform's network request, and the demo showing its peers. Proven on emi, one A35 hosting and the other joining through `cmd wifi connect-network`. 5. **The Bluetooth beacon and exchange on Android** (step 4's Android part and step 6's Bluetooth permissions). `BluetoothGattServer` as host, the joiner over btleplug, rustls over the framed characteristics, and `HotspotOffer` only after both sides proved their identity. Proven on emi between the two A35s. 6. **The text and QR fallback** (step 5). The `WIFI:` payload and its text form, the text path proven on emi and the camera scan left to the home run.