From 9c4d47c1a67ad14b7c4fbad62e04f17d09a15e7c Mon Sep 17 00:00:00 2001 From: alaa-eddine Date: Thu, 15 Oct 2026 14:42:10 +0200 Subject: [PATCH 1/8] feat(dht): expose announce relays with our observed address `QueryReply` gains `to`: our own address as the responding node saw it, already decoded off the wire. `HyperDhtHandle` gains `current_relays()`: the self-announce relays, each paired with that address, mirroring Node hyperdht's `Announcer.relays` (`{ relayAddress: msg.from, peerAddress: msg.to }`). A server advertises these pairs in the `holepunch.relays` of its handshake reply, and a relay forwards a client's PEER_HOLEPUNCH to `peer_address`, so it has to be how that relay reaches us. `current_relay_addresses()` keeps its signature and behaviour. Co-Authored-By: Claude Opus 5.5 --- peeroxide-dht/CHANGELOG.md | 5 ++ peeroxide-dht/src/hyperdht.rs | 133 ++++++++++++++++++++++++++++++---- peeroxide-dht/src/query.rs | 6 ++ peeroxide-dht/src/rpc.rs | 1 + 4 files changed, 130 insertions(+), 15 deletions(-) diff --git a/peeroxide-dht/CHANGELOG.md b/peeroxide-dht/CHANGELOG.md index 4acf541..1cf899e 100644 --- a/peeroxide-dht/CHANGELOG.md +++ b/peeroxide-dht/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `HyperDhtHandle::current_relays()`: the relays of the last self-announce, each paired with our own address as that relay saw it (`RelayInfo { relay_address, peer_address }`), mirroring Node hyperdht's `Announcer.relays`. A server advertises these in the `holepunch.relays` of its handshake reply; the relay forwards a client's PEER_HOLEPUNCH to `peer_address`. `current_relay_addresses()` is unchanged. +- `QueryReply::to`: our own address as the responding node saw it (the `to` field of its reply). + ## [1.7.3](https://github.com/Rightbracket/peeroxide/compare/peeroxide-dht-v1.7.2...peeroxide-dht-v1.7.3) - 2026-07-12 ### Other diff --git a/peeroxide-dht/src/hyperdht.rs b/peeroxide-dht/src/hyperdht.rs index 99b85be..c8c8959 100644 --- a/peeroxide-dht/src/hyperdht.rs +++ b/peeroxide-dht/src/hyperdht.rs @@ -24,7 +24,7 @@ use crate::hyperdht_messages::{ decode_lookup_raw_reply_from_bytes, decode_mutable_get_response_from_bytes, encode_announce_to_bytes, encode_hyper_peer_to_bytes, encode_mutable_put_request_to_bytes, AnnounceMessage, HandshakeMessage, HolepunchMessage, - HolepunchPayload, HyperPeer, MutablePutRequest, NoisePayload, RelayThroughInfo, + HolepunchPayload, HyperPeer, MutablePutRequest, NoisePayload, RelayInfo, RelayThroughInfo, SecretStreamInfo, UdxInfo, ANNOUNCE, ERROR_ABORTED, ERROR_NONE, ERROR_TRY_LATER, FIND_PEER, FIREWALL_OPEN, FIREWALL_UNKNOWN, IMMUTABLE_GET, IMMUTABLE_PUT, LOOKUP, MUTABLE_GET, MUTABLE_PUT, PEER_HANDSHAKE, PEER_HOLEPUNCH, UNANNOUNCE, @@ -518,7 +518,7 @@ pub struct HyperDhtHandle { router: Arc>, server_tx: mpsc::UnboundedSender, admin_tx: mpsc::UnboundedSender, - current_relay_addresses: Arc>>, + current_relays: Arc>>, } impl HyperDhtHandle { @@ -595,7 +595,7 @@ impl HyperDhtHandle { .await?; let mut closest_nodes = Vec::new(); - let mut accepted_relays: Vec = Vec::new(); + let mut accepted_relays: Vec = Vec::new(); for reply in &replies { closest_nodes.push(reply.from.clone()); @@ -655,11 +655,15 @@ impl HyperDhtHandle { // Accumulate this acker into handle-state relay list (Node // `Announcer._commit`: `if (relayAddresses.length < 3) - // relayAddresses.push({ host: msg.from.host, port: msg.from.port })`). + // relayAddresses.push({ host: msg.from.host, port: msg.from.port })` + // and `relays.push({ relayAddress: msg.from, peerAddress: msg.to })`). // We push only once per accepted reply and cap at 3. Dedup is // implicit — each `reply.from` is a distinct DHT node. if accepted_relays.len() < 3 { - accepted_relays.push(reply.from.clone()); + accepted_relays.push(RelayInfo { + relay_address: reply.from.clone(), + peer_address: reply.to.clone(), + }); } } @@ -671,7 +675,7 @@ impl HyperDhtHandle { // the relay list the topic-announce reads on a subsequent refresh. // See peer_discovery::do_refresh for the parallel-join order. if target == hash(&key_pair.public_key) { - if let Ok(mut guard) = self.current_relay_addresses.lock() { + if let Ok(mut guard) = self.current_relays.lock() { *guard = accepted_relays; } } @@ -1073,7 +1077,25 @@ impl HyperDhtHandle { /// /// Mirrors Node Hyperswarm's `Server.relayAddresses` / `Announcer.relayAddresses`. pub fn current_relay_addresses(&self) -> Vec { - self.current_relay_addresses + self.current_relays() + .into_iter() + .map(|relay| relay.relay_address) + .collect() + } + + /// The same relays as [`Self::current_relay_addresses`], each paired with + /// our own address as that relay saw it when it accepted the announce. + /// + /// This is what a server advertises in the `holepunch.relays` of its + /// handshake reply: a client sends its PEER_HOLEPUNCH to `relay_address`, + /// and the relay forwards it to `peer_address`, so `peer_address` must be + /// how that relay reaches *us*. Empty until the first `announce()` on + /// `hash(public_key)` completes. + /// + /// Mirrors Node Hyperswarm's `Announcer.relays` + /// (`{ relayAddress: msg.from, peerAddress: msg.to }`). + pub fn current_relays(&self) -> Vec { + self.current_relays .lock() .map(|g| g.clone()) .unwrap_or_default() @@ -2493,7 +2515,7 @@ pub async fn spawn( router, server_tx, admin_tx, - current_relay_addresses: Arc::new(Mutex::new(Vec::new())), + current_relays: Arc::new(Mutex::new(Vec::new())), }; Ok((join, handle, server_rx)) } @@ -3371,6 +3393,10 @@ mod tests { handle.current_relay_addresses().is_empty(), "current_relay_addresses should be empty before any announce()" ); + assert!( + handle.current_relays().is_empty(), + "current_relays should be empty before any announce()" + ); let _ = handle.destroy().await; } @@ -3393,13 +3419,16 @@ mod tests { let clone = handle.clone(); { - let mut guard = handle - .current_relay_addresses - .lock() - .expect("lock current_relay_addresses"); - *guard = vec![Ipv4Peer { - host: "1.2.3.4".to_string(), - port: 49737, + let mut guard = handle.current_relays.lock().expect("lock current_relays"); + *guard = vec![RelayInfo { + relay_address: Ipv4Peer { + host: "1.2.3.4".to_string(), + port: 49737, + }, + peer_address: Ipv4Peer { + host: "5.6.7.8".to_string(), + port: 61000, + }, }]; } @@ -3408,7 +3437,81 @@ mod tests { assert_eq!(via_clone[0].host, "1.2.3.4"); assert_eq!(via_clone[0].port, 49737); + let relays = clone.current_relays(); + assert_eq!(relays.len(), 1); + assert_eq!(relays[0].relay_address, via_clone[0]); + assert_eq!(relays[0].peer_address.host, "5.6.7.8"); + assert_eq!(relays[0].peer_address.port, 61000); + + let _ = handle.destroy().await; + } + + /// Each relay is paired with our own address as that relay saw it (Node + /// `Announcer.relays`), not with anyone else's: that is where it forwards + /// a client's PEER_HOLEPUNCH. + #[tokio::test] + async fn current_relays_pair_each_relay_with_our_observed_address() { + use libudx::UdxRuntime; + + // A reachable, persistent bootstrap node that accepts announces, and a + // firewalled ephemeral node that announces through it. + let local = |bootstrap: Vec, firewalled: bool| HyperDhtConfig { + dht: crate::rpc::DhtConfig { + bootstrap, + port: 0, + host: "127.0.0.1".to_string(), + firewalled, + ephemeral: Some(firewalled), + ..crate::rpc::DhtConfig::default() + }, + ..HyperDhtConfig::default() + }; + + let runtime = UdxRuntime::new().expect("runtime"); + let (_bs_task, bs, _bs_rx) = spawn(&runtime, local(vec![], false)) + .await + .expect("spawn bootstrap"); + let bs_port = bs.local_port().await.expect("bootstrap port"); + + let bootstrap = vec![format!("127.0.0.1:{bs_port}")]; + let (_task, handle, _rx) = spawn(&runtime, local(bootstrap, true)) + .await + .expect("spawn announcer"); + handle.bootstrapped().await.expect("bootstrapped"); + + let key_pair = KeyPair::generate(); + handle + .announce(hash(&key_pair.public_key), &key_pair, &[]) + .await + .expect("announce"); + + let relays = handle.current_relays(); + assert!(!relays.is_empty(), "the bootstrap node should accept the announce"); + + let ours = handle + .server_socket() + .await + .expect("server socket") + .expect("bound socket") + .local_addr() + .await + .expect("local addr"); + for relay in &relays { + assert_eq!(relay.relay_address.port, bs_port); + assert_eq!(relay.peer_address.host, "127.0.0.1"); + assert_eq!( + relay.peer_address.port, + ours.port(), + "peer_address must be our own address as the relay saw it" + ); + } + assert_eq!( + handle.current_relay_addresses(), + relays.iter().map(|r| r.relay_address.clone()).collect::>() + ); + let _ = handle.destroy().await; + let _ = bs.destroy().await; } } diff --git a/peeroxide-dht/src/query.rs b/peeroxide-dht/src/query.rs index 86f6f72..13646c6 100644 --- a/peeroxide-dht/src/query.rs +++ b/peeroxide-dht/src/query.rs @@ -52,6 +52,9 @@ pub struct QueryReply { pub value: Option>, /// Measured round-trip time for this reply. pub rtt: Duration, + /// Our own address as the responder saw it (the `to` field of its reply). + /// Behind a NAT this is the public mapping that node can reach us on. + pub to: Ipv4Peer, } /// The final result of a completed query. @@ -76,6 +79,7 @@ pub(crate) struct IoResponseData { pub error: u64, pub value: Option>, pub rtt: Duration, + pub to: Ipv4Peer, } /// A request produced by the query engine for the [`crate::rpc::DhtNode`] to send. @@ -270,6 +274,7 @@ impl Query { error: data.error, value: data.value, rtt: data.rtt, + to: data.to, }; self.push_closest(reply); } @@ -734,6 +739,7 @@ mod tests { error: 0, value: None, rtt: Duration::from_millis(5), + to: peer("127.0.0.1", 1), } } diff --git a/peeroxide-dht/src/rpc.rs b/peeroxide-dht/src/rpc.rs index 0227749..ca9684c 100644 --- a/peeroxide-dht/src/rpc.rs +++ b/peeroxide-dht/src/rpc.rs @@ -818,6 +818,7 @@ impl DhtNode { error, value: value.clone(), rtt, + to: to.clone(), }; if let Some((query_id, is_commit)) = self.tid_to_query.remove(&tid) { From da6c3eebc8c3f8715ed8e6590dfe7e8971dc0458 Mon Sep 17 00:00:00 2001 From: alaa-eddine Date: Thu, 15 Oct 2026 14:42:10 +0200 Subject: [PATCH 2/8] fix(swarm): let Node.js clients connect to a firewalled server A Node.js Hyperswarm client could not connect to a firewalled peeroxide server: every attempt timed out ("passive holepunch timed out" here, HOLEPUNCH_ABORTED on the Node side). The server side of the handshake now matches Node hyperdht `lib/server.js` on three points: - Duplicate PEER_HANDSHAKE. A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Every copy now gets the same reply, cached by the Noise message for 20 s. Before, each copy got a fresh responder, and the second replaced the holepunch session of the first. - Holepunch relays. `holepunch.relays[].peer_address` is now our own address as each relay saw it (`HyperDhtHandle::current_relays()`). Before, it held the client's address, so the relay sent the client's PEER_HOLEPUNCH back to the client. - Open or direct clients. A client that says it is FIREWALL_OPEN, or that reached us without a relay, is now dialled directly, as Node does. Before, it was given a holepunch, while it waited for that dial. Tests: two Node-client -> Rust-server interop tests (a firewalled client that holepunches through the relays, and a reachable client that waits to be dialled), a Rust-only test that sends one PEER_HANDSHAKE twice, and unit tests for the reply cache and envelope. The Node helper gains two opt-in flags, `--remote-client` and `--open-client`; without them it behaves as before. Co-Authored-By: Claude Opus 5.5 --- TESTING.md | 13 +- peeroxide/CHANGELOG.md | 7 + peeroxide/src/swarm.rs | 192 +++++++++++++++++----- peeroxide/tests/hyperswarm_interop.rs | 143 ++++++++++++++++ peeroxide/tests/server_handshake_dedup.rs | 131 +++++++++++++++ tests/node/hyperswarm-server.js | 31 +++- 6 files changed, 472 insertions(+), 45 deletions(-) create mode 100644 peeroxide/tests/server_handshake_dedup.rs diff --git a/TESTING.md b/TESTING.md index 67929b5..2f22864 100644 --- a/TESTING.md +++ b/TESTING.md @@ -163,12 +163,17 @@ Passing result: Protomux framing and blind-relay control messages are byte-ident | Suite | Command | Type | Count | |-------|---------|------|-------| -| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 21 | -| Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 1 | +| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 26 | +| Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 3 | +| Duplicate handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 1 | -**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test) +**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut, as on another network, and one more DHT node is started so it can classify its NAT) and `--open-client` (the Node client's DHT node is reachable, like a peer with a public address). -Passing result: Rust Hyperswarm client connects to a Node.js `hyperswarm` server over the DHT. Full flow: DHT bootstrap → topic announce/lookup → PEER_HANDSHAKE → skip-holepunching (same-host) → UDX stream → SecretStream encrypted channel → receive "hello from node". Verifies end-to-end interoperability of the complete Hyperswarm stack. +Passing result: +- `hyperswarm_cross_language_connect`: Rust Hyperswarm client connects to a Node.js `hyperswarm` server over the DHT. Full flow: DHT bootstrap → topic announce/lookup → PEER_HANDSHAKE → skip-holepunching (same-host) → UDX stream → SecretStream encrypted channel → receive "hello from node". Verifies end-to-end interoperability of the complete Hyperswarm stack. +- `hyperswarm_node_client_holepunches_to_rust_server`: a firewalled Node.js client connects to a firewalled Rust server. The client sends one PEER_HANDSHAKE through two relays in parallel, then holepunches through the relays the server advertised; data flows both ways. +- `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node.js client waits for the firewalled Rust server to dial it, as a Node server would; data flows both ways. +- `server_handshake_dedup`: a Rust client sends the same PEER_HANDSHAKE twice to a swarm server and gets byte-identical replies. ### Live Network Validation diff --git a/peeroxide/CHANGELOG.md b/peeroxide/CHANGELOG.md index 5127c16..08a09a4 100644 --- a/peeroxide/CHANGELOG.md +++ b/peeroxide/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Node.js Hyperswarm clients can now connect to a firewalled peeroxide server. Before, every attempt timed out (`passive holepunch timed out` here, `HOLEPUNCH_ABORTED` on the Node side). The server side of the handshake now matches Node hyperdht `lib/server.js` on three points: + - **Duplicate PEER_HANDSHAKE.** A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Each copy now gets the same reply, cached by the Noise message for 20 s. Before, each copy got a fresh Noise responder, and the second replaced the first's holepunch session. + - **Holepunch relays.** The reply's `holepunch.relays[].peer_address` is now our own address as each relay saw it (`HyperDhtHandle::current_relays()`). Before, it held the client's address, so the relay sent the client's PEER_HOLEPUNCH back to the client. + - **Open or direct clients.** A client that says it is `FIREWALL_OPEN`, or that reached us without a relay, is now dialled directly, as Node does. Before, it was given a holepunch, while the Node client waited for that dial. + ## [1.7.3](https://github.com/Rightbracket/peeroxide/compare/peeroxide-v1.7.2...peeroxide-v1.7.3) - 2026-07-12 ### Other diff --git a/peeroxide/src/swarm.rs b/peeroxide/src/swarm.rs index 439e8d6..9efe5bc 100644 --- a/peeroxide/src/swarm.rs +++ b/peeroxide/src/swarm.rs @@ -18,8 +18,8 @@ use peeroxide_dht::hyperdht::{ }; use peeroxide_dht::hyperdht_messages::{ encode_handshake_to_bytes, encode_holepunch_msg_to_bytes, HandshakeMessage, HolepunchInfo, - HolepunchMessage, HolepunchPayload, NoisePayload, RelayInfo, RelayThroughInfo, - SecretStreamInfo, UdxInfo, ERROR_NONE, FIREWALL_CONSISTENT, FIREWALL_RANDOM, FIREWALL_UNKNOWN, + HolepunchMessage, HolepunchPayload, NoisePayload, RelayThroughInfo, SecretStreamInfo, UdxInfo, + ERROR_NONE, FIREWALL_CONSISTENT, FIREWALL_OPEN, FIREWALL_RANDOM, FIREWALL_UNKNOWN, MODE_FROM_RELAY, MODE_FROM_SECOND_RELAY, MODE_FROM_SERVER, MODE_REPLY, }; use peeroxide_dht::messages::Ipv4Peer; @@ -43,6 +43,12 @@ fn next_stream_id() -> u32 { const DEFAULT_MAX_PEERS: usize = 64; const DEFAULT_MAX_PARALLEL: usize = 3; +/// How long the reply to a server-side PEER_HANDSHAKE is kept for duplicates +/// of the same request: the passive holepunch window (10 s) plus the 10 s +/// Node hyperdht keeps a settled handshake around (`HANDSHAKE_CLEAR_WAIT` in +/// `lib/server.js`). +const HANDSHAKE_REPLY_TTL: Duration = Duration::from_secs(20); + // ── Retry backoff tiers (matching Node.js lib/retry-timer.js) ──────────────── // Each tier: [base_ms, jitter1, jitter2, jitter3] // Delay = base + rand(0..j1) + rand(0..j2) + rand(0..j3) @@ -352,6 +358,76 @@ struct SwarmActor { pending_handshakes: HashMap<[u8; 32], u64>, next_holepunch_id: u64, passive_hp_event_tx: mpsc::UnboundedSender, + + // Noise replies to recent PEER_HANDSHAKEs, keyed by the client's Noise + // message. A Node client sends the same message through up to two relays + // in parallel and keeps the first reply; every copy must get that reply. + handshake_replies: HandshakeReplyCache, +} + +/// Noise replies to recent server-side PEER_HANDSHAKE requests, keyed by the +/// client's Noise message. +/// +/// Node hyperdht clients send one handshake through up to two relays at once +/// (`lib/connect.js`, `Semaphore(2)`) and act on whichever reply arrives +/// first, so a server must answer every copy identically, as Node's +/// `Server._onpeerhandshake` does. Answering each copy with a fresh Noise +/// responder gives the client a reply for a session the server then replaces. +struct HandshakeReplyCache { + ttl: Duration, + entries: HashMap, (tokio::time::Instant, Vec)>, +} + +impl HandshakeReplyCache { + fn new(ttl: Duration) -> Self { + Self { + ttl, + entries: HashMap::new(), + } + } + + /// The Noise reply already sent for this client Noise message, if any. + fn get(&mut self, client_noise: &[u8]) -> Option> { + self.prune(); + self.entries.get(client_noise).map(|(_, reply)| reply.clone()) + } + + fn insert(&mut self, client_noise: Vec, reply_noise: Vec) { + self.prune(); + self.entries.insert(client_noise, (tokio::time::Instant::now(), reply_noise)); + } + + fn prune(&mut self) { + let ttl = self.ttl; + self.entries.retain(|_, (at, _)| at.elapsed() < ttl); + } +} + +/// Wrap a Noise reply in the handshake envelope for the path the request took. +/// +/// Per Node `lib/server.js _addHandshake`: +/// - inbound FROM_CLIENT → respond MODE_REPLY (direct reply path). +/// - inbound FROM_RELAY / FROM_SECOND_RELAY → respond MODE_FROM_SERVER +/// with peer_address pointing to the originating client (carried in +/// the inbound msg.peer_address). The dht layer will dispatch this +/// via dht.relay_with_tid back to the FE-holder, which then routes +/// a tid-preserved REPLY to the receiver. +fn encode_server_handshake_reply( + inbound_mode: u64, + peer_address: Option, + noise_reply: Vec, +) -> Option> { + let (mode, peer_address) = match inbound_mode { + MODE_FROM_RELAY | MODE_FROM_SECOND_RELAY => (MODE_FROM_SERVER, peer_address), + _ => (MODE_REPLY, None), + }; + encode_handshake_to_bytes(&HandshakeMessage { + mode, + noise: noise_reply, + peer_address, + relay_address: None, + }) + .ok() } /// State carried between the initial server handshake and the moment the @@ -490,6 +566,7 @@ pub async fn spawn( pending_handshakes: HashMap::new(), next_holepunch_id: 0, passive_hp_event_tx, + handshake_replies: HandshakeReplyCache::new(HANDSHAKE_REPLY_TTL), }; // Keep the DHT runtime alive for the swarm's lifetime. @@ -942,6 +1019,17 @@ impl SwarmActor { ); } + // The same request relayed by another node: answer with the same reply. + if let Some(noise_reply) = self.handshake_replies.get(&msg.noise) { + tracing::debug!( + relay = %format!("{}:{}", from.host, from.port), + "server handshake: duplicate of an answered handshake, resending its reply" + ); + let reply = encode_server_handshake_reply(msg.mode, peer_address, noise_reply); + let _ = reply_tx.send(reply); + return; + } + let noise_kp = NoiseKeypair { public_key: self.key_pair.public_key, secret_key: self.key_pair.secret_key, @@ -986,10 +1074,16 @@ impl SwarmActor { // sure, in which case the existing direct path is the safer // choice (Node hyperdht has the same gate), AND // - we are not forcing all server connections through a relay - // (relay_through bypasses NAT traversal entirely). + // (relay_through bypasses NAT traversal entirely), AND + // - the client cannot simply be dialled. Node `lib/server.js + // _addHandshake` dials a client that says it is FIREWALL_OPEN, or + // that reached us directly rather than through a relay, and such a + // client waits for that dial instead of holepunching. let firewall_state = self.dht.firewalled().await.unwrap_or(FIREWALL_UNKNOWN); + let client_dialable = remote_payload.firewall == FIREWALL_OPEN || !is_forwarded; let attempt_holepunch = matches!(firewall_state, FIREWALL_CONSISTENT | FIREWALL_RANDOM) - && relay_through_info.is_none(); + && relay_through_info.is_none() + && !client_dialable; // Try to stage the passive Holepuncher up front so the reply we // send in this handshake can advertise the puncher's bound port @@ -1027,20 +1121,12 @@ impl SwarmActor { .map_or(self.local_port, |s| s.puncher_port); let addresses4 = self.dht.noise_addresses4(advertise_port).await; - let holepunch_info = holepunch_setup.as_ref().map(|setup| { - let relays: Vec = self - .dht - .current_relay_addresses() - .into_iter() - .map(|relay_addr| RelayInfo { - relay_address: relay_addr, - peer_address: client_address.clone(), - }) - .collect(); - HolepunchInfo { - id: setup.id, - relays, - } + // A client sends its PEER_HOLEPUNCH to `relay_address`, and the relay + // forwards it to `peer_address`: that must be our own address as the + // relay saw it (Node `Announcer.relays`), not the client's. + let holepunch_info = holepunch_setup.as_ref().map(|setup| HolepunchInfo { + id: setup.id, + relays: self.dht.current_relays(), }); let reply_payload = NoisePayload { @@ -1090,28 +1176,12 @@ impl SwarmActor { } }; - // Encode reply with mode + peer_address derived from inbound mode. - // Per Node `lib/server.js _addHandshake`: - // - inbound FROM_CLIENT → respond MODE_REPLY (direct reply path). - // - inbound FROM_RELAY / FROM_SECOND_RELAY → respond MODE_FROM_SERVER - // with peer_address pointing to the originating client (carried in - // the inbound msg.peer_address). The dht layer will dispatch this - // via dht.relay_with_tid back to the FE-holder, which then routes - // a tid-preserved REPLY to the receiver. - let (reply_mode, reply_peer_address) = match msg.mode { - MODE_FROM_RELAY | MODE_FROM_SECOND_RELAY => { - (MODE_FROM_SERVER, peer_address.clone()) - } - _ => (MODE_REPLY, None), - }; - - let reply_msg = HandshakeMessage { - mode: reply_mode, - noise: noise_reply, - peer_address: reply_peer_address, - relay_address: None, - }; - let _ = reply_tx.send(encode_handshake_to_bytes(&reply_msg).ok()); + self.handshake_replies.insert(msg.noise.clone(), noise_reply.clone()); + let _ = reply_tx.send(encode_server_handshake_reply( + msg.mode, + peer_address.clone(), + noise_reply, + )); let remote_pk = nw_result.remote_public_key; @@ -1805,6 +1875,48 @@ mod tests { assert!(j.client); } + #[tokio::test(start_paused = true)] + async fn handshake_reply_cache_returns_the_first_reply_until_it_expires() { + let mut cache = HandshakeReplyCache::new(Duration::from_secs(20)); + assert!(cache.get(b"noise-a").is_none()); + + cache.insert(b"noise-a".to_vec(), b"reply-a".to_vec()); + assert_eq!(cache.get(b"noise-a").as_deref(), Some(&b"reply-a"[..])); + assert!(cache.get(b"noise-b").is_none(), "other handshakes are not affected"); + + tokio::time::advance(Duration::from_secs(19)).await; + assert_eq!(cache.get(b"noise-a").as_deref(), Some(&b"reply-a"[..])); + + tokio::time::advance(Duration::from_secs(2)).await; + assert!(cache.get(b"noise-a").is_none()); + assert!(cache.entries.is_empty(), "expired entries are dropped"); + } + + #[test] + fn server_handshake_reply_follows_the_inbound_mode() { + use peeroxide_dht::hyperdht_messages::{decode_handshake_from_bytes, MODE_FROM_CLIENT}; + + let client = Ipv4Peer { + host: "10.0.0.1".to_string(), + port: 4000, + }; + + for inbound in [MODE_FROM_RELAY, MODE_FROM_SECOND_RELAY] { + let bytes = encode_server_handshake_reply(inbound, Some(client.clone()), vec![7; 48]) + .expect("encode"); + let m = decode_handshake_from_bytes(&bytes).expect("decode"); + assert_eq!(m.mode, MODE_FROM_SERVER); + assert_eq!(m.peer_address.as_ref(), Some(&client)); + assert_eq!(m.noise, vec![7; 48]); + } + + let bytes = encode_server_handshake_reply(MODE_FROM_CLIENT, Some(client), vec![7; 48]) + .expect("encode"); + let m = decode_handshake_from_bytes(&bytes).expect("decode"); + assert_eq!(m.mode, MODE_REPLY); + assert!(m.peer_address.is_none()); + } + #[tokio::test] async fn server_unregistered_on_leave_last_topic() { let config = SwarmConfig::default(); diff --git a/peeroxide/tests/hyperswarm_interop.rs b/peeroxide/tests/hyperswarm_interop.rs index ecd4c10..976e530 100644 --- a/peeroxide/tests/hyperswarm_interop.rs +++ b/peeroxide/tests/hyperswarm_interop.rs @@ -1,8 +1,20 @@ //! M8 interop: Rust peeroxide (Hyperswarm) ↔ Node.js hyperswarm. //! +//! `hyperswarm_cross_language_connect` (Rust client → Node server): //! 1. Node.js creates a local testnet + Hyperswarm, joins a topic as server. //! 2. Rust peeroxide joins the same topic as client, discovers the peer, //! and receives a SwarmConnection through the channel. +//! +//! Node client → Rust server, on the same local testnet: +//! 1. A firewalled Rust peeroxide swarm announces a topic as server. +//! 2. Node.js joins the topic as client and dials the Rust server. +//! +//! `hyperswarm_node_client_holepunches_to_rust_server`: a firewalled Node client +//! (with the same-host LAN shortcut off, as on another network) sends one +//! PEER_HANDSHAKE through two relays in parallel, then holepunches through the +//! relays listed in the server's reply. +//! `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node client +//! waits for the server to dial it, as Node servers do, instead of holepunching. use std::io::{BufRead, Write}; use std::process::{Command, Stdio}; @@ -29,8 +41,14 @@ struct JsSwarm { impl JsSwarm { fn start() -> Self { + Self::start_with(&[]) + } + + /// Start the helper with optional flags (see `tests/node/hyperswarm-server.js`). + fn start_with(flags: &[&str]) -> Self { let mut child = Command::new("node") .arg(server_script_path()) + .args(flags) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::inherit()) @@ -142,6 +160,40 @@ impl JsSwarm { "JS join failed: {reply}" ); } + + fn join_client(&mut self, topic_hex: &str) { + let id = self.send_cmd(serde_json::json!({ "cmd": "join_client", "topic": topic_hex })); + let reply = self.wait_reply(id); + assert!( + reply["ok"].as_bool().unwrap_or(false), + "JS join_client failed: {reply}" + ); + } + + /// Wait for a `data` event whose hex payload decodes to `expected`. + fn wait_data(&self, expected: &str) -> Value { + let want = hex::encode(expected.as_bytes()); + let deadline = std::time::Instant::now() + Duration::from_secs(30); + loop { + { + let lines = self.lines.lock().unwrap(); + for line in lines.iter() { + if let Ok(v) = serde_json::from_str::(line) { + if v["event"].as_str() == Some("data") + && v["payload"].as_str() == Some(want.as_str()) + { + return v; + } + } + } + } + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for JS data event with payload {expected:?}" + ); + std::thread::sleep(Duration::from_millis(50)); + } + } } impl Drop for JsSwarm { @@ -239,3 +291,94 @@ async fn run_interop() -> Result<(), Box> { Ok(()) } + +#[tokio::test] +async fn hyperswarm_node_client_holepunches_to_rust_server() { + node_client_interop(&["--remote-client"]).await; +} + +#[tokio::test] +async fn hyperswarm_open_node_client_connects_to_rust_server() { + node_client_interop(&["--open-client"]).await; +} + +async fn node_client_interop(flags: &[&str]) { + let _ = tracing_subscriber::fmt::try_init(); + + let result = + tokio::time::timeout(Duration::from_secs(90), run_node_client_interop(flags)).await; + + match result { + Ok(Ok(())) => {} + Ok(Err(e)) => panic!("node client {flags:?} -> rust server interop test failed: {e}"), + Err(_) => panic!("node client {flags:?} -> rust server interop test timed out after 90s"), + } +} + +async fn run_node_client_interop(flags: &[&str]) -> Result<(), Box> { + let mut js = JsSwarm::start_with(flags); + let (bs_port, _node_pk) = js.read_ready(); + tracing::info!(bs_port, "JS testnet ready"); + + let topic = discovery_key(b"peeroxide-node-client-rust-server-test"); + let topic_hex = hex::encode(topic); + + // Firewalled like a peer behind a NAT, so the server stages a passive + // holepunch that the Node client has to complete through the relays. + let mut dht_config = DhtConfig::default(); + dht_config.bootstrap = vec![format!("127.0.0.1:{bs_port}")]; + dht_config.port = 0; + dht_config.host = "127.0.0.1".to_string(); + dht_config.firewalled = true; + + let mut hyper_config = HyperDhtConfig::default(); + hyper_config.dht = dht_config; + + let mut config = SwarmConfig::default(); + config.dht = hyper_config; + + let (_task, handle, mut conn_rx) = spawn(config).await?; + let rust_pk = hex::encode(handle.key_pair().public_key); + + handle + .join( + topic, + { + let mut opts = JoinOpts::default(); + opts.client = false; + opts + }, + ) + .await?; + handle.flush().await?; + tracing::info!("Rust announced topic as server"); + + js.join_client(&topic_hex); + tracing::info!("JS joined topic as client"); + + let mut conn = tokio::time::timeout(Duration::from_secs(45), conn_rx.recv()) + .await + .map_err(|_| "timed out waiting for the Node client's connection")? + .ok_or("conn_rx closed without delivering a connection")?; + + assert!(!conn.is_initiator, "the Node client should be the initiator"); + + let connected_event = js.wait_event("connected"); + assert_eq!(connected_event["remotePk"].as_str(), Some(rust_pk.as_str())); + assert_eq!(connected_event["isInitiator"].as_bool(), Some(true)); + + let data = tokio::time::timeout(Duration::from_secs(10), conn.peer.stream.read()) + .await + .map_err(|_| "timed out reading from encrypted stream")? + .map_err(|e| format!("read error: {e}"))? + .ok_or("stream closed without data")?; + assert_eq!(String::from_utf8_lossy(&data), "hello from node"); + + conn.peer.stream.write(b"hello from rust").await?; + js.wait_data("hello from rust"); + + handle.destroy().await?; + drop(js); + + Ok(()) +} diff --git a/peeroxide/tests/server_handshake_dedup.rs b/peeroxide/tests/server_handshake_dedup.rs new file mode 100644 index 0000000..6d8ed0c --- /dev/null +++ b/peeroxide/tests/server_handshake_dedup.rs @@ -0,0 +1,131 @@ +//! A swarm server answers every copy of a PEER_HANDSHAKE with the same reply. +//! +//! Node hyperdht clients send one Noise handshake message through up to two +//! relays in parallel and act on the first reply that arrives (`lib/connect.js`), +//! so a server must answer each copy identically, as Node's +//! `Server._onpeerhandshake` does. Here a Rust client sends the same message +//! twice and checks that both replies are byte-identical and usable. + +use std::time::Duration; + +use libudx::UdxRuntime; +use peeroxide::{spawn as spawn_swarm, JoinOpts, SwarmConfig}; +use peeroxide_dht::crypto::hash; +use peeroxide_dht::hyperdht::{spawn, HyperDhtConfig, KeyPair}; +use peeroxide_dht::hyperdht_messages::{ + decode_handshake_from_bytes, NoisePayload, UdxInfo, FIREWALL_UNKNOWN, PEER_HANDSHAKE, +}; +use peeroxide_dht::messages::Ipv4Peer; +use peeroxide_dht::noise::Keypair as NoiseKeypair; +use peeroxide_dht::noise_wrap::NoiseWrap; +use peeroxide_dht::rpc::{DhtConfig, UserRequestParams}; +use peeroxide_dht::Router; + +fn local_dht(bootstrap: Vec, firewalled: bool) -> HyperDhtConfig { + let mut dht = DhtConfig::default(); + dht.bootstrap = bootstrap; + dht.port = 0; + dht.host = "127.0.0.1".to_string(); + dht.firewalled = firewalled; + let mut config = HyperDhtConfig::default(); + config.dht = dht; + config +} + +#[tokio::test] +async fn duplicate_peer_handshake_gets_the_same_reply() { + let _ = tracing_subscriber::fmt::try_init(); + + let result = tokio::time::timeout(Duration::from_secs(30), run()).await; + match result { + Ok(Ok(())) => {} + Ok(Err(e)) => panic!("duplicate handshake test failed: {e}"), + Err(_) => panic!("duplicate handshake test timed out after 30s"), + } +} + +async fn run() -> Result<(), Box> { + let rt = UdxRuntime::new()?; + + let (_bs_join, bs, _bs_rx) = spawn(&rt, local_dht(vec![], true)).await?; + let bootstrap = vec![format!("127.0.0.1:{}", bs.local_port().await?)]; + + // The server: a swarm announcing a topic. + let mut config = SwarmConfig::default(); + config.dht = local_dht(bootstrap.clone(), true); + let (_task, server, _conn_rx) = spawn_swarm(config).await?; + server + .join([0x5a; 32], { + let mut opts = JoinOpts::default(); + opts.client = false; + opts + }) + .await?; + let server_pk = server.key_pair().public_key; + let server_addr = Ipv4Peer { + host: "127.0.0.1".to_string(), + port: server.dht().local_port().await?, + }; + + // The client: one Noise message, sent twice. + let (_cli_join, client, _cli_rx) = spawn(&rt, local_dht(bootstrap, true)).await?; + client.bootstrapped().await?; + + let client_kp = KeyPair::generate(); + let mut nw = NoiseWrap::new_initiator( + NoiseKeypair { + public_key: client_kp.public_key, + secret_key: client_kp.secret_key, + }, + server_pk, + ); + let noise = nw.send(&NoisePayload { + version: 1, + error: 0, + firewall: FIREWALL_UNKNOWN, + holepunch: None, + addresses4: vec![], + addresses6: vec![], + udx: Some(UdxInfo { + version: 1, + reusable_socket: false, + id: 1, + seq: 0, + }), + secret_stream: None, + relay_through: None, + relay_addresses: None, + })?; + let request = Router::encode_client_handshake(noise, None, Some(server_addr.clone()))?; + + let mut replies = Vec::new(); + for _ in 0..2 { + let mut params = UserRequestParams::default(); + params.command = PEER_HANDSHAKE; + params.target = Some(hash(&server_pk)); + params.value = Some(request.clone()); + let resp = client + .dht() + .request(params, &server_addr.host, server_addr.port) + .await?; + assert_eq!(resp.error, 0, "server returned error code {}", resp.error); + replies.push(resp.value.ok_or("handshake reply without a value")?); + } + + let first = decode_handshake_from_bytes(&replies[0])?; + let second = decode_handshake_from_bytes(&replies[1])?; + assert_eq!( + first.noise, second.noise, + "both copies of one handshake must get the same Noise reply" + ); + + // And that reply completes the client's handshake with the server's key. + let payload = nw.recv(&first.noise)?; + assert_eq!(payload.error, 0); + assert_eq!(nw.finalize()?.remote_public_key, server_pk); + + server.destroy().await?; + client.destroy().await?; + bs.destroy().await?; + Ok(()) +} diff --git a/tests/node/hyperswarm-server.js b/tests/node/hyperswarm-server.js index 1618193..1148f58 100644 --- a/tests/node/hyperswarm-server.js +++ b/tests/node/hyperswarm-server.js @@ -5,6 +5,16 @@ const Hyperswarm = require('hyperswarm') const b4a = require('b4a') const readline = require('readline') +// Optional flags, used by the Node-client tests (no flag: unchanged behaviour). +// --remote-client the client skips hyperdht's same-host LAN shortcut, as a +// client on another network would; every node here is on +// 127.0.0.1, so it would otherwise always take it. One more +// DHT node is started: a holepunching client needs 3 NAT +// samples (hyperdht lib/nat.js) before it will punch. +// --open-client the client's DHT node is reachable (firewalled: false), +// like a peer with a public address. +const flags = new Set(process.argv.slice(2)) + async function main () { const bootstrap = new DHT({ ephemeral: false, @@ -22,9 +32,25 @@ async function main () { await node1.fullyBootstrapped() await node2.fullyBootstrapped() - const serverDht = new DHT({ host: '127.0.0.1', bootstrap: bsAddr }) + const extraNodes = [] + if (flags.has('--remote-client')) { + const node3 = new DHT({ ephemeral: false, firewalled: false, host: '127.0.0.1', bootstrap: bsAddr }) + await node3.fullyBootstrapped() + extraNodes.push(node3) + } + + const serverDht = new DHT({ + host: '127.0.0.1', + bootstrap: bsAddr, + ...(flags.has('--open-client') ? { ephemeral: false, firewalled: false } : {}) + }) await serverDht.fullyBootstrapped() + if (flags.has('--remote-client')) { + const connect = serverDht.connect.bind(serverDht) + serverDht.connect = (publicKey, opts = {}) => connect(publicKey, { ...opts, localConnection: false }) + } + const swarm = new Hyperswarm({ dht: serverDht }) let connResolve = null @@ -127,6 +153,9 @@ async function main () { async function cleanup () { try { await swarm.destroy() } catch (_) {} + for (const node of extraNodes) { + try { await node.destroy() } catch (_) {} + } try { await node2.destroy() } catch (_) {} try { await node1.destroy() } catch (_) {} try { await bootstrap.destroy() } catch (_) {} From bbefa8b6489bbffd054a8e0b12a039fc97c6a292 Mon Sep 17 00:00:00 2001 From: eshork <1829176+eshork@users.noreply.github.com> Date: Sun, 1 Nov 2026 21:00:40 -0500 Subject: [PATCH 3/8] fix(swarm): dial open/direct clients and bound the handshake reply cache Follow-up corrections to the Node-client-to-firewalled-server fix: - Open or direct clients: the server now dials the client (its stream connects to the client's address, so it sends the first packet and opens its own NAT mapping), as Node's `hs.onsocket(...)` does. A firewalled server keeps advertising `holepunch { id, relays }` to such a client, as Node does, so an open Node client waits for that dial instead of dialling the firewalled server itself. - Handshake reply cache: keyed by a hash of the client's Noise message, expired in insertion order (amortised O(1)) and by a periodic sweep, capped at 1024 entries (oldest evicted first), and skips handshakes the server rejects. - Tests: the interop tests wait until the Rust server classifies itself as firewalled and check Node's punch counters to prove which side dialled; the open client is now ephemeral so it really goes through relays; a new direct-client test covers the direct path; the remote client reports FIREWALL_UNKNOWN deterministically. A new Rust-only test sends one handshake through two real relays to a firewalled server and checks that the staged holepunch session for the first reply survives the duplicate. Cache unit tests cover first-reply-wins, the capacity bound and idle expiry. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- TESTING.md | 16 +- peeroxide/CHANGELOG.md | 4 +- peeroxide/src/swarm.rs | 231 ++++++++++++++++++---- peeroxide/tests/hyperswarm_interop.rs | 96 ++++++++- peeroxide/tests/server_handshake_dedup.rs | 212 ++++++++++++++++++-- tests/node/hyperswarm-server.js | 29 ++- 6 files changed, 505 insertions(+), 83 deletions(-) diff --git a/TESTING.md b/TESTING.md index 2f22864..3d61814 100644 --- a/TESTING.md +++ b/TESTING.md @@ -163,17 +163,19 @@ Passing result: Protomux framing and blind-relay control messages are byte-ident | Suite | Command | Type | Count | |-------|---------|------|-------| -| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 26 | -| Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 3 | -| Duplicate handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 1 | +| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 28 | +| Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 4 | +| Duplicate handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 2 | -**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut, as on another network, and one more DHT node is started so it can classify its NAT) and `--open-client` (the Node client's DHT node is reachable, like a peer with a public address). +**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut and reports `FIREWALL_UNKNOWN`, as a client behind a NAT on another network would; one more DHT node is started so it can classify its NAT) `--open-client` (the Node client's DHT node is reachable, like a peer with a public address, but ephemeral, so it reaches the server through relays) and `--direct-client` (reachable and persistent: on this small testnet it stores the server's announce and sends its handshake straight to the server). A `stats` command returns the Node DHT's punch counters (`punches.open` / `consistent` / `random`), which the tests use to check which side dialled. Passing result: - `hyperswarm_cross_language_connect`: Rust Hyperswarm client connects to a Node.js `hyperswarm` server over the DHT. Full flow: DHT bootstrap → topic announce/lookup → PEER_HANDSHAKE → skip-holepunching (same-host) → UDX stream → SecretStream encrypted channel → receive "hello from node". Verifies end-to-end interoperability of the complete Hyperswarm stack. -- `hyperswarm_node_client_holepunches_to_rust_server`: a firewalled Node.js client connects to a firewalled Rust server. The client sends one PEER_HANDSHAKE through two relays in parallel, then holepunches through the relays the server advertised; data flows both ways. -- `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node.js client waits for the firewalled Rust server to dial it, as a Node server would; data flows both ways. -- `server_handshake_dedup`: a Rust client sends the same PEER_HANDSHAKE twice to a swarm server and gets byte-identical replies. +- `hyperswarm_node_client_holepunches_to_rust_server`: a firewalled Node.js client connects to a firewalled Rust server. The test first waits for the server to classify itself as firewalled (`CONSISTENT`/`RANDOM`). The client sends one PEER_HANDSHAKE through two relays in parallel, then holepunches through the relays the server advertised (Node's punch counters confirm a holepunch and no direct dial); data flows both ways. +- `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node.js client reaches the firewalled Rust server through relays, sees an advertised holepunch, and waits for the server to dial it, as a Node server would. Node's punch counters confirm the client neither dialled nor holepunched; on loopback a client dial would also succeed, so this is what shows the server dialled. +- `hyperswarm_direct_node_client_connects_to_rust_server`: a reachable, persistent Node.js client sends its handshake straight to the Rust server (no relay); both sides dial and data flows both ways. +- `server_handshake_dedup::duplicate_peer_handshake_gets_the_same_reply`: a Rust client sends the same PEER_HANDSHAKE twice straight to a swarm server and gets byte-identical replies. +- `server_handshake_dedup::duplicate_through_two_relays_keeps_the_staged_holepunch`: the same PEER_HANDSHAKE reaches a firewalled swarm server through two of its relays. Both copies get the same reply, and a PEER_HOLEPUNCH for the first reply's id is still answered, so the second copy did not replace the staged holepunch session. ### Live Network Validation diff --git a/peeroxide/CHANGELOG.md b/peeroxide/CHANGELOG.md index 08a09a4..3c3d646 100644 --- a/peeroxide/CHANGELOG.md +++ b/peeroxide/CHANGELOG.md @@ -10,9 +10,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Node.js Hyperswarm clients can now connect to a firewalled peeroxide server. Before, every attempt timed out (`passive holepunch timed out` here, `HOLEPUNCH_ABORTED` on the Node side). The server side of the handshake now matches Node hyperdht `lib/server.js` on three points: - - **Duplicate PEER_HANDSHAKE.** A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Each copy now gets the same reply, cached by the Noise message for 20 s. Before, each copy got a fresh Noise responder, and the second replaced the first's holepunch session. + - **Duplicate PEER_HANDSHAKE.** A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Each copy now gets the same reply, cached by a hash of the Noise message for 20 s. Before, each copy got a fresh Noise responder, and the second replaced the first's holepunch session. The cache holds at most 1024 replies (oldest evicted first), expires them in insertion order and on a periodic sweep, and skips handshakes the server rejects, so a flood of distinct handshakes cannot grow it or stall the swarm. - **Holepunch relays.** The reply's `holepunch.relays[].peer_address` is now our own address as each relay saw it (`HyperDhtHandle::current_relays()`). Before, it held the client's address, so the relay sent the client's PEER_HOLEPUNCH back to the client. - - **Open or direct clients.** A client that says it is `FIREWALL_OPEN`, or that reached us without a relay, is now dialled directly, as Node does. Before, it was given a holepunch, while the Node client waited for that dial. + - **Open or direct clients.** A client that says it is `FIREWALL_OPEN`, or that reached us without a relay, is now dialled by the server: its stream connects to the client's address, so the server sends the first packet and opens its own NAT mapping, as Node's `hs.onsocket(...)` does. A firewalled server still advertises `holepunch { id, relays }` to such a client, as Node does, so an open Node client waits for that dial instead of dialling the firewalled server itself. Before, the open client was given a passive holepunch it never performs. ## [1.7.3](https://github.com/Rightbracket/peeroxide/compare/peeroxide-v1.7.2...peeroxide-v1.7.3) - 2026-07-12 diff --git a/peeroxide/src/swarm.rs b/peeroxide/src/swarm.rs index 9efe5bc..5be8f4d 100644 --- a/peeroxide/src/swarm.rs +++ b/peeroxide/src/swarm.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::fmt; use std::net::SocketAddr; use std::sync::atomic::{AtomicU32, Ordering}; @@ -49,6 +49,13 @@ const DEFAULT_MAX_PARALLEL: usize = 3; /// `lib/server.js`). const HANDSHAKE_REPLY_TTL: Duration = Duration::from_secs(20); +/// Upper bound on cached handshake replies. Anyone who knows our announced +/// public key can send distinct valid handshakes, so the cache must not grow +/// with them. Past this bound the oldest reply is evicted first: a flood can +/// then only cost a legitimate client its duplicate-dedup, never block new +/// handshakes. Legitimate traffic stays far below it (each entry lives 20 s). +const HANDSHAKE_REPLY_CACHE_CAPACITY: usize = 1024; + // ── Retry backoff tiers (matching Node.js lib/retry-timer.js) ──────────────── // Each tier: [base_ms, jitter1, jitter2, jitter3] // Delay = base + rand(0..j1) + rand(0..j2) + rand(0..j3) @@ -373,33 +380,68 @@ struct SwarmActor { /// first, so a server must answer every copy identically, as Node's /// `Server._onpeerhandshake` does. Answering each copy with a fresh Noise /// responder gives the client a reply for a session the server then replaces. +/// +/// Entries are keyed by a hash of the client's Noise message and expire in +/// insertion order, so pruning is amortised O(1) per handshake. At most +/// `capacity` entries are kept; the oldest is evicted first. struct HandshakeReplyCache { ttl: Duration, - entries: HashMap, (tokio::time::Instant, Vec)>, + capacity: usize, + entries: HashMap<[u8; 32], Vec>, + order: VecDeque<(tokio::time::Instant, [u8; 32])>, } impl HandshakeReplyCache { - fn new(ttl: Duration) -> Self { + fn new(ttl: Duration, capacity: usize) -> Self { Self { ttl, + capacity, entries: HashMap::new(), + order: VecDeque::new(), } } /// The Noise reply already sent for this client Noise message, if any. fn get(&mut self, client_noise: &[u8]) -> Option> { self.prune(); - self.entries.get(client_noise).map(|(_, reply)| reply.clone()) + self.entries.get(&hash(client_noise)).cloned() } - fn insert(&mut self, client_noise: Vec, reply_noise: Vec) { + /// Record the reply to a client Noise message. The first reply wins: a + /// later insert for the same message is ignored. + fn insert(&mut self, client_noise: &[u8], reply_noise: Vec) { self.prune(); - self.entries.insert(client_noise, (tokio::time::Instant::now(), reply_noise)); + let key = hash(client_noise); + if self.entries.contains_key(&key) || self.capacity == 0 { + return; + } + while self.entries.len() >= self.capacity { + let Some((_, oldest)) = self.order.pop_front() else { break }; + self.entries.remove(&oldest); + } + self.order.push_back((tokio::time::Instant::now(), key)); + self.entries.insert(key, reply_noise); } + /// Drop expired entries. Also run periodically by the actor so replies do + /// not outlive their TTL while no handshakes arrive. fn prune(&mut self) { - let ttl = self.ttl; - self.entries.retain(|_, (at, _)| at.elapsed() < ttl); + while let Some((at, key)) = self.order.front() { + if at.elapsed() < self.ttl { + break; + } + self.entries.remove(key); + self.order.pop_front(); + } + if self.order.is_empty() && self.entries.capacity() > 0 { + self.entries = HashMap::new(); + self.order = VecDeque::new(); + } + } + + #[cfg(test)] + fn len(&self) -> usize { + self.entries.len() } } @@ -566,7 +608,10 @@ pub async fn spawn( pending_handshakes: HashMap::new(), next_holepunch_id: 0, passive_hp_event_tx, - handshake_replies: HandshakeReplyCache::new(HANDSHAKE_REPLY_TTL), + handshake_replies: HandshakeReplyCache::new( + HANDSHAKE_REPLY_TTL, + HANDSHAKE_REPLY_CACHE_CAPACITY, + ), }; // Keep the DHT runtime alive for the swarm's lifetime. @@ -607,6 +652,9 @@ impl SwarmActor { mpsc::unbounded_channel::<[u8; 32]>(); self.server_failure_tx = Some(server_failure_tx); + let mut reply_sweep = tokio::time::interval(HANDSHAKE_REPLY_TTL); + reply_sweep.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { tokio::select! { cmd = cmd_rx.recv() => { @@ -642,6 +690,9 @@ impl SwarmActor { self.handle_passive_holepunch_event(event).await; } } + _ = reply_sweep.tick() => { + self.handshake_replies.prune(); + } } } @@ -1067,23 +1118,24 @@ impl SwarmActor { (None, None) }; - // Decide whether this reply will stage a passive holepunch: - // - the DHT has settled on a firewalled classification - // (CONSISTENT or RANDOM — not OPEN, not UNKNOWN); UNKNOWN - // means NAT analysis hasn't gathered enough samples to be - // sure, in which case the existing direct path is the safer - // choice (Node hyperdht has the same gate), AND - // - we are not forcing all server connections through a relay - // (relay_through bypasses NAT traversal entirely), AND - // - the client cannot simply be dialled. Node `lib/server.js - // _addHandshake` dials a client that says it is FIREWALL_OPEN, or - // that reached us directly rather than through a relay, and such a - // client waits for that dial instead of holepunching. + // Decide how this handshake will connect: + // - `server_firewalled`: the DHT has settled on a firewalled + // classification (CONSISTENT or RANDOM — not OPEN, not UNKNOWN; + // UNKNOWN means NAT analysis hasn't gathered enough samples, and + // the direct path is the safer choice, as in Node hyperdht), and + // we are not forcing all server connections through a relay + // (relay_through bypasses NAT traversal entirely). + // - `client_dialable`: the client says it is FIREWALL_OPEN, or it + // reached us directly rather than through a relay. Node `lib/server.js + // _addHandshake` dials such a client itself (`hs.onsocket(...)`) + // right after replying, so we send the first packet and open our + // own NAT mapping towards it. + // - Otherwise, if we are firewalled, stage a passive holepunch. let firewall_state = self.dht.firewalled().await.unwrap_or(FIREWALL_UNKNOWN); + let server_firewalled = matches!(firewall_state, FIREWALL_CONSISTENT | FIREWALL_RANDOM) + && relay_through_info.is_none(); let client_dialable = remote_payload.firewall == FIREWALL_OPEN || !is_forwarded; - let attempt_holepunch = matches!(firewall_state, FIREWALL_CONSISTENT | FIREWALL_RANDOM) - && relay_through_info.is_none() - && !client_dialable; + let attempt_holepunch = server_firewalled && !client_dialable; // Try to stage the passive Holepuncher up front so the reply we // send in this handshake can advertise the puncher's bound port @@ -1124,10 +1176,28 @@ impl SwarmActor { // A client sends its PEER_HOLEPUNCH to `relay_address`, and the relay // forwards it to `peer_address`: that must be our own address as the // relay saw it (Node `Announcer.relays`), not the client's. - let holepunch_info = holepunch_setup.as_ref().map(|setup| HolepunchInfo { - id: setup.id, - relays: self.dht.current_relays(), - }); + // + // A firewalled Node server advertises `{ id, relays }` even to a + // client it is about to dial (`holepunch: ourRemoteAddr ? null : + // {...}`): an open Node client that sees it waits for our dial + // (`lib/connect.js holepunch`) instead of dialling our firewalled + // address. That id has no staged puncher; a PEER_HOLEPUNCH for it gets + // no reply, as Node's `_onpeerholepunch` aborts one without a puncher. + let holepunch_info = match &holepunch_setup { + Some(setup) => Some(HolepunchInfo { + id: setup.id, + relays: self.dht.current_relays(), + }), + None if server_firewalled && client_dialable => { + let id = self.next_holepunch_id; + self.next_holepunch_id = self.next_holepunch_id.wrapping_add(1); + Some(HolepunchInfo { + id, + relays: self.dht.current_relays(), + }) + } + None => None, + }; let reply_payload = NoisePayload { version: 1, @@ -1176,15 +1246,24 @@ impl SwarmActor { } }; - self.handshake_replies.insert(msg.noise.clone(), noise_reply.clone()); + let remote_pk = nw_result.remote_public_key; + + // Remember the reply only for a handshake we act on: a staged + // holepunch, or a connection the checks below will let through. + // Rejected handshakes must not take up cache space. + let accepted = holepunch_setup.is_some() + || (!self.connections.has(&remote_pk) + && self.connections.len() < self.config.max_peers + && remote_payload.udx.is_some()); + if accepted { + self.handshake_replies.insert(&msg.noise, noise_reply.clone()); + } let _ = reply_tx.send(encode_server_handshake_reply( msg.mode, peer_address.clone(), noise_reply, )); - let remote_pk = nw_result.remote_public_key; - // If we staged a passive holepunch above, hand ownership of the // staged state into the actor's `connects` map and return. Slot // accounting (self.connections.add) is deferred until the firewall @@ -1259,8 +1338,16 @@ impl SwarmActor { let client_addr_for_task = client_address.clone(); let failure_tx = self.server_failure_tx.clone(); tokio::spawn(async move { - match create_server_connection(rh, dht, local_stream_id, &remote_udx, &client_addr_for_task, &nw_result) - .await + match create_server_connection( + rh, + dht, + local_stream_id, + &remote_udx, + &client_addr_for_task, + client_dialable, + &nw_result, + ) + .await { Ok((conn, runtime)) => { let swarm_conn = SwarmConnection { @@ -1662,12 +1749,20 @@ struct HolepunchSetup { abort_task: JoinHandle<()>, } +/// Open the server side of a direct (non-relayed, non-holepunched) UDX stream. +/// +/// With `dial` set, the stream connects to `client_address` straight away, so +/// our SecretStream header is the first packet and opens our NAT mapping +/// towards the client, as Node's `hs.onsocket(...)` does for an open or direct +/// client. Otherwise the stream listens and adopts the source of the first +/// packet that arrives for it. async fn create_server_connection( runtime_handle: Arc, dht: HyperDhtHandle, local_stream_id: u32, remote_udx: &UdxInfo, - _client_address: &Ipv4Peer, + client_address: &Ipv4Peer, + dial: bool, noise_result: &peeroxide_dht::noise_wrap::NoiseWrapResult, ) -> Result<(PeerConnection, UdxRuntime), SwarmError> { let runtime = UdxRuntime::shared(runtime_handle); @@ -1689,7 +1784,18 @@ async fn create_server_connection( })?; let stream = runtime.create_stream(local_stream_id).await?; - stream.set_firewall_hook(&socket, remote_id, |_, _, _| true)?; + if dial { + let ip = client_address.host.parse().map_err(|e| { + SwarmError::Dht(peeroxide_dht::hyperdht::HyperDhtError::StreamEstablishment( + format!("invalid client address {}: {e}", client_address.host), + )) + })?; + stream + .connect(&socket, remote_id, SocketAddr::new(ip, client_address.port)) + .await?; + } else { + stream.set_firewall_hook(&socket, remote_id, |_, _, _| true)?; + } let async_stream = stream.into_async_stream(); let ss = SecretStream::from_session( @@ -1877,19 +1983,68 @@ mod tests { #[tokio::test(start_paused = true)] async fn handshake_reply_cache_returns_the_first_reply_until_it_expires() { - let mut cache = HandshakeReplyCache::new(Duration::from_secs(20)); + let mut cache = HandshakeReplyCache::new(Duration::from_secs(20), 16); assert!(cache.get(b"noise-a").is_none()); - cache.insert(b"noise-a".to_vec(), b"reply-a".to_vec()); + cache.insert(b"noise-a", b"reply-a".to_vec()); assert_eq!(cache.get(b"noise-a").as_deref(), Some(&b"reply-a"[..])); assert!(cache.get(b"noise-b").is_none(), "other handshakes are not affected"); + cache.insert(b"noise-a", b"reply-other".to_vec()); + assert_eq!( + cache.get(b"noise-a").as_deref(), + Some(&b"reply-a"[..]), + "the first reply wins" + ); + tokio::time::advance(Duration::from_secs(19)).await; assert_eq!(cache.get(b"noise-a").as_deref(), Some(&b"reply-a"[..])); tokio::time::advance(Duration::from_secs(2)).await; assert!(cache.get(b"noise-a").is_none()); - assert!(cache.entries.is_empty(), "expired entries are dropped"); + assert_eq!(cache.len(), 0, "expired entries are dropped"); + assert!(cache.order.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn handshake_reply_cache_evicts_the_oldest_entry_at_capacity() { + let mut cache = HandshakeReplyCache::new(Duration::from_secs(20), 3); + for i in 0u8..3 { + cache.insert(&[i], vec![i]); + tokio::time::advance(Duration::from_millis(1)).await; + } + assert_eq!(cache.len(), 3); + + cache.insert(&[3], vec![3]); + assert_eq!(cache.len(), 3, "the cache never grows past its capacity"); + assert!(cache.get(&[0]).is_none(), "the oldest entry is evicted first"); + for i in 1u8..=3 { + assert_eq!(cache.get(&[i]), Some(vec![i])); + } + + // A flood of distinct handshakes stays bounded. + for i in 0u32..10_000 { + cache.insert(&i.to_le_bytes(), vec![0; 8]); + } + assert_eq!(cache.len(), 3); + assert_eq!(cache.order.len(), 3); + } + + #[tokio::test(start_paused = true)] + async fn handshake_reply_cache_prune_expires_entries_without_new_handshakes() { + let mut cache = HandshakeReplyCache::new(Duration::from_secs(20), 16); + cache.insert(b"noise-a", b"reply-a".to_vec()); + tokio::time::advance(Duration::from_secs(10)).await; + cache.insert(b"noise-b", b"reply-b".to_vec()); + + tokio::time::advance(Duration::from_secs(11)).await; + cache.prune(); + assert_eq!(cache.len(), 1, "only the expired entry is dropped"); + + tokio::time::advance(Duration::from_secs(10)).await; + cache.prune(); + assert_eq!(cache.len(), 0); + assert_eq!(cache.entries.capacity(), 0, "an empty cache releases its storage"); } #[test] diff --git a/peeroxide/tests/hyperswarm_interop.rs b/peeroxide/tests/hyperswarm_interop.rs index 976e530..0006e2a 100644 --- a/peeroxide/tests/hyperswarm_interop.rs +++ b/peeroxide/tests/hyperswarm_interop.rs @@ -14,15 +14,23 @@ //! PEER_HANDSHAKE through two relays in parallel, then holepunches through the //! relays listed in the server's reply. //! `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node client -//! waits for the server to dial it, as Node servers do, instead of holepunching. +//! reaches the server through relays and is told the server is holepunchable, +//! so it waits for the server to dial it, as Node servers do. Node's punch +//! counters show it never dialled the server itself. (On loopback that dial +//! would also succeed, so without the counters the test could not tell who +//! dialled.) +//! `hyperswarm_direct_node_client_connects_to_rust_server`: a reachable, +//! persistent Node client stores the server's announce and sends its handshake +//! straight to the server; both sides dial. use std::io::{BufRead, Write}; use std::process::{Command, Stdio}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use peeroxide::{discovery_key, spawn, JoinOpts, SwarmConfig}; +use peeroxide::{discovery_key, spawn, JoinOpts, SwarmConfig, SwarmHandle}; use peeroxide_dht::hyperdht::HyperDhtConfig; +use peeroxide_dht::hyperdht_messages::{FIREWALL_CONSISTENT, FIREWALL_RANDOM}; use peeroxide_dht::rpc::DhtConfig; use serde_json::Value; @@ -170,6 +178,15 @@ impl JsSwarm { ); } + /// The Node DHT's punch counters: `(open, consistent + random)`. + fn punches(&mut self) -> (u64, u64) { + let id = self.send_cmd(serde_json::json!({ "cmd": "stats" })); + let reply = self.wait_reply(id); + let p = &reply["punches"]; + let count = |k: &str| p[k].as_u64().unwrap_or_else(|| panic!("no punches.{k}: {reply}")); + (count("open"), count("consistent") + count("random")) + } + /// Wait for a `data` event whose hex payload decodes to `expected`. fn wait_data(&self, expected: &str) -> Value { let want = hex::encode(expected.as_bytes()); @@ -294,19 +311,39 @@ async fn run_interop() -> Result<(), Box> { #[tokio::test] async fn hyperswarm_node_client_holepunches_to_rust_server() { - node_client_interop(&["--remote-client"]).await; + node_client_interop(&["--remote-client"], ExpectedPath::Holepunch).await; } #[tokio::test] async fn hyperswarm_open_node_client_connects_to_rust_server() { - node_client_interop(&["--open-client"]).await; + node_client_interop(&["--open-client"], ExpectedPath::ServerDials).await; +} + +#[tokio::test] +async fn hyperswarm_direct_node_client_connects_to_rust_server() { + node_client_interop(&["--direct-client"], ExpectedPath::Direct).await; +} + +/// How the Node client is expected to reach the firewalled Rust server. +#[derive(Debug, Clone, Copy, PartialEq)] +enum ExpectedPath { + /// The client holepunches through the relays in the server's reply. + Holepunch, + /// The client waits and the server dials it. + ServerDials, + /// The handshake went straight to the server, not through a relay. Both + /// sides dial (`lib/connect.js connectThroughNode`), no punch is counted. + Direct, } -async fn node_client_interop(flags: &[&str]) { +async fn node_client_interop(flags: &[&str], expected: ExpectedPath) { let _ = tracing_subscriber::fmt::try_init(); - let result = - tokio::time::timeout(Duration::from_secs(90), run_node_client_interop(flags)).await; + let result = tokio::time::timeout( + Duration::from_secs(90), + run_node_client_interop(flags, expected), + ) + .await; match result { Ok(Ok(())) => {} @@ -315,7 +352,27 @@ async fn node_client_interop(flags: &[&str]) { } } -async fn run_node_client_interop(flags: &[&str]) -> Result<(), Box> { +/// Wait until the swarm's DHT classifies itself as firewalled. Until then the +/// server takes the direct path for every client, and a test meant to cover +/// the firewalled server would pass without exercising it. +async fn wait_until_firewalled(handle: &SwarmHandle) -> Result> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let state = handle.dht().firewalled().await?; + if state == FIREWALL_CONSISTENT || state == FIREWALL_RANDOM { + return Ok(state); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("Rust server never classified itself as firewalled (state {state})").into()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +async fn run_node_client_interop( + flags: &[&str], + expected: ExpectedPath, +) -> Result<(), Box> { let mut js = JsSwarm::start_with(flags); let (bs_port, _node_pk) = js.read_ready(); tracing::info!(bs_port, "JS testnet ready"); @@ -323,8 +380,9 @@ async fn run_node_client_interop(flags: &[&str]) -> Result<(), Box Result<(), Box Result<(), Box { + assert_eq!(open, 0, "the Node client should not have dialled the server directly"); + assert!(punched >= 1, "the Node client should have holepunched"); + } + ExpectedPath::ServerDials | ExpectedPath::Direct => { + assert_eq!( + (open, punched), + (0, 0), + "the Node client should not have dialled or holepunched ({expected:?})" + ); + } + } + handle.destroy().await?; drop(js); diff --git a/peeroxide/tests/server_handshake_dedup.rs b/peeroxide/tests/server_handshake_dedup.rs index 6d8ed0c..60b34e8 100644 --- a/peeroxide/tests/server_handshake_dedup.rs +++ b/peeroxide/tests/server_handshake_dedup.rs @@ -3,22 +3,33 @@ //! Node hyperdht clients send one Noise handshake message through up to two //! relays in parallel and act on the first reply that arrives (`lib/connect.js`), //! so a server must answer each copy identically, as Node's -//! `Server._onpeerhandshake` does. Here a Rust client sends the same message -//! twice and checks that both replies are byte-identical and usable. +//! `Server._onpeerhandshake` does. +//! +//! - `duplicate_peer_handshake_gets_the_same_reply`: a Rust client sends the +//! same message straight to the server twice and checks that both replies +//! are byte-identical and usable. +//! - `duplicate_through_two_relays_keeps_the_staged_holepunch`: the same +//! message reaches a firewalled server through two of its relays, so the +//! first copy stages a passive holepunch. The second copy must not replace +//! that session: a PEER_HOLEPUNCH for the first reply's id must still be +//! answered. use std::time::Duration; use libudx::UdxRuntime; -use peeroxide::{spawn as spawn_swarm, JoinOpts, SwarmConfig}; +use peeroxide::{spawn as spawn_swarm, JoinOpts, SwarmConfig, SwarmHandle}; use peeroxide_dht::crypto::hash; use peeroxide_dht::hyperdht::{spawn, HyperDhtConfig, KeyPair}; use peeroxide_dht::hyperdht_messages::{ - decode_handshake_from_bytes, NoisePayload, UdxInfo, FIREWALL_UNKNOWN, PEER_HANDSHAKE, + decode_handshake_from_bytes, decode_holepunch_msg_from_bytes, HolepunchPayload, NoisePayload, + RelayInfo, UdxInfo, ERROR_NONE, FIREWALL_CONSISTENT, FIREWALL_RANDOM, FIREWALL_UNKNOWN, + PEER_HANDSHAKE, PEER_HOLEPUNCH, }; use peeroxide_dht::messages::Ipv4Peer; use peeroxide_dht::noise::Keypair as NoiseKeypair; use peeroxide_dht::noise_wrap::NoiseWrap; use peeroxide_dht::rpc::{DhtConfig, UserRequestParams}; +use peeroxide_dht::secure_payload::SecurePayload; use peeroxide_dht::Router; fn local_dht(bootstrap: Vec, firewalled: bool) -> HyperDhtConfig { @@ -27,6 +38,11 @@ fn local_dht(bootstrap: Vec, firewalled: bool) -> HyperDhtConfig { dht.port = 0; dht.host = "127.0.0.1".to_string(); dht.firewalled = firewalled; + if !firewalled { + // A reachable node joins the routing table at once, as the Node + // testnet nodes (`ephemeral: false`) do. + dht.ephemeral = Some(false); + } let mut config = HyperDhtConfig::default(); config.dht = dht; config @@ -79,23 +95,7 @@ async fn run() -> Result<(), Box> { }, server_pk, ); - let noise = nw.send(&NoisePayload { - version: 1, - error: 0, - firewall: FIREWALL_UNKNOWN, - holepunch: None, - addresses4: vec![], - addresses6: vec![], - udx: Some(UdxInfo { - version: 1, - reusable_socket: false, - id: 1, - seq: 0, - }), - secret_stream: None, - relay_through: None, - relay_addresses: None, - })?; + let noise = nw.send(&client_noise_payload())?; let request = Router::encode_client_handshake(noise, None, Some(server_addr.clone()))?; let mut replies = Vec::new(); @@ -129,3 +129,173 @@ async fn run() -> Result<(), Box> { bs.destroy().await?; Ok(()) } + +fn client_noise_payload() -> NoisePayload { + NoisePayload { + version: 1, + error: 0, + firewall: FIREWALL_UNKNOWN, + holepunch: None, + addresses4: vec![], + addresses6: vec![], + udx: Some(UdxInfo { + version: 1, + reusable_socket: false, + id: 1, + seq: 0, + }), + secret_stream: None, + relay_through: None, + relay_addresses: None, + } +} + +async fn wait_for( + what: &str, + mut probe: impl AsyncFnMut() -> Option, +) -> Result> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + if let Some(v) = probe().await { + return Ok(v); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("timed out waiting for {what}").into()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +async fn relays_of(server: &SwarmHandle) -> Option> { + let relays = server.dht().current_relays(); + (relays.len() >= 2).then_some(relays) +} + +#[tokio::test] +async fn duplicate_through_two_relays_keeps_the_staged_holepunch() { + let _ = tracing_subscriber::fmt::try_init(); + + let result = tokio::time::timeout(Duration::from_secs(60), run_relayed()).await; + match result { + Ok(Ok(())) => {} + Ok(Err(e)) => panic!("relayed duplicate handshake test failed: {e}"), + Err(_) => panic!("relayed duplicate handshake test timed out after 60s"), + } +} + +async fn run_relayed() -> Result<(), Box> { + let rt = UdxRuntime::new()?; + + // Reachable DHT nodes: they store the server's announce, so they relay + // handshakes to it, and they give the firewalled server the NAT samples + // it needs to classify itself. + let (_bs_join, bs, _bs_rx) = spawn(&rt, local_dht(vec![], false)).await?; + let bootstrap = vec![format!("127.0.0.1:{}", bs.local_port().await?)]; + let mut nodes = Vec::new(); + for _ in 0..3 { + let (join, node, rx) = spawn(&rt, local_dht(bootstrap.clone(), false)).await?; + node.bootstrapped().await?; + nodes.push((join, node, rx)); + } + + let mut config = SwarmConfig::default(); + config.dht = local_dht(bootstrap.clone(), true); + let (_task, server, _conn_rx) = spawn_swarm(config).await?; + server + .join([0x5b; 32], { + let mut opts = JoinOpts::default(); + opts.client = false; + opts + }) + .await?; + server.flush().await?; + let server_pk = server.key_pair().public_key; + let target = hash(&server_pk); + + wait_for("the server to classify itself as firewalled", async || { + let state = server.dht().firewalled().await.ok()?; + (state == FIREWALL_CONSISTENT || state == FIREWALL_RANDOM).then_some(()) + }) + .await?; + let relays = wait_for("two announce relays", async || relays_of(&server).await).await?; + + let (_cli_join, client, _cli_rx) = spawn(&rt, local_dht(bootstrap, true)).await?; + client.bootstrapped().await?; + + let client_kp = KeyPair::generate(); + let mut nw = NoiseWrap::new_initiator( + NoiseKeypair { + public_key: client_kp.public_key, + secret_key: client_kp.secret_key, + }, + server_pk, + ); + let request = Router::encode_client_handshake(nw.send(&client_noise_payload())?, None, None)?; + + // One copy through each relay, the second after the first has been + // answered and its passive holepunch staged. + let mut replies = Vec::new(); + for relay in &relays[..2] { + let mut params = UserRequestParams::default(); + params.command = PEER_HANDSHAKE; + params.target = Some(target); + params.value = Some(request.clone()); + let addr = &relay.relay_address; + let resp = client.dht().request(params, &addr.host, addr.port).await?; + assert_eq!(resp.error, 0, "relay {addr:?} returned error code {}", resp.error); + replies.push(decode_handshake_from_bytes( + &resp.value.ok_or("handshake reply without a value")?, + )?); + } + assert_eq!( + replies[0].noise, replies[1].noise, + "both relayed copies of one handshake must get the same Noise reply" + ); + + let payload = nw.recv(&replies[0].noise)?; + assert_eq!(payload.error, 0); + let holepunch = payload + .holepunch + .ok_or("a firewalled server should offer a holepunch to a firewalled client")?; + assert!(!holepunch.relays.is_empty()); + let noise = nw.finalize()?; + + // The session staged for the first copy must still exist. + let secure = SecurePayload::new(noise.holepunch_secret); + let probe = HolepunchPayload { + error: ERROR_NONE, + firewall: FIREWALL_UNKNOWN, + round: 0, + connected: false, + punching: false, + addresses: Some(vec![]), + remote_address: None, + token: Some([7; 32]), + remote_token: None, + }; + let value = Router::encode_client_holepunch(holepunch.id, secure.encrypt(&probe)?, None)?; + let mut params = UserRequestParams::default(); + params.command = PEER_HOLEPUNCH; + params.target = Some(target); + params.value = Some(value); + let addr: &Ipv4Peer = &relays[0].relay_address; + let resp = client.dht().request(params, &addr.host, addr.port).await?; + assert_eq!(resp.error, 0, "holepunch returned error code {}", resp.error); + let reply = decode_holepunch_msg_from_bytes( + &resp + .value + .ok_or("no holepunch reply: the first handshake's session was replaced")?, + )?; + assert_eq!(reply.id, holepunch.id); + let reply_payload = secure.decrypt(&reply.payload)?; + assert_eq!(reply_payload.error, ERROR_NONE); + assert_eq!(reply_payload.remote_token, Some([7; 32])); + + server.destroy().await?; + client.destroy().await?; + for (_, node, _) in nodes { + node.destroy().await?; + } + bs.destroy().await?; + Ok(()) +} diff --git a/tests/node/hyperswarm-server.js b/tests/node/hyperswarm-server.js index 1148f58..4b5809b 100644 --- a/tests/node/hyperswarm-server.js +++ b/tests/node/hyperswarm-server.js @@ -6,13 +6,20 @@ const b4a = require('b4a') const readline = require('readline') // Optional flags, used by the Node-client tests (no flag: unchanged behaviour). -// --remote-client the client skips hyperdht's same-host LAN shortcut, as a -// client on another network would; every node here is on -// 127.0.0.1, so it would otherwise always take it. One more +// --remote-client the client skips hyperdht's same-host LAN shortcut and +// reports FIREWALL_UNKNOWN, as a NATed client on another +// network would; every node here is on 127.0.0.1, so it +// would otherwise take the shortcut. One more // DHT node is started: a holepunching client needs 3 NAT // samples (hyperdht lib/nat.js) before it will punch. // --open-client the client's DHT node is reachable (firewalled: false), -// like a peer with a public address. +// like a peer with a public address. It stays ephemeral, +// so it does not store the server's announce and reaches +// the server through relays. +// --direct-client the client's DHT node is reachable and persistent +// (ephemeral: false). On this small testnet it then stores +// the server's announce itself and sends the handshake +// straight to the server, which is the direct path. const flags = new Set(process.argv.slice(2)) async function main () { @@ -42,13 +49,18 @@ async function main () { const serverDht = new DHT({ host: '127.0.0.1', bootstrap: bsAddr, - ...(flags.has('--open-client') ? { ephemeral: false, firewalled: false } : {}) + ...(flags.has('--open-client') ? { firewalled: false } : {}), + ...(flags.has('--direct-client') ? { ephemeral: false, firewalled: false } : {}) }) await serverDht.fullyBootstrapped() if (flags.has('--remote-client')) { const connect = serverDht.connect.bind(serverDht) serverDht.connect = (publicKey, opts = {}) => connect(publicKey, { ...opts, localConnection: false }) + // Report FIREWALL_UNKNOWN, as a client behind a NAT would. On loopback, + // dht-rpc's NAT check can otherwise decide this node is reachable, and + // the client then says FIREWALL_OPEN and is dialled instead of punching. + serverDht.remoteAddress = () => null } const swarm = new Hyperswarm({ dht: serverDht }) @@ -137,6 +149,13 @@ async function main () { remotePk: b4a.toString(socket.remotePublicKey, 'hex') }) + '\n') + } else if (cmd === 'stats') { + // hyperdht counts how each connection was opened: `open` when the + // client dialled the server itself, `consistent`/`random` when it + // started a holepunch. Neither moves when the client waited for the + // server to dial it. + process.stdout.write(JSON.stringify({ id, ok: true, punches: serverDht.stats.punches }) + '\n') + } else if (cmd === 'shutdown') { await cleanup() process.exit(0) From dca6b64ecc534a5b31065ec8ea3459878d3194b5 Mon Sep 17 00:00:00 2001 From: eshork <1829176+eshork@users.noreply.github.com> Date: Sun, 1 Nov 2026 21:17:09 -0500 Subject: [PATCH 4/8] fix(swarm): release staged holepunches for rejected handshakes; tighten tests Review follow-ups on the previous commit: - A relayed handshake the server would reject (already connected, at max_peers, no UDX info) no longer keeps its staged passive holepunch or a reply-cache entry; the session would only have been dropped when its punch landed. `HolepunchSetup::release()` also frees the puncher on the Noise error paths, not just the abort timer. - `--direct-client` now reports FIREWALL_UNKNOWN, so the direct-client interop test can only pass through the direct (non-relayed) path. - New tests: a server at max_peers = 0 neither caches replies nor keeps a staged holepunch; a Rust client connecting straight to a firewalled swarm server still works now that the server dials it. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- TESTING.md | 8 +- peeroxide/CHANGELOG.md | 2 +- peeroxide/src/swarm.rs | 37 ++- peeroxide/tests/server_handshake_dedup.rs | 350 +++++++++++++++------- tests/node/hyperswarm-server.js | 13 +- 5 files changed, 291 insertions(+), 119 deletions(-) diff --git a/TESTING.md b/TESTING.md index 3d61814..8b18343 100644 --- a/TESTING.md +++ b/TESTING.md @@ -165,17 +165,19 @@ Passing result: Protomux framing and blind-relay control messages are byte-ident |-------|---------|------|-------| | Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 28 | | Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 4 | -| Duplicate handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 2 | +| Server handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 4 | -**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut and reports `FIREWALL_UNKNOWN`, as a client behind a NAT on another network would; one more DHT node is started so it can classify its NAT) `--open-client` (the Node client's DHT node is reachable, like a peer with a public address, but ephemeral, so it reaches the server through relays) and `--direct-client` (reachable and persistent: on this small testnet it stores the server's announce and sends its handshake straight to the server). A `stats` command returns the Node DHT's punch counters (`punches.open` / `consistent` / `random`), which the tests use to check which side dialled. +**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut and reports `FIREWALL_UNKNOWN`, as a client behind a NAT on another network would; one more DHT node is started so it can classify its NAT) `--open-client` (the Node client's DHT node is reachable, like a peer with a public address, but ephemeral, so it reaches the server through relays) and `--direct-client` (persistent but reporting `FIREWALL_UNKNOWN`: on this small testnet it stores the server's announce and sends its handshake straight to the server; through a relay it would have to holepunch). A `stats` command returns the Node DHT's punch counters (`punches.open` / `consistent` / `random`), which the tests use to check which side dialled. Passing result: - `hyperswarm_cross_language_connect`: Rust Hyperswarm client connects to a Node.js `hyperswarm` server over the DHT. Full flow: DHT bootstrap → topic announce/lookup → PEER_HANDSHAKE → skip-holepunching (same-host) → UDX stream → SecretStream encrypted channel → receive "hello from node". Verifies end-to-end interoperability of the complete Hyperswarm stack. - `hyperswarm_node_client_holepunches_to_rust_server`: a firewalled Node.js client connects to a firewalled Rust server. The test first waits for the server to classify itself as firewalled (`CONSISTENT`/`RANDOM`). The client sends one PEER_HANDSHAKE through two relays in parallel, then holepunches through the relays the server advertised (Node's punch counters confirm a holepunch and no direct dial); data flows both ways. - `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node.js client reaches the firewalled Rust server through relays, sees an advertised holepunch, and waits for the server to dial it, as a Node server would. Node's punch counters confirm the client neither dialled nor holepunched; on loopback a client dial would also succeed, so this is what shows the server dialled. -- `hyperswarm_direct_node_client_connects_to_rust_server`: a reachable, persistent Node.js client sends its handshake straight to the Rust server (no relay); both sides dial and data flows both ways. +- `hyperswarm_direct_node_client_connects_to_rust_server`: a persistent Node.js client that reports `FIREWALL_UNKNOWN` sends its handshake straight to the Rust server (no relay); both sides dial and data flows both ways. Node's punch counters confirm no holepunch, which a relayed attempt would have needed. - `server_handshake_dedup::duplicate_peer_handshake_gets_the_same_reply`: a Rust client sends the same PEER_HANDSHAKE twice straight to a swarm server and gets byte-identical replies. - `server_handshake_dedup::duplicate_through_two_relays_keeps_the_staged_holepunch`: the same PEER_HANDSHAKE reaches a firewalled swarm server through two of its relays. Both copies get the same reply, and a PEER_HOLEPUNCH for the first reply's id is still answered, so the second copy did not replace the staged holepunch session. +- `server_handshake_dedup::rejected_relayed_handshake_is_not_cached_or_staged`: a firewalled server with `max_peers = 0` answers each relayed copy with a fresh reply (nothing cached) and keeps no staged holepunch session for a rejected handshake. +- `server_handshake_dedup::rust_client_connects_directly_to_firewalled_server`: a Rust client sends its handshake straight to a firewalled swarm server; the server dials it and data flows both ways (guards Rust to Rust direct connections against the server-side dial). ### Live Network Validation diff --git a/peeroxide/CHANGELOG.md b/peeroxide/CHANGELOG.md index 3c3d646..f26a0b0 100644 --- a/peeroxide/CHANGELOG.md +++ b/peeroxide/CHANGELOG.md @@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Node.js Hyperswarm clients can now connect to a firewalled peeroxide server. Before, every attempt timed out (`passive holepunch timed out` here, `HOLEPUNCH_ABORTED` on the Node side). The server side of the handshake now matches Node hyperdht `lib/server.js` on three points: - - **Duplicate PEER_HANDSHAKE.** A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Each copy now gets the same reply, cached by a hash of the Noise message for 20 s. Before, each copy got a fresh Noise responder, and the second replaced the first's holepunch session. The cache holds at most 1024 replies (oldest evicted first), expires them in insertion order and on a periodic sweep, and skips handshakes the server rejects, so a flood of distinct handshakes cannot grow it or stall the swarm. + - **Duplicate PEER_HANDSHAKE.** A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Each copy now gets the same reply, cached by a hash of the Noise message for 20 s. Before, each copy got a fresh Noise responder, and the second replaced the first's holepunch session. The cache holds at most 1024 replies (oldest evicted first), expires them in insertion order and on a periodic sweep, and skips handshakes the server rejects (already connected, at `max_peers`, no UDX info), releasing any passive holepunch staged for them, so a flood of distinct handshakes cannot grow it or stall the swarm. - **Holepunch relays.** The reply's `holepunch.relays[].peer_address` is now our own address as each relay saw it (`HyperDhtHandle::current_relays()`). Before, it held the client's address, so the relay sent the client's PEER_HOLEPUNCH back to the client. - **Open or direct clients.** A client that says it is `FIREWALL_OPEN`, or that reached us without a relay, is now dialled by the server: its stream connects to the client's address, so the server sends the first packet and opens its own NAT mapping, as Node's `hs.onsocket(...)` does. A firewalled server still advertises `holepunch { id, relays }` to such a client, as Node does, so an open Node client waits for that dial instead of dialling the firewalled server itself. Before, the open client was given a passive holepunch it never performs. diff --git a/peeroxide/src/swarm.rs b/peeroxide/src/swarm.rs index 5be8f4d..a720035 100644 --- a/peeroxide/src/swarm.rs +++ b/peeroxide/src/swarm.rs @@ -1227,7 +1227,7 @@ impl SwarmActor { Err(e) => { tracing::debug!(err = %e, "server handshake: noise send failed"); if let Some(setup) = holepunch_setup { - setup.abort_task.abort(); + setup.release(); } let _ = reply_tx.send(None); return; @@ -1239,7 +1239,7 @@ impl SwarmActor { Err(e) => { tracing::debug!(err = %e, "server handshake: noise finalize failed"); if let Some(setup) = holepunch_setup { - setup.abort_task.abort(); + setup.release(); } let _ = reply_tx.send(None); return; @@ -1248,14 +1248,21 @@ impl SwarmActor { let remote_pk = nw_result.remote_public_key; - // Remember the reply only for a handshake we act on: a staged - // holepunch, or a connection the checks below will let through. - // Rejected handshakes must not take up cache space. - let accepted = holepunch_setup.is_some() - || (!self.connections.has(&remote_pk) - && self.connections.len() < self.config.max_peers - && remote_payload.udx.is_some()); - if accepted { + // A handshake we would reject (already connected, at max_peers, no + // UDX info) neither takes up cache space nor keeps a staged + // holepunch: that session would only be dropped when its punch landed. + let eligible = !self.connections.has(&remote_pk) + && self.connections.len() < self.config.max_peers + && remote_payload.udx.is_some(); + let holepunch_setup = match holepunch_setup { + Some(setup) if !eligible => { + tracing::debug!(pk = %short_hex(&remote_pk), "server: rejecting handshake, releasing staged holepunch"); + setup.release(); + None + } + setup => setup, + }; + if eligible { self.handshake_replies.insert(&msg.noise, noise_reply.clone()); } let _ = reply_tx.send(encode_server_handshake_reply( @@ -1749,6 +1756,16 @@ struct HolepunchSetup { abort_task: JoinHandle<()>, } +impl HolepunchSetup { + /// Drop a staged holepunch that will not be committed: stop its abort + /// timer and close its puncher, as the timeout path does. + fn release(self) { + self.abort_task.abort(); + let mut puncher = self.puncher; + puncher.destroy(); + } +} + /// Open the server side of a direct (non-relayed, non-holepunched) UDX stream. /// /// With `dial` set, the stream connects to `client_address` straight away, so diff --git a/peeroxide/tests/server_handshake_dedup.rs b/peeroxide/tests/server_handshake_dedup.rs index 60b34e8..e4b30c1 100644 --- a/peeroxide/tests/server_handshake_dedup.rs +++ b/peeroxide/tests/server_handshake_dedup.rs @@ -13,11 +13,18 @@ //! first copy stages a passive holepunch. The second copy must not replace //! that session: a PEER_HOLEPUNCH for the first reply's id must still be //! answered. +//! - `rejected_relayed_handshake_is_not_cached_or_staged`: a server with no +//! free connection slot neither caches its replies nor keeps the passive +//! holepunches it staged for them. +//! - `rust_client_connects_directly_to_firewalled_server`: a Rust client sends +//! its handshake straight to a firewalled swarm server (no relay). The +//! server now dials such a client, as Node does; this guards that Rust to +//! Rust direct connections still work with that dial (data flows both ways). use std::time::Duration; use libudx::UdxRuntime; -use peeroxide::{spawn as spawn_swarm, JoinOpts, SwarmConfig, SwarmHandle}; +use peeroxide::{spawn as spawn_swarm, JoinOpts, SwarmConfig, SwarmConnection, SwarmHandle}; use peeroxide_dht::crypto::hash; use peeroxide_dht::hyperdht::{spawn, HyperDhtConfig, KeyPair}; use peeroxide_dht::hyperdht_messages::{ @@ -171,88 +178,180 @@ async fn relays_of(server: &SwarmHandle) -> Option> { (relays.len() >= 2).then_some(relays) } -#[tokio::test] -async fn duplicate_through_two_relays_keeps_the_staged_holepunch() { - let _ = tracing_subscriber::fmt::try_init(); - - let result = tokio::time::timeout(Duration::from_secs(60), run_relayed()).await; - match result { - Ok(Ok(())) => {} - Ok(Err(e)) => panic!("relayed duplicate handshake test failed: {e}"), - Err(_) => panic!("relayed duplicate handshake test timed out after 60s"), - } +/// A firewalled swarm server on a small local testnet, ready to be reached +/// through two of its announce relays. +struct RelayedTestnet { + _rt: UdxRuntime, + bs: peeroxide_dht::hyperdht::HyperDhtHandle, + nodes: Vec, + server: SwarmHandle, + server_conns: tokio::sync::mpsc::Receiver, + client: peeroxide_dht::hyperdht::HyperDhtHandle, + relays: Vec, + server_pk: [u8; 32], } -async fn run_relayed() -> Result<(), Box> { - let rt = UdxRuntime::new()?; +impl RelayedTestnet { + async fn start(max_peers: usize) -> Result> { + let rt = UdxRuntime::new()?; - // Reachable DHT nodes: they store the server's announce, so they relay - // handshakes to it, and they give the firewalled server the NAT samples - // it needs to classify itself. - let (_bs_join, bs, _bs_rx) = spawn(&rt, local_dht(vec![], false)).await?; - let bootstrap = vec![format!("127.0.0.1:{}", bs.local_port().await?)]; - let mut nodes = Vec::new(); - for _ in 0..3 { - let (join, node, rx) = spawn(&rt, local_dht(bootstrap.clone(), false)).await?; - node.bootstrapped().await?; - nodes.push((join, node, rx)); - } + // Reachable DHT nodes: they store the server's announce, so they + // relay handshakes to it, and they give the firewalled server the + // NAT samples it needs to classify itself. + let (_, bs, _) = spawn(&rt, local_dht(vec![], false)).await?; + let bootstrap = vec![format!("127.0.0.1:{}", bs.local_port().await?)]; + let mut nodes = Vec::new(); + for _ in 0..3 { + let (_, node, _) = spawn(&rt, local_dht(bootstrap.clone(), false)).await?; + node.bootstrapped().await?; + nodes.push(node); + } - let mut config = SwarmConfig::default(); - config.dht = local_dht(bootstrap.clone(), true); - let (_task, server, _conn_rx) = spawn_swarm(config).await?; - server - .join([0x5b; 32], { - let mut opts = JoinOpts::default(); - opts.client = false; - opts + let mut config = SwarmConfig::default(); + config.dht = local_dht(bootstrap.clone(), true); + config.max_peers = max_peers; + let (_, server, server_conns) = spawn_swarm(config).await?; + server + .join([0x5b; 32], { + let mut opts = JoinOpts::default(); + opts.client = false; + opts + }) + .await?; + server.flush().await?; + let server_pk = server.key_pair().public_key; + + wait_for("the server to classify itself as firewalled", async || { + let state = server.dht().firewalled().await.ok()?; + (state == FIREWALL_CONSISTENT || state == FIREWALL_RANDOM).then_some(()) }) .await?; - server.flush().await?; - let server_pk = server.key_pair().public_key; - let target = hash(&server_pk); + let relays = wait_for("two announce relays", async || relays_of(&server).await).await?; - wait_for("the server to classify itself as firewalled", async || { - let state = server.dht().firewalled().await.ok()?; - (state == FIREWALL_CONSISTENT || state == FIREWALL_RANDOM).then_some(()) - }) - .await?; - let relays = wait_for("two announce relays", async || relays_of(&server).await).await?; + let (_, client, _) = spawn(&rt, local_dht(bootstrap, true)).await?; + client.bootstrapped().await?; - let (_cli_join, client, _cli_rx) = spawn(&rt, local_dht(bootstrap, true)).await?; - client.bootstrapped().await?; + Ok(Self { + _rt: rt, + bs, + nodes, + server, + server_conns, + client, + relays, + server_pk, + }) + } - let client_kp = KeyPair::generate(); - let mut nw = NoiseWrap::new_initiator( - NoiseKeypair { - public_key: client_kp.public_key, - secret_key: client_kp.secret_key, - }, - server_pk, - ); - let request = Router::encode_client_handshake(nw.send(&client_noise_payload())?, None, None)?; + fn noise_initiator(&self) -> NoiseWrap { + let client_kp = KeyPair::generate(); + NoiseWrap::new_initiator( + NoiseKeypair { + public_key: client_kp.public_key, + secret_key: client_kp.secret_key, + }, + self.server_pk, + ) + } - // One copy through each relay, the second after the first has been - // answered and its passive holepunch staged. - let mut replies = Vec::new(); - for relay in &relays[..2] { + async fn request( + &self, + command: u64, + value: Vec, + relay: usize, + ) -> Result>, Box> { let mut params = UserRequestParams::default(); - params.command = PEER_HANDSHAKE; - params.target = Some(target); - params.value = Some(request.clone()); - let addr = &relay.relay_address; - let resp = client.dht().request(params, &addr.host, addr.port).await?; + params.command = command; + params.target = Some(hash(&self.server_pk)); + params.value = Some(value); + let addr: &Ipv4Peer = &self.relays[relay].relay_address; + let resp = self.client.dht().request(params, &addr.host, addr.port).await?; assert_eq!(resp.error, 0, "relay {addr:?} returned error code {}", resp.error); - replies.push(decode_handshake_from_bytes( - &resp.value.ok_or("handshake reply without a value")?, - )?); + Ok(resp.value) + } + + /// Send one handshake through each of two relays, the second after the + /// first has been answered (and, if accepted, its holepunch staged). + async fn handshake_twice( + &self, + request: &[u8], + ) -> Result<[Vec; 2], Box> { + let mut replies = Vec::new(); + for relay in 0..2 { + let value = self + .request(PEER_HANDSHAKE, request.to_vec(), relay) + .await? + .ok_or("handshake reply without a value")?; + replies.push(decode_handshake_from_bytes(&value)?.noise); + } + Ok([replies.remove(0), replies.remove(0)]) + } + + /// Send one PEER_HOLEPUNCH probe for session `id` and return the + /// decrypted reply, or `None` if the server did not answer. + async fn probe_holepunch( + &self, + id: u64, + secure: &SecurePayload, + ) -> Result, Box> { + let probe = HolepunchPayload { + error: ERROR_NONE, + firewall: FIREWALL_UNKNOWN, + round: 0, + connected: false, + punching: false, + addresses: Some(vec![]), + remote_address: None, + token: Some([7; 32]), + remote_token: None, + }; + let value = Router::encode_client_holepunch(id, secure.encrypt(&probe)?, None)?; + let reply = match self.request(PEER_HOLEPUNCH, value, 0).await { + Ok(Some(v)) => v, + Ok(None) | Err(_) => return Ok(None), + }; + let reply = decode_holepunch_msg_from_bytes(&reply)?; + assert_eq!(reply.id, id); + Ok(Some(secure.decrypt(&reply.payload)?)) + } + + async fn destroy(self) -> Result<(), Box> { + self.server.destroy().await?; + self.client.destroy().await?; + for node in self.nodes { + node.destroy().await?; + } + self.bs.destroy().await?; + Ok(()) } +} + +async fn with_timeout(name: &str, fut: impl std::future::Future>>) { + let _ = tracing_subscriber::fmt::try_init(); + match tokio::time::timeout(Duration::from_secs(60), fut).await { + Ok(Ok(())) => {} + Ok(Err(e)) => panic!("{name} failed: {e}"), + Err(_) => panic!("{name} timed out after 60s"), + } +} + +#[tokio::test] +async fn duplicate_through_two_relays_keeps_the_staged_holepunch() { + with_timeout("relayed duplicate handshake test", run_relayed()).await; +} + +async fn run_relayed() -> Result<(), Box> { + let net = RelayedTestnet::start(64).await?; + let mut nw = net.noise_initiator(); + let request = Router::encode_client_handshake(nw.send(&client_noise_payload())?, None, None)?; + + let replies = net.handshake_twice(&request).await?; assert_eq!( - replies[0].noise, replies[1].noise, + replies[0], replies[1], "both relayed copies of one handshake must get the same Noise reply" ); - let payload = nw.recv(&replies[0].noise)?; + let payload = nw.recv(&replies[0])?; assert_eq!(payload.error, 0); let holepunch = payload .holepunch @@ -262,40 +361,89 @@ async fn run_relayed() -> Result<(), Box> { // The session staged for the first copy must still exist. let secure = SecurePayload::new(noise.holepunch_secret); - let probe = HolepunchPayload { - error: ERROR_NONE, - firewall: FIREWALL_UNKNOWN, - round: 0, - connected: false, - punching: false, - addresses: Some(vec![]), - remote_address: None, - token: Some([7; 32]), - remote_token: None, - }; - let value = Router::encode_client_holepunch(holepunch.id, secure.encrypt(&probe)?, None)?; - let mut params = UserRequestParams::default(); - params.command = PEER_HOLEPUNCH; - params.target = Some(target); - params.value = Some(value); - let addr: &Ipv4Peer = &relays[0].relay_address; - let resp = client.dht().request(params, &addr.host, addr.port).await?; - assert_eq!(resp.error, 0, "holepunch returned error code {}", resp.error); - let reply = decode_holepunch_msg_from_bytes( - &resp - .value - .ok_or("no holepunch reply: the first handshake's session was replaced")?, - )?; - assert_eq!(reply.id, holepunch.id); - let reply_payload = secure.decrypt(&reply.payload)?; - assert_eq!(reply_payload.error, ERROR_NONE); - assert_eq!(reply_payload.remote_token, Some([7; 32])); + let reply = net + .probe_holepunch(holepunch.id, &secure) + .await? + .ok_or("no holepunch reply: the first handshake's session was replaced")?; + assert_eq!(reply.error, ERROR_NONE); + assert_eq!(reply.remote_token, Some([7; 32])); - server.destroy().await?; - client.destroy().await?; - for (_, node, _) in nodes { - node.destroy().await?; - } - bs.destroy().await?; - Ok(()) + net.destroy().await +} + +#[tokio::test] +async fn rejected_relayed_handshake_is_not_cached_or_staged() { + with_timeout("rejected relayed handshake test", run_rejected()).await; +} + +/// A server with no free connection slot (`max_peers = 0`) rejects every +/// handshake. Its replies must not be cached, and it must not keep the +/// passive holepunch it staged for them. +async fn run_rejected() -> Result<(), Box> { + let net = RelayedTestnet::start(0).await?; + let mut nw = net.noise_initiator(); + let request = Router::encode_client_handshake(nw.send(&client_noise_payload())?, None, None)?; + + let replies = net.handshake_twice(&request).await?; + assert_ne!( + replies[0], replies[1], + "a rejected handshake's reply must not be cached: each copy gets a fresh responder" + ); + + // A single, separate handshake: nothing replaces its session, so only + // the rejection itself can have released it. + let mut nw = net.noise_initiator(); + let request = Router::encode_client_handshake(nw.send(&client_noise_payload())?, None, None)?; + let reply = net + .request(PEER_HANDSHAKE, request, 0) + .await? + .ok_or("handshake reply without a value")?; + let payload = nw.recv(&decode_handshake_from_bytes(&reply)?.noise)?; + let holepunch = payload.holepunch.ok_or("the reply should still carry a holepunch")?; + let noise = nw.finalize()?; + + let secure = SecurePayload::new(noise.holepunch_secret); + assert!( + net.probe_holepunch(holepunch.id, &secure).await?.is_none(), + "a rejected handshake must not keep a staged holepunch session" + ); + + net.destroy().await +} + +#[tokio::test] +async fn rust_client_connects_directly_to_firewalled_server() { + with_timeout("direct Rust client test", run_direct_rust_client()).await; +} + +async fn run_direct_rust_client() -> Result<(), Box> { + let mut net = RelayedTestnet::start(64).await?; + let server_addr: std::net::SocketAddr = + format!("127.0.0.1:{}", net.server.dht().local_port().await?).parse()?; + + let mut client_conn = net + .client + .connect_to(&KeyPair::generate(), net.server_pk, server_addr, &net._rt) + .await?; + let mut server_conn = tokio::time::timeout(Duration::from_secs(10), net.server_conns.recv()) + .await + .map_err(|_| "timed out waiting for the server's connection")? + .ok_or("server connection channel closed")?; + assert!(!server_conn.is_initiator); + + client_conn.stream.write(b"ping").await?; + let got = tokio::time::timeout(Duration::from_secs(10), server_conn.peer.stream.read()) + .await + .map_err(|_| "timed out reading on the server")?? + .ok_or("server stream closed")?; + assert_eq!(got, b"ping"); + + server_conn.peer.stream.write(b"pong").await?; + let got = tokio::time::timeout(Duration::from_secs(10), client_conn.stream.read()) + .await + .map_err(|_| "timed out reading on the client")?? + .ok_or("client stream closed")?; + assert_eq!(got, b"pong"); + + net.destroy().await } diff --git a/tests/node/hyperswarm-server.js b/tests/node/hyperswarm-server.js index 4b5809b..c2ecd24 100644 --- a/tests/node/hyperswarm-server.js +++ b/tests/node/hyperswarm-server.js @@ -16,10 +16,11 @@ const readline = require('readline') // like a peer with a public address. It stays ephemeral, // so it does not store the server's announce and reaches // the server through relays. -// --direct-client the client's DHT node is reachable and persistent -// (ephemeral: false). On this small testnet it then stores -// the server's announce itself and sends the handshake -// straight to the server, which is the direct path. +// --direct-client the client's DHT node is persistent (ephemeral: false) +// but reports FIREWALL_UNKNOWN. On this small testnet it +// stores the server's announce itself and sends the +// handshake straight to the server, which is the direct +// path; through a relay it would have to holepunch. const flags = new Set(process.argv.slice(2)) async function main () { @@ -57,9 +58,13 @@ async function main () { if (flags.has('--remote-client')) { const connect = serverDht.connect.bind(serverDht) serverDht.connect = (publicKey, opts = {}) => connect(publicKey, { ...opts, localConnection: false }) + } + if (flags.has('--remote-client') || flags.has('--direct-client')) { // Report FIREWALL_UNKNOWN, as a client behind a NAT would. On loopback, // dht-rpc's NAT check can otherwise decide this node is reachable, and // the client then says FIREWALL_OPEN and is dialled instead of punching. + // For --direct-client this also means only a handshake that really went + // straight to the server can avoid a holepunch. serverDht.remoteAddress = () => null } From f814e36c8e38b293473c6dfda811e9070914dfc3 Mon Sep 17 00:00:00 2001 From: eshork <1829176+eshork@users.noreply.github.com> Date: Sun, 1 Nov 2026 21:21:03 -0500 Subject: [PATCH 5/8] test(interop): skip the LAN shortcut for --direct-client A relayed --direct-client attempt would otherwise take Node's same-host LAN shortcut, which leaves the punch counters at zero, so the counters could not tell a direct handshake from a relayed one. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- TESTING.md | 4 ++-- tests/node/hyperswarm-server.js | 18 +++++++++--------- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/TESTING.md b/TESTING.md index 8b18343..7bd2345 100644 --- a/TESTING.md +++ b/TESTING.md @@ -167,13 +167,13 @@ Passing result: Protomux framing and blind-relay control messages are byte-ident | Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 4 | | Server handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 4 | -**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut and reports `FIREWALL_UNKNOWN`, as a client behind a NAT on another network would; one more DHT node is started so it can classify its NAT) `--open-client` (the Node client's DHT node is reachable, like a peer with a public address, but ephemeral, so it reaches the server through relays) and `--direct-client` (persistent but reporting `FIREWALL_UNKNOWN`: on this small testnet it stores the server's announce and sends its handshake straight to the server; through a relay it would have to holepunch). A `stats` command returns the Node DHT's punch counters (`punches.open` / `consistent` / `random`), which the tests use to check which side dialled. +**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut and reports `FIREWALL_UNKNOWN`, as a client behind a NAT on another network would; one more DHT node is started so it can classify its NAT) `--open-client` (the Node client's DHT node is reachable, like a peer with a public address, but ephemeral, so it reaches the server through relays) and `--direct-client` (persistent, reporting `FIREWALL_UNKNOWN` and skipping the LAN shortcut: on this small testnet it stores the server's announce and sends its handshake straight to the server; through a relay it would have to holepunch). A `stats` command returns the Node DHT's punch counters (`punches.open` / `consistent` / `random`), which the tests use to check which side dialled. Passing result: - `hyperswarm_cross_language_connect`: Rust Hyperswarm client connects to a Node.js `hyperswarm` server over the DHT. Full flow: DHT bootstrap → topic announce/lookup → PEER_HANDSHAKE → skip-holepunching (same-host) → UDX stream → SecretStream encrypted channel → receive "hello from node". Verifies end-to-end interoperability of the complete Hyperswarm stack. - `hyperswarm_node_client_holepunches_to_rust_server`: a firewalled Node.js client connects to a firewalled Rust server. The test first waits for the server to classify itself as firewalled (`CONSISTENT`/`RANDOM`). The client sends one PEER_HANDSHAKE through two relays in parallel, then holepunches through the relays the server advertised (Node's punch counters confirm a holepunch and no direct dial); data flows both ways. - `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node.js client reaches the firewalled Rust server through relays, sees an advertised holepunch, and waits for the server to dial it, as a Node server would. Node's punch counters confirm the client neither dialled nor holepunched; on loopback a client dial would also succeed, so this is what shows the server dialled. -- `hyperswarm_direct_node_client_connects_to_rust_server`: a persistent Node.js client that reports `FIREWALL_UNKNOWN` sends its handshake straight to the Rust server (no relay); both sides dial and data flows both ways. Node's punch counters confirm no holepunch, which a relayed attempt would have needed. +- `hyperswarm_direct_node_client_connects_to_rust_server`: a persistent Node.js client that reports `FIREWALL_UNKNOWN` and skips the LAN shortcut sends its handshake straight to the Rust server (no relay); both sides dial and data flows both ways. Node's punch counters confirm no holepunch, which a relayed attempt would have needed. - `server_handshake_dedup::duplicate_peer_handshake_gets_the_same_reply`: a Rust client sends the same PEER_HANDSHAKE twice straight to a swarm server and gets byte-identical replies. - `server_handshake_dedup::duplicate_through_two_relays_keeps_the_staged_holepunch`: the same PEER_HANDSHAKE reaches a firewalled swarm server through two of its relays. Both copies get the same reply, and a PEER_HOLEPUNCH for the first reply's id is still answered, so the second copy did not replace the staged holepunch session. - `server_handshake_dedup::rejected_relayed_handshake_is_not_cached_or_staged`: a firewalled server with `max_peers = 0` answers each relayed copy with a fresh reply (nothing cached) and keeps no staged holepunch session for a rejected handshake. diff --git a/tests/node/hyperswarm-server.js b/tests/node/hyperswarm-server.js index c2ecd24..973d987 100644 --- a/tests/node/hyperswarm-server.js +++ b/tests/node/hyperswarm-server.js @@ -17,10 +17,11 @@ const readline = require('readline') // so it does not store the server's announce and reaches // the server through relays. // --direct-client the client's DHT node is persistent (ephemeral: false) -// but reports FIREWALL_UNKNOWN. On this small testnet it -// stores the server's announce itself and sends the -// handshake straight to the server, which is the direct -// path; through a relay it would have to holepunch. +// but reports FIREWALL_UNKNOWN and skips the LAN shortcut. +// On this small testnet it stores the server's announce +// itself and sends the handshake straight to the server, +// which is the direct path; through a relay it would have +// to holepunch. const flags = new Set(process.argv.slice(2)) async function main () { @@ -55,16 +56,15 @@ async function main () { }) await serverDht.fullyBootstrapped() - if (flags.has('--remote-client')) { + if (flags.has('--remote-client') || flags.has('--direct-client')) { const connect = serverDht.connect.bind(serverDht) serverDht.connect = (publicKey, opts = {}) => connect(publicKey, { ...opts, localConnection: false }) - } - if (flags.has('--remote-client') || flags.has('--direct-client')) { // Report FIREWALL_UNKNOWN, as a client behind a NAT would. On loopback, // dht-rpc's NAT check can otherwise decide this node is reachable, and // the client then says FIREWALL_OPEN and is dialled instead of punching. - // For --direct-client this also means only a handshake that really went - // straight to the server can avoid a holepunch. + // With the LAN shortcut off as well, a --direct-client handshake that + // went through a relay would have to holepunch; only one that really went + // straight to the server avoids it. serverDht.remoteAddress = () => null } From 97492bed89b51ffe120f84c74010155fd27763eb Mon Sep 17 00:00:00 2001 From: eshork <1829176+eshork@users.noreply.github.com> Date: Sun, 1 Nov 2026 21:35:05 -0500 Subject: [PATCH 6/8] refactor(swarm): select the server handshake path with a pure function `select_server_path` replaces the `server_firewalled` / `client_dialable` / `attempt_holepunch` flags with an explicit `ServerPath` (Relay, Dial, Holepunch, Listen), mirroring Node `lib/server.js _addHandshake`, and is unit-tested as a decision table. `Dial { advertise_holepunch }` carries the "firewalled server still offers a holepunch to a client it dials" rule. The free-slot and UDX checks no longer need the client's key, so they now run before a passive holepunch is staged: a full swarm no longer opens a puncher socket only to release it. Only the already-connected check still releases a staged session. Tests: the max_peers test now asserts no holepunch is staged; a new test covers the already-connected release path. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- TESTING.md | 7 +- peeroxide/CHANGELOG.md | 2 +- peeroxide/src/swarm.rs | 138 ++++++++++++++++------ peeroxide/tests/server_handshake_dedup.rs | 77 +++++++++--- 4 files changed, 168 insertions(+), 56 deletions(-) diff --git a/TESTING.md b/TESTING.md index 7bd2345..8f3cf86 100644 --- a/TESTING.md +++ b/TESTING.md @@ -163,9 +163,9 @@ Passing result: Protomux framing and blind-relay control messages are byte-ident | Suite | Command | Type | Count | |-------|---------|------|-------| -| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 28 | +| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 29 | | Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 4 | -| Server handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 4 | +| Server handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 5 | **Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut and reports `FIREWALL_UNKNOWN`, as a client behind a NAT on another network would; one more DHT node is started so it can classify its NAT) `--open-client` (the Node client's DHT node is reachable, like a peer with a public address, but ephemeral, so it reaches the server through relays) and `--direct-client` (persistent, reporting `FIREWALL_UNKNOWN` and skipping the LAN shortcut: on this small testnet it stores the server's announce and sends its handshake straight to the server; through a relay it would have to holepunch). A `stats` command returns the Node DHT's punch counters (`punches.open` / `consistent` / `random`), which the tests use to check which side dialled. @@ -176,7 +176,8 @@ Passing result: - `hyperswarm_direct_node_client_connects_to_rust_server`: a persistent Node.js client that reports `FIREWALL_UNKNOWN` and skips the LAN shortcut sends its handshake straight to the Rust server (no relay); both sides dial and data flows both ways. Node's punch counters confirm no holepunch, which a relayed attempt would have needed. - `server_handshake_dedup::duplicate_peer_handshake_gets_the_same_reply`: a Rust client sends the same PEER_HANDSHAKE twice straight to a swarm server and gets byte-identical replies. - `server_handshake_dedup::duplicate_through_two_relays_keeps_the_staged_holepunch`: the same PEER_HANDSHAKE reaches a firewalled swarm server through two of its relays. Both copies get the same reply, and a PEER_HOLEPUNCH for the first reply's id is still answered, so the second copy did not replace the staged holepunch session. -- `server_handshake_dedup::rejected_relayed_handshake_is_not_cached_or_staged`: a firewalled server with `max_peers = 0` answers each relayed copy with a fresh reply (nothing cached) and keeps no staged holepunch session for a rejected handshake. +- `server_handshake_dedup::handshake_at_max_peers_is_not_cached_or_staged`: a firewalled server with `max_peers = 0` answers each relayed copy with a fresh reply (nothing cached) and stages no holepunch. +- `server_handshake_dedup::handshake_from_connected_peer_releases_its_staged_holepunch`: a new relayed handshake from an already-connected peer is staged (the server only learns the peer's key from Noise), then released: its reply is not cached and a PEER_HOLEPUNCH for its id gets no answer. - `server_handshake_dedup::rust_client_connects_directly_to_firewalled_server`: a Rust client sends its handshake straight to a firewalled swarm server; the server dials it and data flows both ways (guards Rust to Rust direct connections against the server-side dial). ### Live Network Validation diff --git a/peeroxide/CHANGELOG.md b/peeroxide/CHANGELOG.md index f26a0b0..ec570c8 100644 --- a/peeroxide/CHANGELOG.md +++ b/peeroxide/CHANGELOG.md @@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Node.js Hyperswarm clients can now connect to a firewalled peeroxide server. Before, every attempt timed out (`passive holepunch timed out` here, `HOLEPUNCH_ABORTED` on the Node side). The server side of the handshake now matches Node hyperdht `lib/server.js` on three points: - - **Duplicate PEER_HANDSHAKE.** A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Each copy now gets the same reply, cached by a hash of the Noise message for 20 s. Before, each copy got a fresh Noise responder, and the second replaced the first's holepunch session. The cache holds at most 1024 replies (oldest evicted first), expires them in insertion order and on a periodic sweep, and skips handshakes the server rejects (already connected, at `max_peers`, no UDX info), releasing any passive holepunch staged for them, so a flood of distinct handshakes cannot grow it or stall the swarm. + - **Duplicate PEER_HANDSHAKE.** A Node client sends one Noise message through up to two relays in parallel and acts on the first reply. Each copy now gets the same reply, cached by a hash of the Noise message for 20 s. Before, each copy got a fresh Noise responder, and the second replaced the first's holepunch session. The cache holds at most 1024 replies (oldest evicted first), expires them in insertion order and on a periodic sweep, and skips handshakes the server rejects (already connected, at `max_peers`, no UDX info), staging no passive holepunch for them when the swarm is full (and releasing one staged for an already-connected peer), so a flood of distinct handshakes cannot grow it or stall the swarm. - **Holepunch relays.** The reply's `holepunch.relays[].peer_address` is now our own address as each relay saw it (`HyperDhtHandle::current_relays()`). Before, it held the client's address, so the relay sent the client's PEER_HOLEPUNCH back to the client. - **Open or direct clients.** A client that says it is `FIREWALL_OPEN`, or that reached us without a relay, is now dialled by the server: its stream connects to the client's address, so the server sends the first packet and opens its own NAT mapping, as Node's `hs.onsocket(...)` does. A firewalled server still advertises `holepunch { id, relays }` to such a client, as Node does, so an open Node client waits for that dial instead of dialling the firewalled server itself. Before, the open client was given a passive holepunch it never performs. diff --git a/peeroxide/src/swarm.rs b/peeroxide/src/swarm.rs index a720035..7a6a331 100644 --- a/peeroxide/src/swarm.rs +++ b/peeroxide/src/swarm.rs @@ -1118,24 +1118,14 @@ impl SwarmActor { (None, None) }; - // Decide how this handshake will connect: - // - `server_firewalled`: the DHT has settled on a firewalled - // classification (CONSISTENT or RANDOM — not OPEN, not UNKNOWN; - // UNKNOWN means NAT analysis hasn't gathered enough samples, and - // the direct path is the safer choice, as in Node hyperdht), and - // we are not forcing all server connections through a relay - // (relay_through bypasses NAT traversal entirely). - // - `client_dialable`: the client says it is FIREWALL_OPEN, or it - // reached us directly rather than through a relay. Node `lib/server.js - // _addHandshake` dials such a client itself (`hs.onsocket(...)`) - // right after replying, so we send the first packet and open our - // own NAT mapping towards it. - // - Otherwise, if we are firewalled, stage a passive holepunch. + // Decide how this handshake will connect (see `select_server_path`). let firewall_state = self.dht.firewalled().await.unwrap_or(FIREWALL_UNKNOWN); - let server_firewalled = matches!(firewall_state, FIREWALL_CONSISTENT | FIREWALL_RANDOM) - && relay_through_info.is_none(); - let client_dialable = remote_payload.firewall == FIREWALL_OPEN || !is_forwarded; - let attempt_holepunch = server_firewalled && !client_dialable; + let path = select_server_path( + relay_through_info.is_some(), + firewall_state, + remote_payload.firewall, + is_forwarded, + ); // Try to stage the passive Holepuncher up front so the reply we // send in this handshake can advertise the puncher's bound port @@ -1143,7 +1133,12 @@ impl SwarmActor { // reply with `holepunch: None` (the existing direct path); a // genuinely firewalled receiver will then time out and retry via // discovery rather than getting stuck on a broken half-state. - let holepunch_setup = if attempt_holepunch { + // A handshake we will reject anyway (no free slot, no UDX info) is + // not worth a puncher socket; the already-connected check needs the + // client's key, which Noise only yields after the reply is sent. + let can_accept = self.connections.len() < self.config.max_peers + && remote_payload.udx.is_some(); + let holepunch_setup = if path == ServerPath::Holepunch && can_accept { match self .try_setup_passive_holepunch( remote_payload.firewall, @@ -1176,19 +1171,15 @@ impl SwarmActor { // A client sends its PEER_HOLEPUNCH to `relay_address`, and the relay // forwards it to `peer_address`: that must be our own address as the // relay saw it (Node `Announcer.relays`), not the client's. - // - // A firewalled Node server advertises `{ id, relays }` even to a - // client it is about to dial (`holepunch: ourRemoteAddr ? null : - // {...}`): an open Node client that sees it waits for our dial - // (`lib/connect.js holepunch`) instead of dialling our firewalled - // address. That id has no staged puncher; a PEER_HOLEPUNCH for it gets - // no reply, as Node's `_onpeerholepunch` aborts one without a puncher. - let holepunch_info = match &holepunch_setup { - Some(setup) => Some(HolepunchInfo { + let holepunch_info = match (&holepunch_setup, path) { + (Some(setup), _) => Some(HolepunchInfo { id: setup.id, relays: self.dht.current_relays(), }), - None if server_firewalled && client_dialable => { + // No staged puncher behind this id: a PEER_HOLEPUNCH for it gets + // no reply, as Node's `_onpeerholepunch` aborts one without a + // puncher. + (None, ServerPath::Dial { advertise_holepunch: true }) => { let id = self.next_holepunch_id; self.next_holepunch_id = self.next_holepunch_id.wrapping_add(1); Some(HolepunchInfo { @@ -1196,7 +1187,7 @@ impl SwarmActor { relays: self.dht.current_relays(), }) } - None => None, + _ => None, }; let reply_payload = NoisePayload { @@ -1251,9 +1242,8 @@ impl SwarmActor { // A handshake we would reject (already connected, at max_peers, no // UDX info) neither takes up cache space nor keeps a staged // holepunch: that session would only be dropped when its punch landed. - let eligible = !self.connections.has(&remote_pk) - && self.connections.len() < self.config.max_peers - && remote_payload.udx.is_some(); + // Only the already-connected check can still release one here. + let eligible = can_accept && !self.connections.has(&remote_pk); let holepunch_setup = match holepunch_setup { Some(setup) if !eligible => { tracing::debug!(pk = %short_hex(&remote_pk), "server: rejecting handshake, releasing staged holepunch"); @@ -1351,7 +1341,7 @@ impl SwarmActor { local_stream_id, &remote_udx, &client_addr_for_task, - client_dialable, + matches!(path, ServerPath::Dial { .. }), &nw_result, ) .await @@ -1743,6 +1733,54 @@ impl SwarmActor { } } +/// How the server side of one handshake connects to the client. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ServerPath { + /// `relay_through` is configured: pair with the client through that blind + /// relay. NAT traversal is bypassed entirely. + Relay, + /// Dial the client: connect our stream to its address, so we send the + /// first packet and open our own NAT mapping towards it, as Node's + /// `hs.onsocket(...)` does. With `advertise_holepunch`, the reply still + /// offers `holepunch { id, relays }`, as a firewalled Node server does + /// (`holepunch: ourRemoteAddr ? null : {...}`): an open Node client that + /// sees it waits for our dial instead of dialling our firewalled address. + Dial { advertise_holepunch: bool }, + /// Stage a passive holepunch that the client completes through our + /// announce relays. + Holepunch, + /// Wait for the client's first packet and adopt its source address. + Listen, +} + +/// Choose the server-side path for a handshake, per Node hyperdht +/// `lib/server.js _addHandshake`. +/// +/// - The client is *dialable* when it says it is `FIREWALL_OPEN`, or when its +/// handshake reached us directly rather than through a relay. +/// - We count as firewalled only once the DHT has settled on CONSISTENT or +/// RANDOM. UNKNOWN means NAT analysis hasn't gathered enough samples, and +/// the non-holepunch path is the safer choice there, as in Node. +fn select_server_path( + relay_through: bool, + server_firewall: u64, + client_firewall: u64, + forwarded: bool, +) -> ServerPath { + let server_firewalled = matches!(server_firewall, FIREWALL_CONSISTENT | FIREWALL_RANDOM); + if relay_through { + ServerPath::Relay + } else if client_firewall == FIREWALL_OPEN || !forwarded { + ServerPath::Dial { + advertise_holepunch: server_firewalled, + } + } else if server_firewalled { + ServerPath::Holepunch + } else { + ServerPath::Listen + } +} + /// Output of [`SwarmActor::try_setup_passive_holepunch`]. Holds everything /// the actor needs to (a) build the handshake reply and (b) install the /// in-flight entry once Noise finalisation has produced a `remote_pk` + @@ -2064,6 +2102,40 @@ mod tests { assert_eq!(cache.entries.capacity(), 0, "an empty cache releases its storage"); } + #[test] + fn server_path_follows_node_server_add_handshake() { + use ServerPath::*; + const OPEN: u64 = FIREWALL_OPEN; + const UNKNOWN: u64 = FIREWALL_UNKNOWN; + const CONSISTENT: u64 = FIREWALL_CONSISTENT; + const RANDOM: u64 = FIREWALL_RANDOM; + + // (relay_through, server, client, forwarded) -> path + let cases = [ + // relay_through wins over everything. + ((true, CONSISTENT, OPEN, false), Relay), + ((true, UNKNOWN, UNKNOWN, true), Relay), + // Open or direct clients are dialled; a firewalled server still + // advertises a holepunch to them. + ((false, CONSISTENT, OPEN, true), Dial { advertise_holepunch: true }), + ((false, RANDOM, UNKNOWN, false), Dial { advertise_holepunch: true }), + ((false, UNKNOWN, OPEN, true), Dial { advertise_holepunch: false }), + ((false, OPEN, UNKNOWN, false), Dial { advertise_holepunch: false }), + // A relayed, non-open client holepunches only to a firewalled server. + ((false, CONSISTENT, UNKNOWN, true), Holepunch), + ((false, RANDOM, CONSISTENT, true), Holepunch), + ((false, UNKNOWN, UNKNOWN, true), Listen), + ((false, OPEN, RANDOM, true), Listen), + ]; + for ((relay, server, client, forwarded), want) in cases { + assert_eq!( + select_server_path(relay, server, client, forwarded), + want, + "relay_through={relay} server={server} client={client} forwarded={forwarded}" + ); + } + } + #[test] fn server_handshake_reply_follows_the_inbound_mode() { use peeroxide_dht::hyperdht_messages::{decode_handshake_from_bytes, MODE_FROM_CLIENT}; diff --git a/peeroxide/tests/server_handshake_dedup.rs b/peeroxide/tests/server_handshake_dedup.rs index e4b30c1..6f269cc 100644 --- a/peeroxide/tests/server_handshake_dedup.rs +++ b/peeroxide/tests/server_handshake_dedup.rs @@ -13,9 +13,11 @@ //! first copy stages a passive holepunch. The second copy must not replace //! that session: a PEER_HOLEPUNCH for the first reply's id must still be //! answered. -//! - `rejected_relayed_handshake_is_not_cached_or_staged`: a server with no -//! free connection slot neither caches its replies nor keeps the passive -//! holepunches it staged for them. +//! - `handshake_at_max_peers_is_not_cached_or_staged`: a server with no free +//! connection slot neither caches its replies nor stages a holepunch. +//! - `handshake_from_connected_peer_releases_its_staged_holepunch`: a new +//! relayed handshake from an already-connected peer is staged (the server +//! only learns the peer's key from Noise), then released and not cached. //! - `rust_client_connects_directly_to_firewalled_server`: a Rust client sends //! its handshake straight to a firewalled swarm server (no relay). The //! server now dials such a client, as Node does; this guards that Rust to @@ -244,7 +246,10 @@ impl RelayedTestnet { } fn noise_initiator(&self) -> NoiseWrap { - let client_kp = KeyPair::generate(); + self.noise_initiator_as(&KeyPair::generate()) + } + + fn noise_initiator_as(&self, client_kp: &KeyPair) -> NoiseWrap { NoiseWrap::new_initiator( NoiseKeypair { public_key: client_kp.public_key, @@ -372,14 +377,14 @@ async fn run_relayed() -> Result<(), Box> { } #[tokio::test] -async fn rejected_relayed_handshake_is_not_cached_or_staged() { - with_timeout("rejected relayed handshake test", run_rejected()).await; +async fn handshake_at_max_peers_is_not_cached_or_staged() { + with_timeout("max_peers handshake test", run_at_max_peers()).await; } /// A server with no free connection slot (`max_peers = 0`) rejects every -/// handshake. Its replies must not be cached, and it must not keep the -/// passive holepunch it staged for them. -async fn run_rejected() -> Result<(), Box> { +/// handshake. It must not cache the replies, and must not stage a passive +/// holepunch for them at all. +async fn run_at_max_peers() -> Result<(), Box> { let net = RelayedTestnet::start(0).await?; let mut nw = net.noise_initiator(); let request = Router::encode_client_handshake(nw.send(&client_noise_payload())?, None, None)?; @@ -389,19 +394,53 @@ async fn run_rejected() -> Result<(), Box> { replies[0], replies[1], "a rejected handshake's reply must not be cached: each copy gets a fresh responder" ); + let payload = nw.recv(&replies[0])?; + assert!( + payload.holepunch.is_none(), + "a server with no free slot should not stage a holepunch" + ); - // A single, separate handshake: nothing replaces its session, so only - // the rejection itself can have released it. - let mut nw = net.noise_initiator(); + net.destroy().await +} + +#[tokio::test] +async fn handshake_from_connected_peer_releases_its_staged_holepunch() { + with_timeout("already-connected handshake test", run_already_connected()).await; +} + +/// A peer that is already connected sends a new relayed handshake. The server +/// can only tell after staging a holepunch (the peer's key comes out of the +/// Noise handshake), so it must release that session and not cache the reply. +async fn run_already_connected() -> Result<(), Box> { + let mut net = RelayedTestnet::start(64).await?; + let peer = KeyPair::generate(); + + // Connect the peer first, straight to the server. + let server_addr: std::net::SocketAddr = + format!("127.0.0.1:{}", net.server.dht().local_port().await?).parse()?; + let _conn = net + .client + .connect_to(&peer, net.server_pk, server_addr, &net._rt) + .await?; + let _server_conn = tokio::time::timeout(Duration::from_secs(10), net.server_conns.recv()) + .await + .map_err(|_| "timed out waiting for the first connection")? + .ok_or("server connection channel closed")?; + + let mut nw = net.noise_initiator_as(&peer); let request = Router::encode_client_handshake(nw.send(&client_noise_payload())?, None, None)?; - let reply = net - .request(PEER_HANDSHAKE, request, 0) - .await? - .ok_or("handshake reply without a value")?; - let payload = nw.recv(&decode_handshake_from_bytes(&reply)?.noise)?; - let holepunch = payload.holepunch.ok_or("the reply should still carry a holepunch")?; - let noise = nw.finalize()?; + let replies = net.handshake_twice(&request).await?; + assert_ne!( + replies[0], replies[1], + "a rejected handshake's reply must not be cached: each copy gets a fresh responder" + ); + // Each copy staged, then released, its own session; nothing replaced it. + let payload = nw.recv(&replies[1])?; + let holepunch = payload + .holepunch + .ok_or("the server only learns the peer is connected after staging")?; + let noise = nw.finalize()?; let secure = SecurePayload::new(noise.holepunch_secret); assert!( net.probe_holepunch(holepunch.id, &secure).await?.is_none(), From db18fc7f964eb98070f2a303fecc1a6c96bebe15 Mon Sep 17 00:00:00 2001 From: eshork <1829176+eshork@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:47:25 -0500 Subject: [PATCH 7/8] chore: fix clippy drain_collect lints on current stable clippy 1.98 denies `drain(..).collect()` into the same collection type (clippy::drain_collect). Use `std::mem::take` instead; behaviour is unchanged. Three sites: libudx stream timeout path, protomux cork batch and pending remote-channel messages. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- libudx/src/native/stream.rs | 2 +- peeroxide-dht/src/protomux.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/libudx/src/native/stream.rs b/libudx/src/native/stream.rs index 24b41d2..ce18e53 100644 --- a/libudx/src/native/stream.rs +++ b/libudx/src/native/stream.rs @@ -1585,7 +1585,7 @@ async fn process_incoming( // Fail all pending writes with timeout error let pending: Vec<_> = { let mut guard = inner.lock().unwrap_or_else(|e| e.into_inner()); - guard.pending_writes.drain(..).collect() + std::mem::take(&mut guard.pending_writes) }; for mut pw in pending { if let Some(tx) = pw.tx.take() { diff --git a/peeroxide-dht/src/protomux.rs b/peeroxide-dht/src/protomux.rs index 1127cbd..5e8ebc2 100644 --- a/peeroxide-dht/src/protomux.rs +++ b/peeroxide-dht/src/protomux.rs @@ -780,7 +780,7 @@ fn handle_command(cmd: MuxCommand, state: &mut MuxState, write_queue: &mut VecDe if state.cork_depth > 0 { state.cork_depth -= 1; if state.cork_depth == 0 && !state.batch.is_empty() { - let entries: Vec<(u64, Vec)> = state.batch.drain(..).collect(); + let entries: Vec<(u64, Vec)> = std::mem::take(&mut state.batch); let frame = encode_batch(&entries); if !frame.is_empty() { write_queue.push_back(frame); @@ -860,7 +860,7 @@ fn create_channel_internal( // Drain any buffered messages if let Some(Some(rc)) = state.remote.get_mut(remote_id as usize - 1) { - let pending: Vec<_> = rc.pending.drain(..).collect(); + let pending: Vec<_> = std::mem::take(&mut rc.pending); for (msg_type, data) in pending { let _ = event_tx.send(ChannelEvent::Message { message_type: msg_type as u32, From 12d1c4ea6a98e9c6cef51deda974166f27f4edcf Mon Sep 17 00:00:00 2001 From: eshork <1829176+eshork@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:57:52 -0500 Subject: [PATCH 8/8] chore(cli): silence fetch_update deprecation on current stable Rust 1.99 deprecates `AtomicUsize::fetch_update` in favour of `try_update`, which does not exist at our MSRV (1.85), so CI clippy (-D warnings) fails on stable. Move the three identical saturating decrements into one `saturating_dec` helper with a scoped `#[allow(deprecated)]`; behaviour is unchanged. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- peeroxide-cli/src/cmd/chat/tui/status.rs | 29 ++++++++++++------------ 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/peeroxide-cli/src/cmd/chat/tui/status.rs b/peeroxide-cli/src/cmd/chat/tui/status.rs index e8861f0..ccdd633 100644 --- a/peeroxide-cli/src/cmd/chat/tui/status.rs +++ b/peeroxide-cli/src/cmd/chat/tui/status.rs @@ -7,6 +7,17 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +/// Decrement `counter`, saturating at zero. +/// +/// `fetch_update` is deprecated in favour of `try_update` (Rust 1.99), which +/// does not exist at our MSRV (1.85). +#[allow(deprecated)] +fn saturating_dec(counter: &AtomicUsize) { + let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| { + Some(v.saturating_sub(1)) + }); +} + use arc_swap::ArcSwap; use tokio::sync::Notify; @@ -97,11 +108,7 @@ impl StatusState { /// Decrement `send_pending` (saturating) and notify. pub fn dec_send_pending(&self) { // saturating: don't wrap if mismatched inc/dec ever sneak in. - let _ = self.send_pending.fetch_update( - Ordering::Relaxed, - Ordering::Relaxed, - |v| Some(v.saturating_sub(1)), - ); + saturating_dec(&self.send_pending); self.dirty.notify_one(); } @@ -124,11 +131,7 @@ impl StatusState { /// Decrement the in-flight counter (saturating at zero). pub fn dec_recv_in_flight(&self) { - let _ = self - .recv_pending - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| { - Some(v.saturating_sub(1)) - }); + saturating_dec(&self.recv_pending); self.dirty.notify_one(); } @@ -141,11 +144,7 @@ impl StatusState { /// Decrement the any-DHT-op counter (saturating at zero). pub fn dec_dht_active(&self) { - let _ = self - .dht_active - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| { - Some(v.saturating_sub(1)) - }); + saturating_dec(&self.dht_active); self.dirty.notify_one(); }