OverviewFilesCommitsIssuesDocsReleasesPeers

hyperswarm-peer commits / ed8df3a

hyperdht: dedup inbound handshakes by their Noise message, as hyperdht does
A client's handshake can arrive more than once — retried, or relayed through a second node. The server answered each copy with a FRESH Noise reply, so the client and server could derive different streams: the duplicate replaced the live connection and the session read nothing (keetd M3d: header_recv=0 on the public DHT). Like hyperdht's server.js _onpeerhandshake (_connects keyed by the noise bytes), the responder now keys an active handshake by BLAKE2b(noise) and answers a repeat with the reply it already sent. test_handshake_dedup: red 3/9 before, 9/9 after (plain and ASan). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
ed8df3a85452bdf8cd51cd5d69b746e13c7dbbdf
by plankatron <fleeky@gmail.com> · 2026-10-01T00:28:25.000Z

Changed files (4) +290 -46

src/hyperdht.cpp +70 -46
……@@ -48,6 +48,7 @@ HyperDht::HyperDht()
4848HyperDht::~HyperDht() {
4949 for (int i = 0; i < _handshake_count; i++) {
5050 delete _handshakes[i].noise;
51 delete _handshakes[i].reply_noise;
5152 }
5253 sodium_memzero(&_keypair, sizeof(_keypair));
5354}
……@@ -674,9 +675,74 @@ void HyperDht::_handle_handshake(const DhtRequest &req) {
674675 if (!dht_msg::handshake_decode(s, hmsg)) return;
675676 if (!hmsg.noise || hmsg.noise_len == 0) return;
676677
678 // Send a Noise message 2 for THIS request, wrapped as its mode needs (see below).
679 auto send_reply = [&](const std::vector<uint8_t> &nr) {
680 HandshakeMessage reply_hmsg;
681 memset(&reply_hmsg, 0, sizeof(reply_hmsg));
682 reply_hmsg.noise = nr.data();
683 reply_hmsg.noise_len = nr.size();
684
685 if (hmsg.mode == HS_MODE_FROM_RELAY && hmsg.has_peer_address) {
686 // Relay protocol: send FROM_SERVER REQUEST back to the relay node.
687 reply_hmsg.mode = HS_MODE_FROM_SERVER;
688 reply_hmsg.peer_address = hmsg.peer_address; // Original client address.
689 reply_hmsg.has_peer_address = true;
690
691 DHT_LOG("[HyperDht] _handle_handshake: relay mode, sending FROM_SERVER request back to relay %d.%d.%d.%d:%d (peerAddr=%d.%d.%d.%d:%d)\n",
692 req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port,
693 hmsg.peer_address.ip[0], hmsg.peer_address.ip[1], hmsg.peer_address.ip[2], hmsg.peer_address.ip[3], hmsg.peer_address.port);
694
695 EncodeState rhs = { 0, 0, nullptr };
696 dht_msg::handshake_preencode(rhs, reply_hmsg);
697 std::vector<uint8_t> reply_hs_buf(rhs.end);
698 rhs.buffer = reply_hs_buf.data();
699 rhs.start = 0;
700 dht_msg::handshake_encode(rhs, reply_hmsg);
701
702 // Send as REQUEST (not response!) so the relay can process FROM_SERVER.
703 // Use the same TID so the relay chain preserves the transaction.
704 _rpc->_send_request(req.tid, req.from, nullptr,
705 false, HYPERDHT_CMD_PEER_HANDSHAKE,
706 req.has_target ? req.target : nullptr,
707 reply_hs_buf.data(), reply_hs_buf.size());
708 } else {
709 // Direct connection (FROM_CLIENT): reply with REPLY as a response.
710 reply_hmsg.mode = HS_MODE_REPLY;
711
712 DHT_LOG("[HyperDht] _handle_handshake: direct mode, sending REPLY response to %d.%d.%d.%d:%d\n",
713 req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port);
714
715 EncodeState rhs = { 0, 0, nullptr };
716 dht_msg::handshake_preencode(rhs, reply_hmsg);
717 std::vector<uint8_t> reply_hs_buf(rhs.end);
718 rhs.buffer = reply_hs_buf.data();
719 rhs.start = 0;
720 dht_msg::handshake_encode(rhs, reply_hmsg);
721
722 _rpc->_send_response(req.tid, req.from, true, nullptr,
723 nullptr, 0, 0,
724 reply_hs_buf.data(), reply_hs_buf.size());
725 }
726 };
727
728 // A repeat of a handshake we are still holding: the same reply, nothing new
729 // (hyperdht server.js _onpeerhandshake keys by hex(noise) and reuses the reply).
730 uint8_t noise_key[32];
731 crypto_generichash(noise_key, 32, hmsg.noise, hmsg.noise_len, nullptr, 0);
732 for (int i = 0; i < _handshake_count; i++) {
733 DhtHandshake &held = _handshakes[i];
734 if (!held.active || held.is_initiator || !held.has_noise_key || !held.reply_noise) continue;
735 if (memcmp(held.noise_key, noise_key, 32) != 0) continue;
736 DHT_LOG("[HyperDht] _handle_handshake: repeat of handshake id=%u — resending its reply\n", held.id);
737 send_reply(*held.reply_noise);
738 return;
739 }
740
677741 // Create responder handshake.
678742 DhtHandshake *hs = _alloc_handshake(false, nullptr, req.from);
679743 if (!hs) return;
744 memcpy(hs->noise_key, noise_key, 32);
745 hs->has_noise_key = true;
680746
681747 // Save client address — needed so the responder can finalize the
682748 // UDX connection back to the client (LAN check or holepunch target).
……@@ -757,52 +823,8 @@ void HyperDht::_handle_handshake(const DhtRequest &req) {
757823 // REPLY response to the original client.
758824 // When we receive FROM_CLIENT directly, we reply with REPLY as a response.
759825
760 HandshakeMessage reply_hmsg;
761 memset(&reply_hmsg, 0, sizeof(reply_hmsg));
762 reply_hmsg.noise = noise_reply.data();
763 reply_hmsg.noise_len = noise_reply.size();
764
765 if (hmsg.mode == HS_MODE_FROM_RELAY && hmsg.has_peer_address) {
766 // Relay protocol: send FROM_SERVER REQUEST back to the relay node.
767 reply_hmsg.mode = HS_MODE_FROM_SERVER;
768 reply_hmsg.peer_address = hmsg.peer_address; // Original client address.
769 reply_hmsg.has_peer_address = true;
770
771 DHT_LOG("[HyperDht] _handle_handshake: relay mode, sending FROM_SERVER request back to relay %d.%d.%d.%d:%d (peerAddr=%d.%d.%d.%d:%d)\n",
772 req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port,
773 hmsg.peer_address.ip[0], hmsg.peer_address.ip[1], hmsg.peer_address.ip[2], hmsg.peer_address.ip[3], hmsg.peer_address.port);
774
775 EncodeState rhs = { 0, 0, nullptr };
776 dht_msg::handshake_preencode(rhs, reply_hmsg);
777 std::vector<uint8_t> reply_hs_buf(rhs.end);
778 rhs.buffer = reply_hs_buf.data();
779 rhs.start = 0;
780 dht_msg::handshake_encode(rhs, reply_hmsg);
781
782 // Send as REQUEST (not response!) so the relay can process FROM_SERVER.
783 // Use the same TID so the relay chain preserves the transaction.
784 _rpc->_send_request(req.tid, req.from, nullptr,
785 false, HYPERDHT_CMD_PEER_HANDSHAKE,
786 req.has_target ? req.target : nullptr,
787 reply_hs_buf.data(), reply_hs_buf.size());
788 } else {
789 // Direct connection (FROM_CLIENT): reply with REPLY as a response.
790 reply_hmsg.mode = HS_MODE_REPLY;
791
792 DHT_LOG("[HyperDht] _handle_handshake: direct mode, sending REPLY response to %d.%d.%d.%d:%d\n",
793 req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port);
794
795 EncodeState rhs = { 0, 0, nullptr };
796 dht_msg::handshake_preencode(rhs, reply_hmsg);
797 std::vector<uint8_t> reply_hs_buf(rhs.end);
798 rhs.buffer = reply_hs_buf.data();
799 rhs.start = 0;
800 dht_msg::handshake_encode(rhs, reply_hmsg);
801
802 _rpc->_send_response(req.tid, req.from, true, nullptr,
803 nullptr, 0, 0,
804 reply_hs_buf.data(), reply_hs_buf.size());
805 }
826 send_reply(noise_reply);
827 hs->reply_noise = new std::vector<uint8_t>(noise_reply);
806828
807829 // Finalize.
808830 if (hs->noise->is_complete()) {
……@@ -1251,6 +1273,8 @@ DhtHandshake *HyperDht::_find_handshake(uint32_t id) {
12511273void HyperDht::_free_handshake(DhtHandshake *hs) {
12521274 delete hs->noise;
12531275 hs->noise = nullptr;
1276 delete hs->reply_noise; // before compaction copies the last slot over this one
1277 hs->reply_noise = nullptr;
12541278 hs->active = false;
12551279
12561280 int idx = static_cast<int>(hs - _handshakes);
src/hyperdht.h +12 -0
……@@ -114,6 +114,15 @@ struct DhtHandshake {
114114 dht_msg::NoisePayload remote_payload;
115115 bool has_remote_payload;
116116
117 // RESPONDER DEDUP (hyperdht server.js _onpeerhandshake: `_connects.get(hex(noise))`).
118 // A repeated copy of the same inbound handshake — a retried request, or the same one
119 // relayed by a second node — gets the cached reply of the first instead of a new
120 // handshake, stream id and connection. Keyed by BLAKE2b-256 of the Noise message 1;
121 // `reply_noise` is the Noise message 2 sent for it, OWNED (freed in _free_handshake).
122 uint8_t noise_key[32];
123 bool has_noise_key;
124 std::vector<uint8_t> *reply_noise;
125
117126 // ⛔ WHEN THIS SLOT WAS TAKEN. Without it nothing could ever reclaim a slot,
118127 // the 16-slot pool filled with handshakes that had already served their
119128 // purpose, and _handle_handshake answered every later peer with a bare
……@@ -198,6 +207,9 @@ public:
198207 // Process time-based events.
199208 void tick(uint64_t now_ms);
200209
210 // Handshakes currently held (initiator + responder). For gates.
211 int handshake_count() const { return _handshake_count; }
212
201213 // ---- Topic management (matches JS Hyperswarm join/leave) ----
202214
203215 // Join a topic with server and/or client mode.
test/build_tests.sh +9 -0
……@@ -286,6 +286,15 @@ g++ -o test_handshake_pool \
286286 $UDX_OBJS $LDFLAGS
287287echo " OK: test_handshake_pool"
288288
289# ---- test_handshake_dedup: a repeated inbound handshake gets the SAME reply ----
290echo "Building test_handshake_dedup..."
291g++ $CXXFLAGS -o $OBJ/test_handshake_dedup.o test_handshake_dedup.cpp
292g++ -o test_handshake_dedup \
293 $OBJ/test_handshake_dedup.o $OBJ/noise_ik.o $OBJ/noise_handshake.o \
294 $OBJ/noise_stream.o $OBJ/dht_rpc.o $OBJ/holepuncher.o $OBJ/hyperdht.o \
295 $UDX_OBJS $LDFLAGS
296echo " OK: test_handshake_dedup"
297
289298# ---- test_find_peer_exhausted ----
290299# A connect attempt must fail when find_peer EXHAUSTS, not 30s later. See
291300# docs/PARITY-hyperswarm.md §11.2.
test/test_handshake_dedup.cpp +199 -0
……@@ -0,0 +1,199 @@
1// Gate: a REPEATED inbound handshake gets the SAME reply, not a second connection.
2//
3// hyperdht's server keys every inbound PEER_HANDSHAKE by its Noise message bytes
4// (server.js _onpeerhandshake, k = hex(noise)) and answers a repeat — a retried request,
5// or the same handshake arriving by a second path — with the cached reply of the first
6// (`_connects.get(k)`). Ours allocated a new handshake per copy: a new Noise reply, a new
7// udx stream id, a second connection that then "replaced" the first, leaving the client
8// bound to a stream we had destroyed. keetd's M3d runs saw it on the public DHT (each JS
9// dial arriving twice; header_recv=0 on every stream).
10//
11// 1 the same handshake twice, direct, same tid: ONE handshake, the listener told ONCE,
12// and the two reply packets byte-identical
13// 2 the same handshake again, now relayed (FROM_RELAY) by another node: still one
14// handshake, still told once, and a reply goes to that relay
15// 3 a DIFFERENT handshake: a new one is made (dedup must not swallow everything)
16#include <cstdio>
17#include <cstring>
18#include <vector>
19
20#include <arpa/inet.h>
21#include <netinet/in.h>
22#include <poll.h>
23#include <sys/socket.h>
24#include <unistd.h>
25
26#include <sodium.h>
27
28#include "dht_messages.h"
29#include "dht_rpc.h"
30#include "hyperdht.h"
31#include "noise_ik.h"
32
33extern "C" {
34#include <udx.h>
35#include <uv.h>
36}
37
38static int passed = 0, failed = 0;
39static void check (bool ok, const char *what) { (ok ? passed : failed)++; printf(" %s %s\n", ok ? "PASS" : "FAIL", what); }
40
41// a plain UDP socket standing in for the client (or a relay): where replies land
42struct Catcher {
43 int fd;
44 uint16_t port;
45 Catcher () {
46 fd = socket(AF_INET, SOCK_DGRAM, 0);
47 sockaddr_in a{};
48 a.sin_family = AF_INET;
49 a.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
50 bind(fd, (sockaddr *)&a, sizeof a);
51 socklen_t l = sizeof a;
52 getsockname(fd, (sockaddr *)&a, &l);
53 port = ntohs(a.sin_port);
54 }
55 ~Catcher () { close(fd); }
56 // pump the loop until a packet arrives (or ~1s)
57 std::vector<uint8_t> next (uv_loop_t *loop) {
58 for (int i = 0; i < 100; i++) {
59 uv_run(loop, UV_RUN_NOWAIT);
60 pollfd p{ fd, POLLIN, 0 };
61 if (poll(&p, 1, 10) > 0) {
62 uint8_t b[2048];
63 const ssize_t n = recv(fd, b, sizeof b, 0);
64 if (n > 0) return std::vector<uint8_t>(b, b + n);
65 }
66 }
67 return {};
68 }
69};
70
71static std::vector<uint8_t> encode_hs (const HandshakeMessage &h) {
72 EncodeState s = { 0, 0, nullptr };
73 dht_msg::handshake_preencode(s, h);
74 std::vector<uint8_t> b(s.end);
75 s.buffer = b.data();
76 s.start = 0;
77 dht_msg::handshake_encode(s, h);
78 return b;
79}
80
81int main () {
82 if (sodium_init() < 0) return 2;
83 uv_loop_t loop;
84 uv_loop_init(&loop);
85 udx_t udx;
86 udx_init(&loop, &udx, nullptr);
87 udx_socket_t sock;
88 udx_socket_init(&udx, &sock, nullptr);
89 sockaddr_in addr;
90 uv_ip4_addr("127.0.0.1", 0, &addr);
91 udx_socket_bind(&sock, (sockaddr *)&addr, 0);
92
93 DhtRpc rpc;
94 rpc.init(&sock, &loop);
95 Ed25519KeyPair server_kp;
96 crypto_sign_keypair(server_kp.public_key, server_kp.secret_key);
97 HyperDht dht;
98 dht.init(&rpc, server_kp);
99 dht.listen();
100 dht.tick(1000);
101
102 int told = 0;
103 dht.on_handshake_request = [&] (uint32_t, const uint8_t *, const dht_msg::NoisePayload &) { told++; };
104
105 uint8_t ns[32]; // NS_PEER_HANDSHAKE, as test_handshake_pool derives it
106 {
107 const char *name = "hyperswarm/dht";
108 uint8_t base[32];
109 crypto_generichash(base, 32, (const uint8_t *)name, strlen(name), nullptr, 0);
110 const uint8_t id = 0;
111 crypto_generichash_state st;
112 crypto_generichash_init(&st, nullptr, 0, 32);
113 crypto_generichash_update(&st, base, 32);
114 crypto_generichash_update(&st, &id, 1);
115 crypto_generichash_final(&st, ns, 32);
116 }
117
118 auto make_noise = [&] (uint32_t udx_id) {
119 Ed25519KeyPair kp;
120 crypto_sign_keypair(kp.public_key, kp.secret_key);
121 NoiseIK ik(true, kp, ns, server_kp.public_key);
122 dht_msg::NoisePayload p;
123 memset(&p, 0, sizeof p);
124 p.version = 1;
125 p.firewall = 1;
126 p.has_udx = true;
127 p.udx.id = udx_id;
128 p.has_secret_stream = true;
129 EncodeState ps = { 0, 0, nullptr };
130 dht_msg::noise_payload_preencode(ps, p);
131 std::vector<uint8_t> pb(ps.end);
132 ps.buffer = pb.data();
133 ps.start = 0;
134 dht_msg::noise_payload_encode(ps, p);
135 return ik.send(pb.data(), pb.size());
136 };
137
138 auto deliver = [&] (const std::vector<uint8_t> &value, uint16_t from_port, uint16_t tid) {
139 DhtRequest req;
140 memset(&req, 0, sizeof req);
141 req.command = HYPERDHT_CMD_PEER_HANDSHAKE;
142 req.tid = tid;
143 req.from.ip[0] = 127; req.from.ip[3] = 1;
144 req.from.port = from_port;
145 req.value = value.data();
146 req.value_len = value.size();
147 if (rpc.on_request) rpc.on_request(req);
148 };
149
150 printf("\n--- 1: the same handshake twice, direct ---\n");
151 Catcher client;
152 const auto noise = make_noise(4242);
153 HandshakeMessage h;
154 memset(&h, 0, sizeof h);
155 h.mode = HS_MODE_FROM_CLIENT;
156 h.noise = const_cast<uint8_t *>(noise.data());
157 h.noise_len = noise.size();
158 const auto direct = encode_hs(h);
159 deliver(direct, client.port, 77);
160 const auto r1 = client.next(&loop);
161 deliver(direct, client.port, 77);
162 const auto r2 = client.next(&loop);
163 check(!r1.empty() && !r2.empty(), "both copies were answered");
164 check(r1 == r2, "the two replies are byte-identical (the cached reply)");
165 check(dht.handshake_count() == 1, "one handshake held, not two");
166 check(told == 1, "the listener was told once");
167
168 printf("\n--- 2: the same handshake, relayed by another node ---\n");
169 Catcher relay;
170 HandshakeMessage hr;
171 memset(&hr, 0, sizeof hr);
172 hr.mode = HS_MODE_FROM_RELAY;
173 hr.noise = const_cast<uint8_t *>(noise.data());
174 hr.noise_len = noise.size();
175 hr.has_peer_address = true;
176 hr.peer_address.ip[0] = 127; hr.peer_address.ip[3] = 1;
177 hr.peer_address.port = client.port;
178 deliver(encode_hs(hr), relay.port, 78);
179 check(!relay.next(&loop).empty(), "the relay got a reply");
180 check(dht.handshake_count() == 1, "still one handshake");
181 check(told == 1, "the listener was still told only once");
182
183 printf("\n--- 3: a different handshake ---\n");
184 Catcher other;
185 const auto noise2 = make_noise(4343);
186 HandshakeMessage h2;
187 memset(&h2, 0, sizeof h2);
188 h2.mode = HS_MODE_FROM_CLIENT;
189 h2.noise = const_cast<uint8_t *>(noise2.data());
190 h2.noise_len = noise2.size();
191 deliver(encode_hs(h2), other.port, 79);
192 check(!other.next(&loop).empty(), "answered");
193 check(dht.handshake_count() == 2 && told == 2, "a new handshake, the listener told");
194
195 udx_socket_close(&sock);
196 uv_run(&loop, UV_RUN_NOWAIT);
197 printf("\n%s test_handshake_dedup: %d/%d\n", failed ? "FAIL" : "PASS", passed, passed + failed);
198 return failed ? 1 : 0;
199}