#include "hyperdht.h"
#include <cstdio>
#include <sodium.h>
#include "debug_log.h"
// ---- HyperDht implementation ----
HyperDht::HyperDht()
: _rpc(nullptr), _listening(false),
_handshake_count(0), _handshake_id_counter(0),
_query_info_count(0), _topic_count(0) {
if (sodium_init() < 0) {
DHT_LOG("HyperDht: sodium_init failed\n");
}
memset(&_keypair, 0, sizeof(_keypair));
memset(_handshakes, 0, sizeof(_handshakes));
// Zero QueryInfo POD members safely (QueryInfo contains std::vector).
for (auto &qi : _query_infos) {
qi.query_id = 0;
memset(qi.topic, 0, 32);
qi.is_announce = qi.is_lookup = qi.is_find_peer = false;
qi.is_immutable_get = qi.is_immutable_put = false;
qi.is_mutable_get = qi.is_mutable_put = false;
memset(qi.storage_key, 0, 32);
qi.mutable_seq = 0;
// put_value (std::vector) already default-constructed.
memset(qi.put_signature, 0, 64);
qi.active = false;
}
memset(_topics, 0, sizeof(_topics));
// Compute namespace hashes.
// crypto.namespace('hyperswarm/dht', [ANNOUNCE=4, UNANNOUNCE=5, MUTABLE_PUT=6, PEER_HANDSHAKE=0, PEER_HOLEPUNCH=1])
const uint8_t ids[] = { 4, 5, 6, 0, 1 };
uint8_t ns_buf[5 * 32];
NoiseIK::compute_namespace("hyperswarm/dht", ids, 5, ns_buf);
memcpy(_ns_announce, ns_buf + 0 * 32, 32);
memcpy(_ns_unannounce, ns_buf + 1 * 32, 32);
memcpy(_ns_mutable_put, ns_buf + 2 * 32, 32);
memcpy(_ns_peer_handshake, ns_buf + 3 * 32, 32);
memcpy(_ns_peer_holepunch, ns_buf + 4 * 32, 32);
}
HyperDht::~HyperDht() {
for (int i = 0; i < _handshake_count; i++) {
delete _handshakes[i].noise;
delete _handshakes[i].reply_noise;
}
sodium_memzero(&_keypair, sizeof(_keypair));
}
void HyperDht::init(DhtRpc *rpc, const Ed25519KeyPair &keypair) {
_rpc = rpc;
memcpy(&_keypair, &keypair, sizeof(Ed25519KeyPair));
// Wire up RPC callbacks.
_rpc->on_response = [this](const DhtResponse &res, uint32_t tag) {
_on_rpc_response(res, tag);
};
_rpc->on_request = [this](const DhtRequest &req) {
_on_rpc_request(req);
};
_rpc->on_query_complete = [this](uint32_t query_id) {
// Find matching query info and forward.
for (int i = 0; i < _query_info_count; i++) {
if (_query_infos[i].active && _query_infos[i].query_id == query_id) {
auto &qi = _query_infos[i];
// For announce queries, store the closest responded nodes.
// Holders are recorded per node in the commit callback now
// (_note_announce_holder), not guessed from the routing table
// after the fact.
// Handle storage query completions.
if (qi.is_immutable_put) {
if (on_immutable_put_complete) {
on_immutable_put_complete(qi.storage_key);
}
} else if (qi.is_immutable_get && qi.put_value.empty()) {
// Value not found during query — fire callback with null.
if (on_immutable_get_complete) {
on_immutable_get_complete(qi.storage_key, nullptr, 0);
}
} else if (qi.is_mutable_put) {
if (on_mutable_put_complete) {
on_mutable_put_complete(true);
}
} else if (qi.is_mutable_get) {
// Fire callback with best result found during query.
if (!qi.put_value.empty()) {
if (on_mutable_get_complete) {
on_mutable_get_complete(qi.topic, qi.put_value.data(),
qi.put_value.size(), qi.mutable_seq);
}
} else {
if (on_mutable_get_complete) {
on_mutable_get_complete(qi.topic, nullptr, 0, 0);
}
}
}
qi.active = false;
if (on_query_complete) on_query_complete(query_id);
break;
}
}
// Chain lookup after announce for topics with both server+client sessions.
// Matches JS Hyperswarm: await announce(), then for await (lookup()) {...}
for (int j = 0; j < _topic_count; j++) {
TopicDiscovery *td = &_topics[j];
if (td->active && td->pending_lookup && td->current_query_id == query_id) {
td->pending_lookup = false;
DHT_DISC("[HyperDht] CHAIN: announce qid=%u done, starting lookup for topic %02x%02x%02x%02x...\n",
query_id, td->topic[0], td->topic[1], td->topic[2], td->topic[3]);
td->current_query_id = lookup(td->topic);
DHT_DISC("[HyperDht] CHAIN: chained lookup qid=%u\n", td->current_query_id);
break;
}
}
};
}
// ---- Lookup ----
uint32_t HyperDht::lookup(const uint8_t *topic) {
// LOOKUP is a non-internal query (command = HYPERDHT_CMD_LOOKUP).
// target = topic hash.
DHT_DISC("[HyperDht] lookup: topic=%02x%02x%02x%02x...\n",
topic[0], topic[1], topic[2], topic[3]);
uint32_t qid = _rpc->query(topic, false, HYPERDHT_CMD_LOOKUP,
nullptr, 0, 0, false);
DHT_DISC("[HyperDht] lookup: qid=%u\n", qid);
// Track query.
if (_query_info_count < HYPERDHT_MAX_QUERIES) {
auto &qi = _query_infos[_query_info_count++];
qi.query_id = qid;
memcpy(qi.topic, topic, 32);
qi.is_announce = false;
qi.is_lookup = true;
qi.is_find_peer = false;
qi.is_immutable_get = false;
qi.is_immutable_put = false;
qi.is_mutable_get = false;
qi.is_mutable_put = false;
qi.put_value.clear();
qi.active = true;
}
return qid;
}
// ---- Find Peer ----
uint32_t HyperDht::find_peer(const uint8_t *public_key) {
// target = BLAKE2b(publicKey) — matches JS findPeer() which hashes the public key.
uint8_t target[32];
crypto_generichash(target, 32, public_key, 32, nullptr, 0);
// Use FIND_PEER command (not LOOKUP) — matches JS HyperDHT.findPeer().
// FIND_PEER checks the router's forwarding table on each DHT node,
// which returns the peer record directly from the node that stored the announce.
uint32_t qid = _rpc->query(target, false, HYPERDHT_CMD_FIND_PEER,
nullptr, 0, 0, false);
// Track query.
if (_query_info_count < HYPERDHT_MAX_QUERIES) {
auto &qi = _query_infos[_query_info_count++];
qi.query_id = qid;
memcpy(qi.topic, target, 32);
qi.is_announce = false;
qi.is_lookup = false;
qi.is_find_peer = true;
qi.is_immutable_get = false;
qi.is_immutable_put = false;
qi.is_mutable_get = false;
qi.is_mutable_put = false;
qi.put_value.clear();
qi.active = true;
}
return qid;
}
// ---- Announce ----
uint32_t HyperDht::announce(const uint8_t *topic) {
// Matches JS announce pattern:
// Traversal: LOOKUP with no value (finds K closest, collects tokens)
// Commit: ANNOUNCE with value + token (stores on closest nodes)
// The signature must be computed per-node because it includes the
// receiving node's ID and the per-node token in the signable hash.
// Capture topic for the lambda (topic pointer may not survive).
uint8_t topic_copy[32];
memcpy(topic_copy, topic, 32);
DHT_DISC("[HyperDht] announce: topic=%02x%02x%02x%02x... (LOOKUP traversal -> ANNOUNCE commit)\n",
topic[0], topic[1], topic[2], topic[3]);
uint32_t qid = _rpc->query_with_commit_fn(
topic, false,
HYPERDHT_CMD_LOOKUP, nullptr, 0, // Traversal: LOOKUP, no value
HYPERDHT_CMD_ANNOUNCE, // Commit: ANNOUNCE
[this, topic_copy](const uint8_t *target, const uint8_t *token,
const uint8_t *node_id,
const IPv4Address &from) -> std::vector<uint8_t> {
// ⛔ THIS is where we learn who holds our record. The reply's `from`
// is the node that took the announce; nothing else knows it. See
// _note_announce_holder and PARITY-hyperswarm.md §15.
_note_announce_holder(topic_copy, from, token);
return _encode_announce_value_for_node(topic_copy, token, node_id);
},
0);
if (_query_info_count < HYPERDHT_MAX_QUERIES) {
auto &qi = _query_infos[_query_info_count++];
qi.query_id = qid;
memcpy(qi.topic, topic, 32);
qi.is_announce = true;
qi.is_lookup = false;
qi.is_find_peer = false;
qi.is_immutable_get = false;
qi.is_immutable_put = false;
qi.is_mutable_get = false;
qi.is_mutable_put = false;
qi.put_value.clear();
qi.active = true;
}
return qid;
}
void HyperDht::unannounce(const uint8_t *topic) {
// ⛔ UNANNOUNCE IS A TRAVERSAL, NOT A REPLAY OF STORED NODES.
//
// This used to encode ONE value and send it to the addresses saved from the
// last announce. Two things were wrong with that, and the first hid the
// second:
//
// 1. the signable was the old ns || topic || peer form, so nothing we sent
// ever verified anywhere (see ann_signable);
// 2. the correct signable binds the RECEIVING NODE'S id and THAT node's
// token, so a single encoded value is structurally incapable of
// verifying at more than one node. Stored tokens also rotate, so by the
// time we leave, the saved ones are usually stale anyway.
//
// JS does a fresh LOOKUP and unannounces per reply
// (index.js:207 lookupAndUnannounce -> COMMANDS.LOOKUP). We reuse the same
// traversal+commit machinery announce already uses, which is where the
// per-node token and id come from.
//
// ⚠ KNOWN REMAINING DIVERGENCE (PARITY-hyperswarm.md A2): JS fires the
// unannounce from map() DURING traversal, for every reply whose peer list
// actually contains our key, whereas this commits at the closest nodes only.
// Ours therefore withdraws from fewer nodes than JS does. That is a separate
// defect with a separate fix and it is not addressed here.
uint8_t topic_copy[32];
memcpy(topic_copy, topic, 32);
_rpc->query_with_commit_fn(
topic, false,
HYPERDHT_CMD_LOOKUP, nullptr, 0, // traversal: LOOKUP, no value
HYPERDHT_CMD_UNANNOUNCE, // commit: UNANNOUNCE
[this, topic_copy](const uint8_t *target, const uint8_t *token,
const uint8_t *node_id,
const IPv4Address &) -> std::vector<uint8_t> {
return _encode_ann_value_for_node(_ns_unannounce, topic_copy, token, node_id);
},
0);
_announced_topics.erase(_topic_hex(topic));
}
// ---- Connect (handshake through relay) ----
// Matches JS: router.peerHandshake() → dht.request() (direct, NOT iterative query).
// The relay node routes: FROM_CLIENT → FROM_RELAY → FROM_SERVER → REPLY.
uint32_t HyperDht::connect(const uint8_t *remote_public_key,
const IPv4Address &relay) {
DhtHandshake *hs = _alloc_handshake(true, remote_public_key, relay);
if (!hs) return 0;
// Create Noise IK handshake as initiator.
hs->noise = new NoiseIK(true, _keypair, _ns_peer_handshake, remote_public_key);
// Build NoisePayload for Message 1.
// Matches JS connectThroughNode(): { error: NONE, firewall, addresses4, udx, secretStream }.
dht_msg::NoisePayload payload;
memset(&payload, 0, sizeof(payload));
payload.version = 1;
payload.error = 0; // ERROR.NONE
payload.firewall = _rpc->local_firewall();
// Include our addresses (local + NAT-detected external) so the server
// can attempt direct connect. The JS reference sends external addresses
// from DHT response `to` fields, not just getsockname().
payload.addresses4_count = _rpc->get_all_addresses(payload.addresses4, 8);
// UDX info — required by JS server to set up the encrypted stream.
payload.has_udx = true;
payload.udx.reusable_socket = false;
payload.udx.id = hs->id; // Stream ID for UDX.
payload.udx.seq = 0;
// SecretStream marker.
payload.has_secret_stream = true;
// Encode the noise payload.
EncodeState ps = { 0, 0, nullptr };
dht_msg::noise_payload_preencode(ps, payload);
std::vector<uint8_t> payload_buf(ps.end);
ps.buffer = payload_buf.data();
ps.start = 0;
dht_msg::noise_payload_encode(ps, payload);
// Noise IK Message 1.
auto noise_msg = hs->noise->send(payload_buf.data(), payload_buf.size());
// Wrap in HandshakeMessage with mode=FROM_CLIENT (0).
// Matches JS: c.encode(handshake, { mode: FROM_CLIENT, noise, peerAddress, relayAddress })
HandshakeMessage hmsg;
memset(&hmsg, 0, sizeof(hmsg));
hmsg.mode = HS_MODE_FROM_CLIENT; // Critical: must be 0, not 1.
hmsg.noise = noise_msg.data();
hmsg.noise_len = noise_msg.size();
// peerAddress and relayAddress are not set (null) for basic connect.
EncodeState hs_state = { 0, 0, nullptr };
dht_msg::handshake_preencode(hs_state, hmsg);
std::vector<uint8_t> hs_buf(hs_state.end);
hs_state.buffer = hs_buf.data();
hs_state.start = 0;
dht_msg::handshake_encode(hs_state, hmsg);
// Send as DIRECT request to the relay node (not iterative query).
// Matches JS: dht.request({ command: PEER_HANDSHAKE, target, value }, to)
// target = BLAKE2b(remote_public_key) — so the relay knows where to route.
uint8_t target[32];
crypto_generichash(target, 32, remote_public_key, 32, nullptr, 0);
// tag = hs->id so the response routes back to _on_rpc_response → _find_handshake(tag).
_rpc->send_request(HYPERDHT_CMD_PEER_HANDSHAKE, target,
hs_buf.data(), hs_buf.size(),
relay, false, hs->id);
return hs->id;
}
// ---- Listen ----
void HyperDht::listen() {
_listening = true;
// Self-announce under BLAKE2b(own_public_key).
// Matches JS: Server.listen() → dht.announce(hash(keyPair.publicKey), keyPair, ...)
// This allows connect_to_peer(pk) → find_peer(BLAKE2b(pk)) to find us.
crypto_generichash(_self_target, 32, _keypair.public_key, 32, nullptr, 0);
announce(_self_target);
_self_announce_count++;
// ⛔ AND SCHEDULE THE NEXT ONE. This used to announce once and stop.
//
// The self-announce is not a TopicDiscovery — it goes straight to announce()
// — so _tick_topics never saw it and nothing refreshed it. DHT records
// expire. When ours did, find_peer(our public key) found nothing and this
// peer was unreachable until the process restarted, while its topic records
// (which ARE refreshed) kept it looking discoverable.
//
// JS keeps this alive in a dedicated component, hyperdht's Announcer, one
// per server, targeting hash(keyPair.publicKey) — see SELF_ANNOUNCE_UPDATE_MS.
_self_announce_at = _last_tick_ms + SELF_ANNOUNCE_UPDATE_MS;
_self_ping_at = _last_tick_ms + SELF_ANNOUNCE_PING_MS;
}
int HyperDht::self_holder_count() const {
auto it = _announced_topics.find(_topic_hex(_self_target));
return it == _announced_topics.end() ? 0 : it->second.node_count;
}
int HyperDht::self_holder_tokens() const {
auto it = _announced_topics.find(_topic_hex(_self_target));
if (it == _announced_topics.end()) return 0;
int n = 0;
for (int i = 0; i < it->second.node_count; i++) {
static const uint8_t zero[32] = { 0 };
if (memcmp(it->second.nodes[i].token, zero, 32) != 0) n++;
}
return n;
}
void HyperDht::_ping_self_holders(uint64_t now_ms) {
auto it = _announced_topics.find(_topic_hex(_self_target));
if (it == _announced_topics.end() || it->second.node_count == 0) return;
const AnnouncedTopic &at = it->second;
// Judge the round that has just elapsed before starting the next one.
// JS: if (active < Math.min(pings.length, MIN_ACTIVE)) this.refresh()
if (_self_ping_sent > 0) {
const int needed = _self_ping_sent < SELF_ANNOUNCE_MIN_ACTIVE
? _self_ping_sent
: SELF_ANNOUNCE_MIN_ACTIVE;
if (_self_ping_alive < needed) {
DHT_DISC("[HyperDht] only %d/%d holders answered — re-announcing now "
"instead of in %llums\n", _self_ping_alive, _self_ping_sent,
(unsigned long long)(_self_announce_at - now_ms));
_self_announce_at = now_ms; // _tick_self_announce fires this tick
}
}
_self_ping_sent = 0;
_self_ping_alive = 0;
for (int i = 0; i < at.node_count; i++) {
_rpc->ping(at.nodes[i].address, HYPERDHT_TAG_SELF_PING);
_self_ping_sent++;
}
_self_ping_rounds++;
_self_ping_at = now_ms + SELF_ANNOUNCE_PING_MS;
}
void HyperDht::_tick_self_announce(uint64_t now_ms) {
if (!_listening || _self_announce_at == 0) return;
if (now_ms < _self_announce_at) return;
DHT_DISC("[HyperDht] re-announcing self %02x%02x%02x%02x...\n",
_self_target[0], _self_target[1], _self_target[2], _self_target[3]);
announce(_self_target);
_self_announce_count++;
_self_announce_at = now_ms + SELF_ANNOUNCE_UPDATE_MS;
}
// ---- Tick ----
void HyperDht::tick(uint64_t now_ms) {
// Clean up expired query infos.
for (int i = 0; i < _query_info_count; ) {
if (!_query_infos[i].active) {
_query_infos[i] = _query_infos[_query_info_count - 1];
_query_info_count--;
} else {
i++;
}
}
_last_tick_ms = now_ms;
// Are the nodes holding our record still there? Runs BEFORE the announce
// tick so a failed round can pull the re-announce forward into this tick.
if (_listening && _self_ping_at != 0 && now_ms >= _self_ping_at) {
_ping_self_holders(now_ms);
}
// Keep our own record alive. Without this a listening peer is announced
// exactly once and becomes unreachable when that record expires.
_tick_self_announce(now_ms);
// Reclaim stale handshakes. Without this the pool is one-way and a
// long-lived responder eventually stops answering anyone (§8).
_sweep_handshakes(now_ms);
// Tick topic discoveries (re-announce/re-lookup on schedule).
_tick_topics(now_ms);
}
// ---- RPC response handler ----
void HyperDht::_on_rpc_response(const DhtResponse &res, uint32_t tag) {
// A holder answered its liveness ping. Counted, not acted on here: the
// verdict is passed at the START of the next round, so a slow reply inside
// the window still counts. See _ping_self_holders.
if (tag == HYPERDHT_TAG_SELF_PING) {
_self_ping_alive++;
return;
}
// Check if this is a response to a handshake.
DhtHandshake *hs = _find_handshake(tag);
if (hs && hs->is_initiator) {
DHT_LOG("[HyperDht] handshake response: tag=%u value=%p len=%zu from=%d.%d.%d.%d:%d\n",
tag, (void *)res.value, res.value_len,
res.from.ip[0], res.from.ip[1], res.from.ip[2], res.from.ip[3], res.from.port);
}
if (hs && hs->is_initiator && res.value && res.value_len > 0) {
// Decode the handshake response.
EncodeState s = { 0, res.value_len, const_cast<uint8_t *>(res.value) };
HandshakeMessage hmsg;
if (dht_msg::handshake_decode(s, hmsg) &&
hmsg.mode == HS_MODE_REPLY &&
hmsg.noise && hmsg.noise_len > 0) {
// Extract server address from relay's peerAddress field.
// Matches JS: serverAddress = hs.peerAddress || to
if (hmsg.has_peer_address) {
hs->server_address = hmsg.peer_address;
hs->has_server_address = true;
} else {
// Not relayed — server address is the relay address itself.
hs->server_address = hs->relay_address;
hs->has_server_address = true;
}
// Process Noise IK Message 2.
auto payload_buf = hs->noise->recv(hmsg.noise, hmsg.noise_len);
if (hs->noise->is_complete() && !payload_buf.empty()) {
// Decode the remote's NoisePayload from the handshake response.
EncodeState pds = { 0, payload_buf.size(), payload_buf.data() };
dht_msg::noise_payload_decode(pds, hs->remote_payload);
hs->has_remote_payload = true;
_finalize_handshake(hs);
}
}
return;
}
// Handle storage responses (immutable/mutable GET).
_handle_storage_response(res, tag);
// Check if this is a lookup/find_peer response with peer data.
DHT_LOG("[HyperDht] _on_rpc_response: tag=%u value=%p len=%zu from=%d.%d.%d.%d:%d\n",
tag, (void *)res.value, res.value_len,
res.from.ip[0], res.from.ip[1], res.from.ip[2], res.from.ip[3], res.from.port);
// Check if this is a holepunch response (tag = 0x10000 | conn_id).
if (on_holepunch_response && (tag & 0xFFFF0000) == 0x10000) {
on_holepunch_response(tag, res.value, res.value_len, res.from);
return; // Holepunch responses are fully handled by the connector.
}
if (res.value && res.value_len > 0) {
DHT_LOG("[HyperDht] GOT VALUE! tag=%u len=%zu, searching %d query_infos\n",
tag, res.value_len, _query_info_count);
// Route response value to the specific query that issued this request.
// JS: each query has its own map/onresult callback — responses are never broadcast.
bool matched = false;
for (int i = 0; i < _query_info_count; i++) {
if (!_query_infos[i].active) continue;
DHT_LOG("[HyperDht] qi[%d] qid=%u announce=%d lookup=%d findpeer=%d\n",
i, _query_infos[i].query_id,
_query_infos[i].is_announce,
_query_infos[i].is_lookup,
_query_infos[i].is_find_peer);
if (_query_infos[i].query_id != tag) continue; // Match by query ID.
matched = true;
if (_query_infos[i].is_find_peer) {
DHT_LOG("[HyperDht] routing to _process_find_peer_value\n");
_process_find_peer_value(_query_infos[i].query_id,
res.value, res.value_len, res.from);
} else if (_query_infos[i].is_lookup) {
DHT_DISC("[HyperDht] routing to _process_lookup_value\n");
_process_lookup_value(_query_infos[i].query_id,
res.value, res.value_len, res.from);
} else if (_query_infos[i].is_announce) {
// Announce traversal uses LOOKUP command — DHT nodes return
// peer values (other peers on this topic) during traversal.
// Process them as lookup values so peer discovery works.
DHT_DISC("[HyperDht] routing announce-phase value to _process_lookup_value\n");
_process_lookup_value(_query_infos[i].query_id,
res.value, res.value_len, res.from);
} else {
DHT_LOG("[HyperDht] VALUE DROPPED: unknown query type\n");
}
break; // Found matching query.
}
if (!matched) {
DHT_LOG("[HyperDht] VALUE DROPPED: no matching query_info for tag=%u\n", tag);
}
}
}
// ---- RPC request handler ----
void HyperDht::_on_rpc_request(const DhtRequest &req) {
if (req.command == HYPERDHT_CMD_PEER_HANDSHAKE) {
_handle_handshake(req);
return; // Handshake sends its own response.
}
if (req.command == HYPERDHT_CMD_PEER_HOLEPUNCH) {
// Incoming holepunch from relay (FROM_RELAY mode).
// Dispatch to connector which decrypts and processes the payload,
// then we reply with our holepunch response.
DHT_LOG("[HyperDht] incoming PEER_HOLEPUNCH from %d.%d.%d.%d:%d\n",
req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port);
if (on_holepunch_request) {
on_holepunch_request(req);
}
// Reply with closer nodes (relay expects a response).
_reply_to_query(req);
return;
}
if (req.command >= HYPERDHT_CMD_MUTABLE_PUT &&
req.command <= HYPERDHT_CMD_IMMUTABLE_GET) {
_handle_storage_request(req);
}
// For ALL non-internal queries (FIND_PEER, LOOKUP, ANNOUNCE, etc.),
// respond with closer nodes + token + optional value.
// This matches JS dht-rpc behavior where every node responds to queries.
// Without this, iterative queries that visit our node would time out.
_reply_to_query(req);
}
void HyperDht::_reply_to_query(const DhtRequest &req) {
// Compute closer nodes from routing table.
IPv4Address closer[20]; // DHT_MAX_CLOSER_NODES
int closer_count = 0;
if (req.has_target) {
closer_count = _rpc->table().closest_addresses(req.target, closer, 20);
}
uint8_t token[32];
_rpc->generate_token(req.from, token);
// Check if we have a local record for this target.
const uint8_t *value = nullptr;
size_t value_len = 0;
std::vector<uint8_t> record;
if (_listening && req.has_target) {
// Check self-announce target: BLAKE2b(own_pk).
uint8_t self_target[32];
crypto_generichash(self_target, 32, _keypair.public_key, 32, nullptr, 0);
bool target_match = (memcmp(req.target, self_target, 32) == 0);
// Also check topic announces.
if (!target_match) {
for (int i = 0; i < _topic_count; i++) {
if (_topics[i].active && memcmp(_topics[i].topic, req.target, 32) == 0) {
target_match = true;
break;
}
}
}
if (target_match) {
// Build our peer record to return.
dht_msg::PeerRecord pr;
memset(&pr, 0, sizeof(pr));
memcpy(pr.public_key, _keypair.public_key, 32);
pr.relay_count = 0;
EncodeState ps = { 0, 0, nullptr };
dht_msg::peer_record_preencode(ps, pr);
record.resize(ps.end);
ps.buffer = record.data();
ps.start = 0;
dht_msg::peer_record_encode(ps, pr);
value = record.data();
value_len = record.size();
DHT_LOG("[HyperDht] _reply_to_query: returning own record len=%zu for cmd=%u\n",
value_len, req.command);
}
}
_rpc->_send_response(req.tid, req.from, true, token,
closer, closer_count, 0, value, value_len);
}
// ---- Handle incoming handshake ----
void HyperDht::_handle_handshake(const DhtRequest &req) {
DHT_LOG("[HyperDht] _handle_handshake: from=%d.%d.%d.%d:%d value_len=%zu\n",
req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port,
req.value_len);
if (!_listening) return;
if (!req.value || req.value_len == 0) return;
// Decode HandshakeMessage.
EncodeState s = { 0, req.value_len, const_cast<uint8_t *>(req.value) };
HandshakeMessage hmsg;
if (!dht_msg::handshake_decode(s, hmsg)) return;
if (!hmsg.noise || hmsg.noise_len == 0) return;
// Send a Noise message 2 for THIS request, wrapped as its mode needs (see below).
auto send_reply = [&](const std::vector<uint8_t> &nr) {
HandshakeMessage reply_hmsg;
memset(&reply_hmsg, 0, sizeof(reply_hmsg));
reply_hmsg.noise = nr.data();
reply_hmsg.noise_len = nr.size();
if (hmsg.mode == HS_MODE_FROM_RELAY && hmsg.has_peer_address) {
// Relay protocol: send FROM_SERVER REQUEST back to the relay node.
reply_hmsg.mode = HS_MODE_FROM_SERVER;
reply_hmsg.peer_address = hmsg.peer_address; // Original client address.
reply_hmsg.has_peer_address = true;
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",
req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port,
hmsg.peer_address.ip[0], hmsg.peer_address.ip[1], hmsg.peer_address.ip[2], hmsg.peer_address.ip[3], hmsg.peer_address.port);
EncodeState rhs = { 0, 0, nullptr };
dht_msg::handshake_preencode(rhs, reply_hmsg);
std::vector<uint8_t> reply_hs_buf(rhs.end);
rhs.buffer = reply_hs_buf.data();
rhs.start = 0;
dht_msg::handshake_encode(rhs, reply_hmsg);
// Send as REQUEST (not response!) so the relay can process FROM_SERVER.
// Use the same TID so the relay chain preserves the transaction.
_rpc->_send_request(req.tid, req.from, nullptr,
false, HYPERDHT_CMD_PEER_HANDSHAKE,
req.has_target ? req.target : nullptr,
reply_hs_buf.data(), reply_hs_buf.size());
} else {
// Direct connection (FROM_CLIENT): reply with REPLY as a response.
reply_hmsg.mode = HS_MODE_REPLY;
DHT_LOG("[HyperDht] _handle_handshake: direct mode, sending REPLY response to %d.%d.%d.%d:%d\n",
req.from.ip[0], req.from.ip[1], req.from.ip[2], req.from.ip[3], req.from.port);
EncodeState rhs = { 0, 0, nullptr };
dht_msg::handshake_preencode(rhs, reply_hmsg);
std::vector<uint8_t> reply_hs_buf(rhs.end);
rhs.buffer = reply_hs_buf.data();
rhs.start = 0;
dht_msg::handshake_encode(rhs, reply_hmsg);
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 0,
reply_hs_buf.data(), reply_hs_buf.size());
}
};
// A repeat of a handshake we are still holding: the same reply, nothing new
// (hyperdht server.js _onpeerhandshake keys by hex(noise) and reuses the reply).
uint8_t noise_key[32];
crypto_generichash(noise_key, 32, hmsg.noise, hmsg.noise_len, nullptr, 0);
for (int i = 0; i < _handshake_count; i++) {
DhtHandshake &held = _handshakes[i];
if (!held.active || held.is_initiator || !held.has_noise_key || !held.reply_noise) continue;
if (memcmp(held.noise_key, noise_key, 32) != 0) continue;
DHT_LOG("[HyperDht] _handle_handshake: repeat of handshake id=%u — resending its reply\n", held.id);
send_reply(*held.reply_noise);
return;
}
// Create responder handshake.
DhtHandshake *hs = _alloc_handshake(false, nullptr, req.from);
if (!hs) return;
memcpy(hs->noise_key, noise_key, 32);
hs->has_noise_key = true;
// Save client address — needed so the responder can finalize the
// UDX connection back to the client (LAN check or holepunch target).
if (hmsg.mode == HS_MODE_FROM_RELAY && hmsg.has_peer_address) {
// Relay protocol: client address is in the relay's peerAddress field.
hs->client_address = hmsg.peer_address;
hs->has_client_address = true;
} else if (hmsg.mode == HS_MODE_FROM_CLIENT) {
// Direct connection: client address is simply req.from.
hs->client_address = req.from;
hs->has_client_address = true;
}
hs->noise = new NoiseIK(false, _keypair, _ns_peer_handshake, nullptr);
// Process Noise IK Message 1.
auto payload_buf = hs->noise->recv(hmsg.noise, hmsg.noise_len);
if (payload_buf.empty()) {
_free_handshake(hs);
return;
}
// Store remote public key.
memcpy(hs->remote_public_key, hs->noise->remote_public_key(), 32);
// Decode the NoisePayload from the decrypted payload.
dht_msg::NoisePayload remote_payload;
EncodeState ps = { 0, payload_buf.size(), payload_buf.data() };
dht_msg::noise_payload_decode(ps, remote_payload);
hs->firewall = remote_payload.firewall;
// Notify listener.
if (on_handshake_request) {
on_handshake_request(hs->id, hs->remote_public_key, remote_payload);
}
// Build response NoisePayload.
// Matches JS server.js _addHandshake(): includes error, firewall, addresses, udx, secretStream.
dht_msg::NoisePayload reply_payload;
memset(&reply_payload, 0, sizeof(reply_payload));
reply_payload.version = 1;
reply_payload.error = 0; // ERROR.NONE — client sent valid version + udx.
reply_payload.firewall = _rpc->local_firewall();
// Include our addresses (local + NAT-detected external) so client
// can attempt direct connect.
reply_payload.addresses4_count = _rpc->get_all_addresses(reply_payload.addresses4, 8);
// UDX info — REQUIRED by JS client to set up the encrypted stream.
// Without this, JS client rejects with ERROR.ABORTED.
reply_payload.has_udx = true;
reply_payload.udx.reusable_socket = false;
reply_payload.udx.id = hs->id; // Stream ID for UDX.
reply_payload.udx.seq = 0;
// SecretStream marker — tells client we support @hyperswarm/secret-stream.
reply_payload.has_secret_stream = true;
EncodeState rps = { 0, 0, nullptr };
dht_msg::noise_payload_preencode(rps, reply_payload);
std::vector<uint8_t> reply_buf(rps.end);
rps.buffer = reply_buf.data();
rps.start = 0;
dht_msg::noise_payload_encode(rps, reply_payload);
// Noise IK Message 2 (responder → initiator).
auto noise_reply = hs->noise->send(reply_buf.data(), reply_buf.size());
// Build the reply HandshakeMessage based on incoming mode.
// The relay protocol is a multi-hop chain:
// Client → Relay: FROM_CLIENT (request)
// Relay → Server: FROM_RELAY (request, peerAddress=client)
// Server → Relay: FROM_SERVER (request, peerAddress=client)
// Relay → Client: REPLY (response)
//
// When we receive FROM_RELAY, we must send FROM_SERVER as a REQUEST
// (not a response!) back to the relay. The relay then sends the final
// REPLY response to the original client.
// When we receive FROM_CLIENT directly, we reply with REPLY as a response.
send_reply(noise_reply);
hs->reply_noise = new std::vector<uint8_t>(noise_reply);
// Finalize.
if (hs->noise->is_complete()) {
_finalize_handshake(hs);
}
}
// ---- Process find_peer value ----
// FIND_PEER returns a single peer record (encoded m.peer) from the DHT node
// that has the announce forwarding entry. The `from` address of the response
// IS the relay node — the DHT node we send PEER_HANDSHAKE to.
void HyperDht::_process_find_peer_value(uint32_t query_id, const uint8_t *value,
size_t value_len, const IPv4Address &from) {
EncodeState s = { 0, value_len, const_cast<uint8_t *>(value) };
dht_msg::PeerRecord record;
if (!dht_msg::peer_record_decode(s, record)) return;
FoundPeer fp;
memcpy(fp.public_key, record.public_key, 32);
memcpy(fp.relay_addresses, record.relay_addresses,
record.relay_count * sizeof(IPv4Address));
fp.relay_count = record.relay_count;
fp.relay_node = from; // The DHT node that stored this peer — send PEER_HANDSHAKE here.
_note_peer_for_query(query_id);
if (on_peer_found) on_peer_found(query_id, fp);
}
// ---- Process lookup value ----
void HyperDht::_process_lookup_value(uint32_t query_id,
const uint8_t *value, size_t value_len,
const IPv4Address &from) {
DHT_DISC("[HyperDht] _process_lookup_value: qid=%u len=%zu from=%d.%d.%d.%d:%d\n",
query_id, value_len,
from.ip[0], from.ip[1], from.ip[2], from.ip[3], from.port);
EncodeState s = { 0, value_len, const_cast<uint8_t *>(value) };
dht_msg::LookupReply reply;
if (!dht_msg::lookup_reply_decode(s, reply)) {
DHT_DISC("[HyperDht] _process_lookup_value: DECODE FAILED! hex[0..8]=");
for (size_t b = 0; b < value_len && b < 8; b++) DHT_LOG("%02x", value[b]);
DHT_LOG("\n");
return;
}
DHT_DISC("[HyperDht] _process_lookup_value: decoded %d peers\n", reply.peer_count);
for (int i = 0; i < reply.peer_count; i++) {
// Skip peers with all-zero public keys (truncated buffer).
static const uint8_t zero_pk[32] = {};
if (memcmp(reply.peers[i].public_key, zero_pk, 32) == 0) continue;
DHT_DISC("[HyperDht] _process_lookup_value: peer[%d] pk=%02x%02x%02x%02x... relays=%d\n",
i, reply.peers[i].public_key[0], reply.peers[i].public_key[1],
reply.peers[i].public_key[2], reply.peers[i].public_key[3],
reply.peers[i].relay_count);
FoundPeer fp;
memset(&fp, 0, sizeof(fp));
memcpy(fp.public_key, reply.peers[i].public_key, 32);
memcpy(fp.relay_addresses, reply.peers[i].relay_addresses,
reply.peers[i].relay_count * sizeof(IPv4Address));
fp.relay_count = reply.peers[i].relay_count;
fp.relay_node = from; // The DHT node that stored this peer — send PEER_HANDSHAKE here.
_note_peer_for_query(query_id);
if (on_peer_found) on_peer_found(query_id, fp);
}
}
// ---- Announce encoding ----
// ⛔ THE SIGNABLE IS BUILT HERE AND NOWHERE ELSE.
//
// It used to be built twice — once in the announce encoder, once in the
// unannounce encoder — and the two DIVERGED. Announce carried the correct
// 64-byte ns || generichash_batch(...); unannounce carried a variable-length
// ns || topic || peer. JS verifies with annSignable() for BOTH
// (persistent.js:282), so every unannounce we ever sent failed
// crypto_sign_verify_detached and was DISCARDED SILENTLY (persistent.js:64) —
// a peer that left never withdrew its record, and the topic kept answering for
// a daemon that was gone.
//
// JS has exactly one function for both namespaces. That is why theirs could not
// drift, and it is why there is one here now.
void HyperDht::ann_signable(uint8_t *out64, const uint8_t *ns,
const uint8_t *target, const uint8_t *node_id,
const uint8_t *token,
const uint8_t *peer_buf, size_t peer_len) {
memcpy(out64, ns, 32);
// JS: crypto_generichash_batch(hash, [target, id, token, encode(m.peer, ann.peer), ann.refresh || EMPTY])
// The refresh is empty for us. Feeding zero bytes to blake2b is a no-op, so
// it is absent here rather than fed as an empty buffer — the digest is the
// same either way, and this says so instead of leaving a reader to wonder.
crypto_generichash_state st;
crypto_generichash_init(&st, nullptr, 0, 32);
crypto_generichash_update(&st, target, 32);
crypto_generichash_update(&st, node_id, 32);
crypto_generichash_update(&st, token, 32);
crypto_generichash_update(&st, peer_buf, peer_len);
crypto_generichash_final(&st, out64 + 32, 32);
}
std::vector<uint8_t> HyperDht::_encode_ann_value_for_node(
const uint8_t *ns, const uint8_t *topic, const uint8_t *token,
const uint8_t *node_id) {
dht_msg::AnnounceMessage msg;
memset(&msg, 0, sizeof(msg));
// Peer record: our public key, no relay addresses.
msg.has_peer = true;
memcpy(msg.peer.public_key, _keypair.public_key, 32);
msg.peer.relay_count = 0;
msg.has_signature = true;
// The peer record as it goes on the wire — the signable hashes THIS, not the
// fields, so the two can never disagree about padding or order.
EncodeState ps = { 0, 0, nullptr };
dht_msg::peer_record_preencode(ps, msg.peer);
std::vector<uint8_t> peer_buf(ps.end);
ps.buffer = peer_buf.data();
ps.start = 0;
dht_msg::peer_record_encode(ps, msg.peer);
uint8_t signable[64];
ann_signable(signable, ns, topic, node_id, token, peer_buf.data(), peer_buf.size());
crypto_sign_detached(msg.signature, nullptr, signable, 64, _keypair.secret_key);
EncodeState as = { 0, 0, nullptr };
dht_msg::announce_preencode(as, msg);
std::vector<uint8_t> buf(as.end);
as.buffer = buf.data();
as.start = 0;
dht_msg::announce_encode(as, msg);
return buf;
}
std::vector<uint8_t> HyperDht::_encode_announce_value_for_node(
const uint8_t *topic, const uint8_t *token, const uint8_t *node_id) {
return _encode_ann_value_for_node(_ns_announce, topic, token, node_id);
}
void HyperDht::_sign_announce(const uint8_t *signable, size_t signable_len,
uint8_t *signature) {
crypto_sign_detached(signature, nullptr, signable, signable_len,
_keypair.secret_key);
}
void HyperDht::_note_announce_holder(const uint8_t *topic, const IPv4Address &from,
const uint8_t *token) {
// ⛔ THE NODES THAT ACTUALLY TOOK THE ANNOUNCE, recorded as they commit.
//
// This used to be _store_announced_nodes(), which ran after the query and
// filled the table from `table().closest()` — our own routing table's idea
// of who is nearby, not who accepted anything — and then wrote a zeroed
// token with the comment "Token not available here". So we could not say
// which nodes held our record, could not check whether they were alive, and
// could not produce a per-node token for anything later.
//
// JS records the committing node from the reply itself
// (announcer.js:265-267). `from` here is that same reply's origin.
std::string key = _topic_hex(topic);
AnnouncedTopic &at = _announced_topics[key];
memcpy(at.topic, topic, 32);
// A node re-committing on a refresh updates its token rather than adding a
// duplicate — the token rotates and the stale one is worthless.
for (int i = 0; i < at.node_count; i++) {
if (memcmp(at.nodes[i].address.ip, from.ip, 4) == 0 &&
at.nodes[i].address.port == from.port) {
memcpy(at.nodes[i].token, token, 32);
return;
}
}
if (at.node_count >= 20) return;
at.nodes[at.node_count].address = from;
memcpy(at.nodes[at.node_count].token, token, 32);
at.node_count++;
}
std::string HyperDht::_topic_hex(const uint8_t *topic) {
static const char hex_chars[] = "0123456789abcdef";
std::string result(64, '0');
for (int i = 0; i < 32; i++) {
result[i * 2] = hex_chars[(topic[i] >> 4) & 0x0f];
result[i * 2 + 1] = hex_chars[topic[i] & 0x0f];
}
return result;
}
// ---- Topic discovery management ----
// Matches JS Hyperswarm's PeerDiscovery/PeerDiscoverySession.
// - join_topic() creates or reuses a TopicDiscovery, increments session count.
// - leave_topic() decrements; unannounces when server sessions hit 0; frees when all 0.
// - _tick_topics() handles scheduled re-announce/re-lookup (10min + jitter).
bool HyperDht::join_topic(const uint8_t *topic, bool server, bool client) {
if (!server && !client) return false;
TopicDiscovery *td = _find_topic_mut(topic);
if (!td) {
td = _alloc_topic(topic);
if (!td) return false; // Slots exhausted.
}
// Increment session counters (matches JS PeerDiscoverySession.refresh()).
if (server) td->server_sessions++;
if (client) td->client_sessions++;
// Immediate refresh on first join — matches JS PeerDiscovery initial refresh.
// JS Hyperswarm does: await announce(), then lookup() (sequential, not else-if).
if (td->last_refresh == 0) {
if (td->server_sessions > 0) {
td->needs_unannounce = true;
td->current_query_id = announce(topic);
// If also a client, queue lookup for after announce completes.
// Matches JS: await announce(), then for await (lookup()) {...}
if (td->client_sessions > 0) {
td->pending_lookup = true;
}
} else if (td->client_sessions > 0) {
td->current_query_id = lookup(topic);
}
td->last_refresh = 1; // Mark as "has been refreshed" (real time set in _tick).
}
return true;
}
void HyperDht::leave_topic(const uint8_t *topic) {
TopicDiscovery *td = _find_topic_mut(topic);
if (!td) return;
// Decrement session counts. We decrement ONE server AND ONE client session
// (matching a single join_topic(server=true, client=true) call).
// For more precise session tracking, the caller should track which sessions
// they hold. This simplified version assumes symmetric join/leave.
if (td->server_sessions > 0) td->server_sessions--;
if (td->client_sessions > 0) td->client_sessions--;
// If server sessions dropped to 0 and we announced, unannounce.
if (td->server_sessions == 0 && td->needs_unannounce) {
unannounce(topic);
td->needs_unannounce = false;
}
// If all sessions are gone, remove the discovery entirely.
if (td->server_sessions == 0 && td->client_sessions == 0) {
_free_topic(td);
}
}
const TopicDiscovery *HyperDht::find_topic(const uint8_t *topic) const {
for (int i = 0; i < _topic_count; i++) {
if (_topics[i].active && memcmp(_topics[i].topic, topic, 32) == 0) {
return &_topics[i];
}
}
return nullptr;
}
const uint8_t *HyperDht::get_query_topic(uint32_t query_id) const {
for (int i = 0; i < _query_info_count; i++) {
if (_query_infos[i].active && _query_infos[i].query_id == query_id &&
(_query_infos[i].is_lookup || _query_infos[i].is_announce)) {
return _query_infos[i].topic;
}
}
return nullptr;
}
TopicDiscovery *HyperDht::_find_topic_mut(const uint8_t *topic) {
for (int i = 0; i < _topic_count; i++) {
if (_topics[i].active && memcmp(_topics[i].topic, topic, 32) == 0) {
return &_topics[i];
}
}
return nullptr;
}
TopicDiscovery *HyperDht::_alloc_topic(const uint8_t *topic) {
if (_topic_count >= TOPIC_MAX_DISCOVERIES) return nullptr;
TopicDiscovery *td = &_topics[_topic_count++];
memset(td, 0, sizeof(TopicDiscovery));
memcpy(td->topic, topic, 32);
td->active = true;
return td;
}
void HyperDht::_free_topic(TopicDiscovery *td) {
td->active = false;
int idx = static_cast<int>(td - _topics);
if (idx < _topic_count - 1) {
_topics[idx] = _topics[_topic_count - 1];
}
_topic_count--;
}
void HyperDht::_tick_topics(uint64_t now_ms) {
for (int i = 0; i < _topic_count; i++) {
TopicDiscovery *td = &_topics[i];
if (!td->active) continue;
// Schedule initial refresh time if not set.
if (td->refresh_at == 0) {
_schedule_topic_refresh(td, now_ms, false);
td->last_refresh = now_ms;
continue;
}
// Check if it's time to refresh.
if (now_ms >= td->refresh_at) {
// Check for overdue (laptop wake grace period).
// Matches JS: if (Date.now() - startTime > delay + DELAY_GRACE_PERIOD)
uint64_t expected_delay = td->refresh_at - td->last_refresh;
bool overdue = (now_ms - td->last_refresh) > expected_delay + TOPIC_REFRESH_GRACE_MS;
if (overdue) {
// Overdue — schedule eager refresh (jitter only, no 10min wait).
_schedule_topic_refresh(td, now_ms, true);
} else {
// Normal refresh.
_refresh_topic(td, now_ms);
}
}
}
}
void HyperDht::_refresh_topic(TopicDiscovery *td, uint64_t now_ms) {
td->last_refresh = now_ms;
// Announce then lookup (matches JS PeerDiscovery._refresh — both run, not else-if).
if (td->server_sessions > 0) {
td->needs_unannounce = true;
td->current_query_id = announce(td->topic);
// Chain lookup after announce completes (matches JS sequential behavior).
if (td->client_sessions > 0) {
td->pending_lookup = true;
}
} else if (td->client_sessions > 0) {
// If we previously announced but have no server sessions, do lookup+unannounce.
if (td->needs_unannounce) {
unannounce(td->topic);
td->needs_unannounce = false;
}
td->current_query_id = lookup(td->topic);
}
// Schedule next refresh.
_schedule_topic_refresh(td, now_ms, false);
}
void HyperDht::_schedule_topic_refresh(TopicDiscovery *td, uint64_t now_ms, bool eager) {
// Matches JS: const delay = !eager ? REFRESH_INTERVAL + jitter : jitter
uint64_t jitter = _random_jitter();
uint64_t delay = eager ? jitter : (TOPIC_REFRESH_INTERVAL_MS + jitter);
// ⛔ A TOPIC LOOKING FOR PEERS AND FINDING NONE RETRIES SOON, NOT IN TEN
// MINUTES. See TOPIC_DISCOVERY_RETRY_MS. The condition is deliberately
// narrow: client sessions only (a pure server has nothing to look up), and
// only until the FIRST peer is seen, after which this topic is on the JS
// cadence for the rest of its life.
if (!eager && td->client_sessions > 0 && td->peers_seen == 0) {
int shift = td->empty_refreshes;
if (shift > TOPIC_DISCOVERY_RETRY_SHIFTS) shift = TOPIC_DISCOVERY_RETRY_SHIFTS;
uint64_t backoff = TOPIC_DISCOVERY_RETRY_MS << shift;
if (backoff > TOPIC_DISCOVERY_RETRY_MAX_MS) backoff = TOPIC_DISCOVERY_RETRY_MAX_MS;
// A small jitter so two peers that started together do not stay in
// lockstep, querying at the same instant on every retry for ever.
delay = backoff + (jitter % 1000);
if (td->empty_refreshes < TOPIC_DISCOVERY_RETRY_SHIFTS + 1) td->empty_refreshes++;
DHT_DISC("[HyperDht] topic %02x%02x%02x%02x: no peers yet, retrying in %llums\n",
td->topic[0], td->topic[1], td->topic[2], td->topic[3],
(unsigned long long)delay);
}
td->refresh_at = now_ms + delay;
}
void HyperDht::_note_peer_for_query(uint32_t query_id) {
// A peer arrived on this query — find the topic that issued it and mark it
// as having succeeded, which retires the early-retry schedule above.
for (int i = 0; i < _topic_count; i++) {
TopicDiscovery *td = &_topics[i];
if (td->active && td->current_query_id == query_id) {
if (td->peers_seen == 0) {
DHT_DISC("[HyperDht] topic %02x%02x%02x%02x: first peer — "
"back to the 10-minute cadence\n",
td->topic[0], td->topic[1], td->topic[2], td->topic[3]);
}
td->peers_seen++;
td->empty_refreshes = 0;
return;
}
}
}
uint64_t HyperDht::_random_jitter() const {
// Random value in [0, TOPIC_REFRESH_JITTER_MS).
return static_cast<uint64_t>(randombytes_uniform(
static_cast<uint32_t>(TOPIC_REFRESH_JITTER_MS)));
}
// ---- Handshake management ----
DhtHandshake *HyperDht::_alloc_handshake(bool initiator, const uint8_t *remote_pk,
const IPv4Address &relay) {
if (_handshake_count >= MAX_HANDSHAKES) {
// ⛔ NEVER FAIL SILENTLY HERE. This used to be a bare `return nullptr`
// whose caller answered with a bare `return`, so a responder that had
// filled its pool stopped replying to every peer, for ever, with nothing
// in any log to say so. Three days of hyper-games #72 were spent looking
// at the initiator because this end produced no evidence at all.
fprintf(stderr, "[HyperDht] handshake pool FULL (%d/%d) — refusing a %s "
"handshake. Peers will see silence.\n",
_handshake_count, MAX_HANDSHAKES, initiator ? "outgoing" : "incoming");
return nullptr;
}
DhtHandshake &hs = _handshakes[_handshake_count++];
memset(&hs, 0, sizeof(hs));
hs.created_at_ms = _last_tick_ms;
hs.id = ++_handshake_id_counter;
hs.is_initiator = initiator;
hs.relay_address = relay;
hs.active = true;
if (remote_pk) {
memcpy(hs.remote_public_key, remote_pk, 32);
}
return &hs;
}
DhtHandshake *HyperDht::_find_handshake(uint32_t id) {
for (int i = 0; i < _handshake_count; i++) {
if (_handshakes[i].active && _handshakes[i].id == id) {
return &_handshakes[i];
}
}
return nullptr;
}
void HyperDht::_free_handshake(DhtHandshake *hs) {
delete hs->noise;
hs->noise = nullptr;
delete hs->reply_noise; // before compaction copies the last slot over this one
hs->reply_noise = nullptr;
hs->active = false;
int idx = static_cast<int>(hs - _handshakes);
if (idx < _handshake_count - 1) {
_handshakes[idx] = _handshakes[_handshake_count - 1];
}
_handshake_count--;
}
void HyperDht::_sweep_handshakes(uint64_t now) {
// Walk WITHOUT advancing on a free: _free_handshake compacts by moving the
// last entry into the freed slot, so the element now at `i` has not been
// examined yet. Incrementing here would skip it.
for (int i = 0; i < _handshake_count; ) {
DhtHandshake &hs = _handshakes[i];
// Allocated before any tick, so it has no birth time yet. Adopt it now
// rather than treating age-since-epoch as "infinitely old", which would
// reclaim a handshake allocated microseconds before the first tick.
if (hs.created_at_ms == 0) {
hs.created_at_ms = now;
i++;
continue;
}
if (now - hs.created_at_ms > HANDSHAKE_TIMEOUT_MS) {
DHT_LOG("[HyperDht] reclaiming handshake id=%u after %llums\n",
hs.id, (unsigned long long)(now - hs.created_at_ms));
_free_handshake(&hs);
} else {
i++;
}
}
}
void HyperDht::_finalize_handshake(DhtHandshake *hs) {
if (!hs->noise || !hs->noise->is_complete()) return;
// Store remote public key.
memcpy(hs->remote_public_key, hs->noise->remote_public_key(), 32);
// Copy derived keys.
memcpy(hs->tx_key, hs->noise->tx_key(), 32);
memcpy(hs->rx_key, hs->noise->rx_key(), 32);
memcpy(hs->hash, hs->noise->handshake_hash(), 64);
// Derive holepunch secret.
hs->noise->derive_holepunch_secret(_ns_peer_holepunch, hs->holepunch_secret);
hs->complete = true;
if (on_handshake_complete) {
on_handshake_complete(hs->id, *hs);
}
}
// ---- DHT Storage API ----
// Matches JS hyperdht immutablePut/Get, mutablePut/Get (index.js + persistent.js).
void HyperDht::immutable_put(const uint8_t *value, size_t value_len) {
// target = BLAKE2b(value) — content-addressed.
uint8_t target[32];
crypto_generichash(target, 32, value, value_len, nullptr, 0);
// Store locally first.
uint8_t key[32];
_storage.put_immutable(value, value_len, key);
// Matches JS immutablePut pattern:
// Traversal: IMMUTABLE_GET with no value (finds K closest, collects tokens)
// Commit: IMMUTABLE_PUT with value + token (stores on closest nodes)
uint32_t qid = _rpc->query_with_commit(
target, false,
HYPERDHT_CMD_IMMUTABLE_GET, nullptr, 0, // Traversal: GET, no value
HYPERDHT_CMD_IMMUTABLE_PUT, value, value_len, // Commit: PUT with value
0);
if (_query_info_count < HYPERDHT_MAX_QUERIES) {
auto &qi = _query_infos[_query_info_count++];
qi.query_id = qid;
memset(qi.topic, 0, 32);
memcpy(qi.storage_key, target, 32);
qi.is_announce = false;
qi.is_lookup = false;
qi.is_find_peer = false;
qi.is_immutable_get = false;
qi.is_immutable_put = true;
qi.is_mutable_get = false;
qi.is_mutable_put = false;
qi.mutable_seq = 0;
qi.put_value.assign(value, value + value_len);
memset(qi.put_signature, 0, 64);
qi.active = true;
}
}
void HyperDht::immutable_get(const uint8_t *key) {
// Check local cache first.
std::vector<uint8_t> cached;
if (_storage.get_immutable(key, cached)) {
if (on_immutable_get_complete) {
on_immutable_get_complete(key, cached.data(), cached.size());
}
return;
}
// Query network — IMMUTABLE_GET with target=key.
uint32_t qid = _rpc->query(key, false, HYPERDHT_CMD_IMMUTABLE_GET,
nullptr, 0, 0, false);
if (_query_info_count < HYPERDHT_MAX_QUERIES) {
auto &qi = _query_infos[_query_info_count++];
qi.query_id = qid;
memset(qi.topic, 0, 32);
memcpy(qi.storage_key, key, 32);
qi.is_announce = false;
qi.is_lookup = false;
qi.is_find_peer = false;
qi.is_immutable_get = true;
qi.is_immutable_put = false;
qi.is_mutable_get = false;
qi.is_mutable_put = false;
qi.mutable_seq = 0;
qi.put_value.clear();
memset(qi.put_signature, 0, 64);
qi.active = true;
}
}
void HyperDht::mutable_put(const uint8_t *value, size_t value_len, uint64_t seq) {
// target = BLAKE2b(publicKey).
uint8_t target[32];
crypto_generichash(target, 32, _keypair.public_key, 32, nullptr, 0);
// Sign: signable = NS.MUTABLE_PUT(32) || BLAKE2b(encode(seq, value))(32).
uint8_t signature[64];
DhtStorage::sign_mutable(seq, value, value_len, _ns_mutable_put,
_keypair.secret_key, signature);
// Encode MutablePutRequest for the wire.
// Wire: fixed32(publicKey) + uint(seq) + buffer(value) + fixed64(signature).
MutablePutRequest mpr;
memcpy(mpr.public_key, _keypair.public_key, 32);
mpr.seq = seq;
mpr.value = value;
mpr.value_len = value_len;
memcpy(mpr.signature, signature, 64);
EncodeState ps = { 0, 0, nullptr };
dht_msg::mutable_put_preencode(ps, mpr);
std::vector<uint8_t> encoded(ps.end);
ps.buffer = encoded.data();
ps.start = 0;
dht_msg::mutable_put_encode(ps, mpr);
// Store locally.
_storage.put_mutable(_keypair.public_key, seq, value, value_len,
signature, _ns_mutable_put);
// Matches JS mutablePut pattern:
// Traversal: MUTABLE_GET with seq=0 (finds K closest, collects tokens)
// Commit: MUTABLE_PUT with encoded MutablePutRequest + token
EncodeState seq0s = { 0, 0, nullptr };
cenc::uint_preencode(seq0s, (uint64_t)0);
std::vector<uint8_t> seq0_buf(seq0s.end);
seq0s.buffer = seq0_buf.data();
seq0s.start = 0;
cenc::uint_encode(seq0s, (uint64_t)0);
uint32_t qid = _rpc->query_with_commit(
target, false,
HYPERDHT_CMD_MUTABLE_GET, seq0_buf.data(), seq0_buf.size(), // Traversal: GET, seq=0
HYPERDHT_CMD_MUTABLE_PUT, encoded.data(), encoded.size(), // Commit: PUT with value
0);
if (_query_info_count < HYPERDHT_MAX_QUERIES) {
auto &qi = _query_infos[_query_info_count++];
qi.query_id = qid;
memcpy(qi.topic, _keypair.public_key, 32);
memcpy(qi.storage_key, target, 32);
qi.is_announce = false;
qi.is_lookup = false;
qi.is_find_peer = false;
qi.is_immutable_get = false;
qi.is_immutable_put = false;
qi.is_mutable_get = false;
qi.is_mutable_put = true;
qi.mutable_seq = seq;
qi.put_value.assign(value, value + value_len);
memcpy(qi.put_signature, signature, 64);
qi.active = true;
}
}
void HyperDht::mutable_get(const uint8_t *public_key, uint64_t min_seq) {
// target = BLAKE2b(publicKey).
uint8_t target[32];
crypto_generichash(target, 32, public_key, 32, nullptr, 0);
// Check local cache first.
MutableGetResponse cached;
if (_storage.get_mutable(public_key, min_seq, cached)) {
if (on_mutable_get_complete) {
on_mutable_get_complete(public_key, cached.value.data(),
cached.value.size(), cached.seq);
}
return;
}
// Encode min_seq as value for MUTABLE_GET request.
// Matches JS: value = c.encode(c.uint, userSeq).
EncodeState vs = { 0, 0, nullptr };
cenc::uint_preencode(vs, min_seq);
std::vector<uint8_t> seq_buf(vs.end);
vs.buffer = seq_buf.data();
vs.start = 0;
cenc::uint_encode(vs, min_seq);
// Query network.
uint32_t qid = _rpc->query(target, false, HYPERDHT_CMD_MUTABLE_GET,
seq_buf.data(), seq_buf.size(), 0, false);
if (_query_info_count < HYPERDHT_MAX_QUERIES) {
auto &qi = _query_infos[_query_info_count++];
qi.query_id = qid;
memcpy(qi.topic, public_key, 32); // Store PK for signature verification.
memcpy(qi.storage_key, target, 32);
qi.is_announce = false;
qi.is_lookup = false;
qi.is_find_peer = false;
qi.is_immutable_get = false;
qi.is_immutable_put = false;
qi.is_mutable_get = true;
qi.is_mutable_put = false;
qi.mutable_seq = min_seq;
qi.put_value.clear();
memset(qi.put_signature, 0, 64);
qi.active = true;
}
}
// ---- Handle incoming storage requests (from other DHT nodes) ----
// Matches JS persistent.js: onimmutableput, onimmutableget, onmutableput, onmutableget.
void HyperDht::_handle_storage_request(const DhtRequest &req) {
switch (req.command) {
case HYPERDHT_CMD_IMMUTABLE_PUT: {
// Matches JS persistent.js onimmutableput: requires target, token, and value.
if (!req.has_target || !req.value || req.value_len == 0) {
return; // No reply — matches JS behavior (silent return).
}
if (!req.has_token || !_rpc->validate_token(req.from, req.token)) {
return; // No reply — token required for commit phase.
}
bool ok = _storage.handle_immutable_put(req.target, req.value, req.value_len);
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, ok ? 0 : 1, nullptr, 0);
break;
}
case HYPERDHT_CMD_IMMUTABLE_GET: {
if (!req.has_target) {
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 1, nullptr, 0);
return;
}
std::vector<uint8_t> value;
if (_storage.handle_immutable_get(req.target, value)) {
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 0,
value.data(), value.size());
} else {
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 0, nullptr, 0);
}
break;
}
case HYPERDHT_CMD_MUTABLE_PUT: {
// Matches JS persistent.js onmutableput: requires target, token, and value.
if (!req.has_target || !req.value || req.value_len == 0) {
return; // No reply — matches JS behavior.
}
if (!req.has_token || !_rpc->validate_token(req.from, req.token)) {
return; // No reply — token required for commit phase.
}
// Decode MutablePutRequest.
EncodeState s = { 0, req.value_len, const_cast<uint8_t *>(req.value) };
MutablePutRequest mpr;
if (!dht_msg::mutable_put_decode(s, mpr)) {
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 1, nullptr, 0);
return;
}
// Verify target == BLAKE2b(publicKey) — matches JS persistent.js check.
uint8_t pk_hash[32];
crypto_generichash(pk_hash, 32, mpr.public_key, 32, nullptr, 0);
if (memcmp(pk_hash, req.target, 32) != 0) {
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 1, nullptr, 0);
return;
}
uint32_t err = _storage.put_mutable(mpr.public_key, mpr.seq,
mpr.value, mpr.value_len,
mpr.signature, _ns_mutable_put);
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, err, nullptr, 0);
break;
}
case HYPERDHT_CMD_MUTABLE_GET: {
if (!req.has_target) {
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 1, nullptr, 0);
return;
}
// Decode min_seq from request value (matches JS: c.decode(c.uint, req.value)).
uint64_t min_seq = 0;
if (req.value && req.value_len > 0) {
EncodeState vs = { 0, req.value_len, const_cast<uint8_t *>(req.value) };
min_seq = cenc::uint_decode(vs);
}
MutableGetResponse resp;
if (_storage.handle_mutable_get(req.target, min_seq, resp)) {
// Encode MutableGetResponse.
EncodeState ps = { 0, 0, nullptr };
dht_msg::mutable_get_response_preencode(ps, resp);
std::vector<uint8_t> encoded(ps.end);
ps.buffer = encoded.data();
ps.start = 0;
dht_msg::mutable_get_response_encode(ps, resp);
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 0,
encoded.data(), encoded.size());
} else {
_rpc->_send_response(req.tid, req.from, true, nullptr,
nullptr, 0, 0, nullptr, 0);
}
break;
}
}
}
// ---- Handle storage responses (immutable/mutable GET results) ----
void HyperDht::_handle_storage_response(const DhtResponse &res, uint32_t tag) {
if (!res.value || res.value_len == 0) return;
for (int i = 0; i < _query_info_count; i++) {
auto &qi = _query_infos[i];
if (!qi.active) continue;
if (qi.query_id != tag) continue; // Route to the specific query.
if (qi.is_immutable_get && qi.put_value.empty()) {
// Verify BLAKE2b(value) == storage_key.
uint8_t hash[32];
crypto_generichash(hash, 32, res.value, res.value_len, nullptr, 0);
if (memcmp(hash, qi.storage_key, 32) == 0) {
_storage.handle_immutable_put(qi.storage_key, res.value, res.value_len);
qi.put_value.assign(res.value, res.value + res.value_len);
if (on_immutable_get_complete) {
on_immutable_get_complete(qi.storage_key, res.value, res.value_len);
}
}
} else if (qi.is_mutable_get) {
// Decode MutableGetResponse, verify signature, track best seq.
EncodeState ms = { 0, res.value_len, const_cast<uint8_t *>(res.value) };
MutableGetResponse mgr;
if (dht_msg::mutable_get_response_decode(ms, mgr) &&
mgr.seq >= qi.mutable_seq) {
// qi.topic holds the original public key for verification.
if (DhtStorage::verify_mutable(mgr.signature, mgr.seq,
mgr.value.data(), mgr.value.size(),
qi.topic, _ns_mutable_put)) {
_storage.put_mutable(qi.topic, mgr.seq,
mgr.value.data(), mgr.value.size(),
mgr.signature, _ns_mutable_put);
qi.mutable_seq = mgr.seq;
qi.put_value = mgr.value;
memcpy(qi.put_signature, mgr.signature, 64);
}
}
}
}
}