#include "dht_rpc.h"
#include <sodium.h>
#include "debug_log.h"
#include <cstdlib>
#include <algorithm>
// ---- Static helpers ----
static void sockaddr_to_ipv4(const struct sockaddr_in *sa, IPv4Address &out) {
uint32_t ip = sa->sin_addr.s_addr; // Network byte order.
out.ip[0] = static_cast<uint8_t>(ip & 0xff);
out.ip[1] = static_cast<uint8_t>((ip >> 8) & 0xff);
out.ip[2] = static_cast<uint8_t>((ip >> 16) & 0xff);
out.ip[3] = static_cast<uint8_t>((ip >> 24) & 0xff);
out.port = ntohs(sa->sin_port);
}
static void ipv4_to_sockaddr(const IPv4Address &addr, struct sockaddr_in &sa) {
memset(&sa, 0, sizeof(sa));
sa.sin_family = AF_INET;
sa.sin_port = htons(addr.port);
sa.sin_addr.s_addr = static_cast<uint32_t>(addr.ip[0])
| (static_cast<uint32_t>(addr.ip[1]) << 8)
| (static_cast<uint32_t>(addr.ip[2]) << 16)
| (static_cast<uint32_t>(addr.ip[3]) << 24);
}
static bool ipv4_equals(const IPv4Address &a, const IPv4Address &b) {
return memcmp(a.ip, b.ip, 4) == 0 && a.port == b.port;
}
static bool node_id_equals(const NodeId &a, const NodeId &b) {
return memcmp(a.bytes, b.bytes, 32) == 0;
}
static bool node_id_is_zero(const NodeId &id) {
for (int i = 0; i < 32; i++) {
if (id.bytes[i] != 0) return false;
}
return true;
}
// XOR distance compare for query closest list.
static int xor_distance_compare(const uint8_t *target, const uint8_t *a, const uint8_t *b) {
for (int i = 0; i < 32; i++) {
uint8_t da = target[i] ^ a[i];
uint8_t db = target[i] ^ b[i];
if (da < db) return -1;
if (da > db) return 1;
}
return 0;
}
// UDP send completion callback.
static void on_udp_send_complete(udx_socket_send_t *req, int status) {
auto *send_req = reinterpret_cast<DhtRpc::UdpSendReq *>(req);
(void)status;
delete send_req;
}
// ---- DhtRpc implementation ----
DhtRpc::DhtRpc()
: _socket(nullptr), _loop(nullptr), _has_local_addr(false),
_iface_addr_count(0),
_bootstrap_count(0), _bootstrapped(false), _bootstrapping(false),
_bootstrap_pending(0), _inflight_count(0), _tid_counter(0),
_query_count(0), _query_id_counter(0),
_secrets_initialized(false), _secret_rotate_counter(DHT_SECRET_ROTATE_INTERVAL),
_last_tick_ms(0) {
if (sodium_init() < 0) {
DHT_LOG("DhtRpc: sodium_init failed\n");
}
memset(&_local_addr, 0, sizeof(_local_addr));
memset(_iface_addrs, 0, sizeof(_iface_addrs));
memset(_bootstrap_nodes, 0, sizeof(_bootstrap_nodes));
memset(_inflight, 0, sizeof(_inflight));
memset(_queries, 0, sizeof(_queries));
// Random starting TID (matches JS: Math.random() * 65536 | 0).
_tid_counter = static_cast<uint16_t>(randombytes_random() & 0xffff);
}
DhtRpc::~DhtRpc() {
// Free any active queries.
for (int i = 0; i < _query_count; i++) {
delete _queries[i];
}
// Free inflight buffers.
for (int i = 0; i < _inflight_count; i++) {
delete[] _inflight[i].buffer;
}
sodium_memzero(_secrets, sizeof(_secrets));
}
void DhtRpc::init(udx_socket_t *socket, uv_loop_t *loop) {
_socket = socket;
_loop = loop;
}
void DhtRpc::set_bootstrap(const char **nodes, int count) {
_bootstrap_count = 0;
for (int i = 0; i < count && i < DHT_MAX_BOOTSTRAP_NODES; i++) {
IPv4Address addr;
if (cenc::ipv4_address_parse(nodes[i], addr)) {
_bootstrap_nodes[_bootstrap_count].address = addr;
_bootstrap_nodes[_bootstrap_count].resolved = true;
_bootstrap_count++;
}
}
}
void DhtRpc::set_local_address(const IPv4Address &addr) {
_local_addr = addr;
_has_local_addr = true;
// Compute our node ID from our address.
NodeId id = cenc::node_id_from_address(addr);
_table.init(id);
}
void DhtRpc::bootstrap() {
if (_bootstrapping) return;
_bootstrapping = true;
_bootstrap_pending = 0;
// Send FIND_NODE(our_id) to each bootstrap node.
for (int i = 0; i < _bootstrap_count; i++) {
find_node(_table.id().bytes, _bootstrap_nodes[i].address, 0xFFFFFFFF);
_bootstrap_pending++;
}
}
uint16_t DhtRpc::_next_tid() {
uint16_t tid = _tid_counter++;
if (_tid_counter == 0) _tid_counter = 1; // Avoid TID 0 after wrap.
return tid;
}
void DhtRpc::ping(const IPv4Address &to, uint32_t tag) {
uint16_t tid = _next_tid();
_alloc_inflight(tid, to, true, DHT_CMD_PING, nullptr, tag);
_send_request(tid, to, nullptr, true, DHT_CMD_PING, nullptr, nullptr, 0);
}
void DhtRpc::find_node(const uint8_t *target, const IPv4Address &to, uint32_t tag) {
uint16_t tid = _next_tid();
_alloc_inflight(tid, to, true, DHT_CMD_FIND_NODE, target, tag);
_send_request(tid, to, nullptr, true, DHT_CMD_FIND_NODE, target, nullptr, 0);
}
uint32_t DhtRpc::query(const uint8_t *target, bool internal, uint32_t command,
const uint8_t *value, size_t value_len,
uint32_t tag, bool commit) {
if (_query_count >= 16) return 0; // Too many concurrent queries.
auto *q = new DhtQuery{}; // Value-init: zeroes POD, default-constructs C++ members.
q->id = ++_query_id_counter;
memcpy(q->target, target, 32);
q->internal = internal;
q->command = command;
q->value = value;
q->value_len = value_len;
q->tag = q->id; // Route responses back via query ID (matches JS per-query callback model).
q->commit = commit;
// Seed from routing table.
KademliaNode nodes[DHT_QUERY_K];
int n = _table.closest(target, nodes, DHT_QUERY_K);
for (int i = 0; i < n; i++) {
_add_query_peer(q, nodes[i].id, nodes[i].address);
}
// If not enough from table, add bootstrap nodes.
if (q->closest_count < DHT_QUERY_K) {
for (int i = 0; i < _bootstrap_count && q->closest_count < DHT_QUERY_K; i++) {
NodeId bid = cenc::node_id_from_address(_bootstrap_nodes[i].address);
_add_query_peer(q, bid, _bootstrap_nodes[i].address);
}
}
_queries[_query_count++] = q;
// _advance_query may COMPLETE — and delete — a query with no one to ask (an empty
// routing table: an announce at startup). Take the id first.
const uint32_t id = q->id;
_advance_query(q);
return id;
}
uint32_t DhtRpc::query_with_commit(const uint8_t *target, bool internal,
uint32_t traverse_command,
const uint8_t *traverse_value, size_t traverse_value_len,
uint32_t commit_command,
const uint8_t *commit_value, size_t commit_value_len,
uint32_t tag) {
if (_query_count >= 16) return 0;
auto *q = new DhtQuery{}; // Value-init: zeroes POD, default-constructs C++ members.
q->id = ++_query_id_counter;
memcpy(q->target, target, 32);
q->internal = internal;
q->command = traverse_command;
q->value = traverse_value;
q->value_len = traverse_value_len;
q->tag = q->id; // Route responses back via query ID.
q->commit = true;
// Set commit-phase overrides.
q->has_commit_override = true;
q->commit_command = commit_command;
q->commit_value_buf.assign(commit_value, commit_value + commit_value_len);
// Seed from routing table.
KademliaNode nodes[DHT_QUERY_K];
int n = _table.closest(target, nodes, DHT_QUERY_K);
for (int i = 0; i < n; i++) {
_add_query_peer(q, nodes[i].id, nodes[i].address);
}
if (q->closest_count < DHT_QUERY_K) {
for (int i = 0; i < _bootstrap_count && q->closest_count < DHT_QUERY_K; i++) {
NodeId bid = cenc::node_id_from_address(_bootstrap_nodes[i].address);
_add_query_peer(q, bid, _bootstrap_nodes[i].address);
}
}
_queries[_query_count++] = q;
// _advance_query may COMPLETE — and delete — a query with no one to ask (an empty
// routing table: an announce at startup). Take the id first.
const uint32_t id = q->id;
_advance_query(q);
return id;
}
uint32_t DhtRpc::query_with_commit_fn(
const uint8_t *target, bool internal,
uint32_t traverse_command,
const uint8_t *traverse_value, size_t traverse_value_len,
uint32_t commit_command,
std::function<std::vector<uint8_t>(const uint8_t *, const uint8_t *,
const uint8_t *, const IPv4Address &)> commit_fn,
uint32_t tag) {
if (_query_count >= 16) return 0;
auto *q = new DhtQuery{}; // Value-init: zeroes POD, default-constructs C++ members.
q->id = ++_query_id_counter;
memcpy(q->target, target, 32);
q->internal = internal;
q->command = traverse_command;
q->value = traverse_value;
q->value_len = traverse_value_len;
q->tag = q->id; // Route responses back via query ID.
q->commit = true;
// Set commit-phase overrides with per-node callback.
q->has_commit_override = true;
q->commit_command = commit_command;
q->commit_value_fn = std::move(commit_fn);
// Seed from routing table.
KademliaNode nodes[DHT_QUERY_K];
int n = _table.closest(target, nodes, DHT_QUERY_K);
for (int i = 0; i < n; i++) {
_add_query_peer(q, nodes[i].id, nodes[i].address);
}
if (q->closest_count < DHT_QUERY_K) {
for (int i = 0; i < _bootstrap_count && q->closest_count < DHT_QUERY_K; i++) {
NodeId bid = cenc::node_id_from_address(_bootstrap_nodes[i].address);
_add_query_peer(q, bid, _bootstrap_nodes[i].address);
}
}
_queries[_query_count++] = q;
// _advance_query may COMPLETE — and delete — a query with no one to ask (an empty
// routing table: an announce at startup). Take the id first.
const uint32_t id = q->id;
_advance_query(q);
return id;
}
void DhtRpc::on_message(const uint8_t *data, size_t len, const struct sockaddr_in *from) {
if (len < 2) return;
char addr_buf[32];
snprintf(addr_buf, sizeof(addr_buf), "%d.%d.%d.%d:%d",
((uint8_t*)&from->sin_addr.s_addr)[0], ((uint8_t*)&from->sin_addr.s_addr)[1],
((uint8_t*)&from->sin_addr.s_addr)[2], ((uint8_t*)&from->sin_addr.s_addr)[3],
ntohs(from->sin_port));
if (data[0] == DHT_RESPONSE_ID) {
DhtResponse res;
if (_decode_response(data, len, from, res)) {
_handle_response(res);
}
} else if (data[0] == DHT_REQUEST_ID) {
DhtRequest req;
if (_decode_request(data, len, from, req)) {
_handle_request(req);
}
}
}
void DhtRpc::tick(uint64_t now_ms) {
_last_tick_ms = now_ms;
// Process inflight timeouts.
for (int i = 0; i < _inflight_count; ) {
DhtInflight &inf = _inflight[i];
if (!inf.active) { i++; continue; }
uint64_t elapsed = now_ms - inf.sent_at_ms;
if (elapsed >= DHT_INFLIGHT_TIMEOUT_MS) {
if (inf.retries_left > 0) {
// Retry.
inf.retries_left--;
inf.sent_count++;
inf.sent_at_ms = now_ms;
_udp_send(inf.buffer, inf.buffer_len, inf.to);
} else {
// Timed out — update associated query/bootstrap counters
// before freeing the inflight slot.
uint32_t inf_tag = inf.tag;
IPv4Address inf_to = inf.to;
// Handle bootstrap ping timeout.
if (inf_tag == 0xFFFFFFFF) {
_bootstrap_pending--;
if (_bootstrap_pending <= 0 && !_bootstrapped) {
_bootstrapped = true;
query(_table.id().bytes, true, DHT_CMD_FIND_NODE,
nullptr, 0, 0xFFFFFFFE, false);
if (on_bootstrapped) on_bootstrapped();
}
}
// Update query counters for timed-out requests.
for (int qi = 0; qi < _query_count; qi++) {
DhtQuery *q = _queries[qi];
if (!q || q->complete) continue;
// Commit phase: tag == query id.
if (q->committing && inf_tag == q->id) {
q->commit_inflight--;
q->errors++;
if (q->commit_inflight <= 0) {
q->committing = false;
q->complete = true;
if (on_query_complete) on_query_complete(q->id);
_queries[qi] = _queries[_query_count - 1];
_queries[_query_count - 1] = nullptr;
_query_count--;
delete q;
}
break;
}
// Traversal phase: match by address (same logic as
// _handle_response traversal matching).
if (!q->committing) {
bool matched = false;
for (int j = 0; j < q->closest_count; j++) {
if (ipv4_equals(q->closest[j].address, inf_to) &&
q->closest[j].queried && !q->closest[j].responded) {
q->closest[j].responded = true;
q->inflight--;
q->errors++;
matched = true;
break;
}
}
if (matched) break;
}
}
_free_inflight(&inf);
}
}
i++;
}
// Advance active queries.
for (int i = 0; i < _query_count; i++) {
if (_queries[i] && !_queries[i]->complete) {
_advance_query(_queries[i]);
}
}
// Secret rotation (every DHT_SECRET_ROTATE_INTERVAL ticks).
// The JS rotates every 10 drain cycles (750ms each), so ~7.5s.
// We tick at 1ms, so use a counter.
// Actually, we approximate: rotate every ~7500ms.
static uint64_t last_rotate = 0;
if (now_ms - last_rotate >= 7500) {
last_rotate = now_ms;
_rotate_secrets();
}
}
// ---- Message encoding ----
void DhtRpc::_send_request(uint16_t tid, const IPv4Address &to, const uint8_t *token,
bool internal, uint32_t command, const uint8_t *target,
const uint8_t *value, size_t value_len) {
bool has_id = _has_local_addr;
// Preencode to compute size.
// Layout: type(1) + flags(1) + tid(2) + to_addr(6)
// + [id(32)] + [token(32)] + command(varint) + [target(32)] + [value(buffer)]
EncodeState s = { 0, 1 + 1 + 2 + 6, nullptr };
if (has_id) s.end += 32;
if (token) s.end += 32;
cenc::uint_preencode(s, command);
if (target) s.end += 32;
if (value && value_len > 0) cenc::buffer_preencode(s, value, value_len);
if (s.end > 1200) return; // Message too large.
// Allocate and encode.
uint8_t buf[1200];
s.buffer = buf;
s.start = 0;
// Byte 0: REQUEST_ID.
s.buffer[s.start++] = DHT_REQUEST_ID;
// Byte 1: Flags.
uint8_t flags = 0;
if (has_id) flags |= DHT_FLAG_HAS_ID;
if (token) flags |= DHT_FLAG_TOKEN;
if (internal) flags |= DHT_FLAG_INTERNAL;
if (target) flags |= DHT_FLAG_TARGET;
if (value && value_len > 0) flags |= DHT_FLAG_VALUE;
s.buffer[s.start++] = flags;
// TID.
cenc::uint16_encode(s, tid);
// To address (6 bytes).
cenc::ipv4_address_encode(s, to);
// Optional ID (our node ID).
if (has_id) cenc::fixed32_encode(s, _table.id().bytes);
// Optional token.
if (token) cenc::fixed32_encode(s, token);
// Command (varint).
cenc::uint_encode(s, command);
// Optional target.
if (target) cenc::fixed32_encode(s, target);
// Optional value.
if (value && value_len > 0) cenc::buffer_encode(s, value, value_len);
size_t msg_len = s.start;
// Cache in inflight for retries.
DhtInflight *inf = _find_inflight(tid);
if (inf) {
inf->buffer = new uint8_t[msg_len];
memcpy(inf->buffer, buf, msg_len);
inf->buffer_len = msg_len;
inf->sent_at_ms = _last_tick_ms ? _last_tick_ms : uv_now(_loop);
}
_udp_send(buf, msg_len, to);
}
void DhtRpc::_send_response(uint16_t tid, const IPv4Address &to, bool include_id,
const uint8_t *token, const IPv4Address *closer_nodes,
int closer_count, uint32_t error,
const uint8_t *value, size_t value_len) {
EncodeState s = { 0, 1 + 1 + 2 + 6, nullptr };
if (include_id) s.end += 32;
if (token) s.end += 32;
if (closer_count > 0) cenc::ipv4_array_preencode(s, closer_nodes, closer_count);
if (error > 0) cenc::uint_preencode(s, error);
if (value && value_len > 0) cenc::buffer_preencode(s, value, value_len);
if (s.end > 1200) return;
uint8_t buf[1200];
s.buffer = buf;
s.start = 0;
s.buffer[s.start++] = DHT_RESPONSE_ID;
uint8_t flags = 0;
if (include_id) flags |= DHT_FLAG_HAS_ID;
if (token) flags |= DHT_FLAG_TOKEN;
if (closer_count > 0) flags |= DHT_FLAG_CLOSER_NODES;
if (error > 0) flags |= DHT_FLAG_ERROR;
if (value && value_len > 0) flags |= DHT_FLAG_VALUE;
s.buffer[s.start++] = flags;
cenc::uint16_encode(s, tid);
cenc::ipv4_address_encode(s, to);
if (include_id) cenc::fixed32_encode(s, _table.id().bytes);
if (token) cenc::fixed32_encode(s, token);
if (closer_count > 0) cenc::ipv4_array_encode(s, closer_nodes, closer_count);
if (error > 0) cenc::uint_encode(s, error);
if (value && value_len > 0) cenc::buffer_encode(s, value, value_len);
_udp_send(buf, s.start, to);
}
// ---- Message decoding ----
bool DhtRpc::_decode_response(const uint8_t *data, size_t len,
const struct sockaddr_in *from, DhtResponse &out) {
memset(&out, 0, sizeof(out));
sockaddr_to_ipv4(from, out.from);
EncodeState s = { 1, len, const_cast<uint8_t *>(data) };
uint8_t flags = static_cast<uint8_t>(cenc::uint_decode(s));
out.tid = cenc::uint16_decode(s);
out.to = cenc::ipv4_address_decode(s);
if (flags & DHT_FLAG_HAS_ID) {
const uint8_t *id_bytes = cenc::fixed32_decode(s);
if (!id_bytes) return false;
memcpy(out.from_id.bytes, id_bytes, 32);
// Validate: ID must match BLAKE2b(from_address).
NodeId expected = cenc::node_id_from_address(out.from);
out.has_id = node_id_equals(out.from_id, expected);
if (!out.has_id) {
memset(out.from_id.bytes, 0, 32);
}
}
if (flags & DHT_FLAG_TOKEN) {
const uint8_t *tok = cenc::fixed32_decode(s);
if (!tok) return false;
memcpy(out.token, tok, 32);
out.has_token = true;
}
if (flags & DHT_FLAG_CLOSER_NODES) {
out.closer_count = cenc::ipv4_array_decode(s, out.closer_nodes, DHT_MAX_CLOSER_NODES);
}
if (flags & DHT_FLAG_ERROR) {
out.error = static_cast<uint32_t>(cenc::uint_decode(s));
}
if (flags & DHT_FLAG_VALUE) {
BufView bv = cenc::buffer_decode(s);
out.value = bv.data;
out.value_len = bv.len;
}
return true;
}
bool DhtRpc::_decode_request(const uint8_t *data, size_t len,
const struct sockaddr_in *from, DhtRequest &out) {
memset(&out, 0, sizeof(out));
sockaddr_to_ipv4(from, out.from);
EncodeState s = { 1, len, const_cast<uint8_t *>(data) };
uint8_t flags = static_cast<uint8_t>(cenc::uint_decode(s));
out.tid = cenc::uint16_decode(s);
out.to = cenc::ipv4_address_decode(s);
if (flags & DHT_FLAG_HAS_ID) {
const uint8_t *id_bytes = cenc::fixed32_decode(s);
if (!id_bytes) return false;
memcpy(out.from_id.bytes, id_bytes, 32);
NodeId expected = cenc::node_id_from_address(out.from);
out.has_id = node_id_equals(out.from_id, expected);
if (!out.has_id) {
memset(out.from_id.bytes, 0, 32);
}
}
if (flags & DHT_FLAG_TOKEN) {
const uint8_t *tok = cenc::fixed32_decode(s);
if (!tok) return false;
memcpy(out.token, tok, 32);
out.has_token = true;
}
out.internal = (flags & DHT_FLAG_INTERNAL) != 0;
out.command = static_cast<uint32_t>(cenc::uint_decode(s));
if (flags & DHT_FLAG_TARGET) {
const uint8_t *t = cenc::fixed32_decode(s);
if (!t) return false;
memcpy(out.target, t, 32);
out.has_target = true;
}
if (flags & DHT_FLAG_VALUE) {
BufView bv = cenc::buffer_decode(s);
out.value = bv.data;
out.value_len = bv.len;
}
return true;
}
// ---- Response handling ----
void DhtRpc::_handle_response(const DhtResponse &res) {
// Match to inflight request.
DhtInflight *inf = _find_inflight(res.tid);
if (!inf) return;
uint32_t rtt_ms = 0;
if (_last_tick_ms >= inf->sent_at_ms) {
rtt_ms = static_cast<uint32_t>(_last_tick_ms - inf->sent_at_ms);
}
uint32_t tag = inf->tag;
uint32_t command = inf->command;
bool internal = inf->internal;
IPv4Address to = inf->to;
// Free inflight slot.
_free_inflight(inf);
// Feed NAT sampler with the response's "to" field (how remote sees us).
if (res.to.port != 0) {
_nat_sampler.add(res.to, res.from);
}
// Update routing table with responding node.
if (res.has_id) {
KademliaNode kn;
kn.id = res.from_id;
kn.address = res.from;
kn.added_at = _last_tick_ms;
_table.add(kn);
}
// Add closer nodes to routing table.
for (int i = 0; i < res.closer_count; i++) {
NodeId nid = cenc::node_id_from_address(res.closer_nodes[i]);
// Don't add ourselves.
if (node_id_equals(nid, _table.id())) continue;
KademliaNode kn;
kn.id = nid;
kn.address = res.closer_nodes[i];
kn.added_at = _last_tick_ms;
_table.add(kn);
}
// Bootstrap tracking.
if (tag == 0xFFFFFFFF) {
_bootstrap_pending--;
if (_bootstrap_pending <= 0 && !_bootstrapped) {
// Initial bootstrap round complete. Now run a full iterative
// FIND_NODE(self) to populate the routing table properly.
// Matches JS dht-rpc bootstrap: full iterative query after initial round.
_bootstrapped = true;
// Run iterative FIND_NODE query to discover more nodes.
// Use internal=true, command=FIND_NODE, no commit.
query(_table.id().bytes, true, DHT_CMD_FIND_NODE,
nullptr, 0, 0xFFFFFFFE, false);
if (on_bootstrapped) on_bootstrapped();
}
}
// Route to iterative query if applicable.
bool routed = false;
for (int i = 0; i < _query_count; i++) {
DhtQuery *q = _queries[i];
if (!q || q->complete) continue;
// Handle commit-phase responses (matched by tag = query id).
if (q->committing && tag == q->id) {
q->commit_inflight--;
q->commit_responses++;
if (q->commit_inflight <= 0) {
// All commit requests completed — truly finish the query.
q->committing = false;
q->complete = true;
if (on_query_complete) on_query_complete(q->id);
// Remove from active queries.
_queries[i] = _queries[_query_count - 1];
_queries[_query_count - 1] = nullptr;
_query_count--;
delete q;
}
routed = true;
break;
}
// Check if this response belongs to this query (traversal phase).
if (q->committing) continue; // Skip traversal matching during commit.
for (int j = 0; j < q->closest_count; j++) {
if (ipv4_equals(q->closest[j].address, res.from) && q->closest[j].queried && !q->closest[j].responded) {
q->closest[j].responded = true;
q->inflight--;
if (res.error == 0) {
q->successes++;
// Add to responded list if closer than current worst.
if (res.has_id) {
if (q->responded_count < DHT_QUERY_K) {
auto &rp = q->responded[q->responded_count];
rp.id = res.from_id;
rp.address = res.from;
rp.responded = true;
if (res.has_token) {
memcpy(rp.token, res.token, 32);
rp.has_token = true;
} else {
rp.has_token = false;
}
q->responded_count++;
} else {
int cmp = xor_distance_compare(q->target, res.from_id.bytes,
q->responded[q->responded_count - 1].id.bytes);
if (cmp < 0) {
auto &rp = q->responded[q->responded_count - 1];
rp.id = res.from_id;
rp.address = res.from;
if (res.has_token) {
memcpy(rp.token, res.token, 32);
rp.has_token = true;
} else {
rp.has_token = false;
}
}
}
// Sort responded by distance.
for (int a = q->responded_count - 1; a > 0; a--) {
if (xor_distance_compare(q->target,
q->responded[a - 1].id.bytes,
q->responded[a].id.bytes) <= 0) break;
auto tmp = q->responded[a];
q->responded[a] = q->responded[a - 1];
q->responded[a - 1] = tmp;
}
}
} else {
q->errors++;
}
// Add closer nodes to query.
for (int k = 0; k < res.closer_count; k++) {
NodeId nid = cenc::node_id_from_address(res.closer_nodes[k]);
if (node_id_equals(nid, _table.id())) continue;
_add_query_peer(q, nid, res.closer_nodes[k]);
}
routed = true;
break;
}
}
if (routed) break; // One response → one query match. Prevents cross-query contamination.
}
// Fire callback.
DhtResponse res_copy = res;
res_copy.rtt_ms = rtt_ms;
if (on_response) on_response(res_copy, tag);
}
// ---- Request handling ----
void DhtRpc::_handle_request(const DhtRequest &req) {
if (!req.internal) {
// Non-internal: forward to user callback.
if (on_request) on_request(req);
return;
}
// Handle internal commands.
switch (req.command) {
case DHT_CMD_PING: {
// Respond with closer nodes (if target present).
IPv4Address closer[DHT_MAX_CLOSER_NODES];
int closer_count = 0;
if (req.has_target) {
closer_count = _table.closest_addresses(req.target, closer, DHT_MAX_CLOSER_NODES);
}
uint8_t token[32];
generate_token(req.from, token);
_send_response(req.tid, req.from, _has_local_addr, token,
closer, closer_count, 0, nullptr, 0);
break;
}
case DHT_CMD_FIND_NODE: {
// Return closest nodes to the target.
// FIND_NODE is a routing-only command — no token needed.
// Tokens are only for queries that will later commit (announce, etc.).
IPv4Address closer[DHT_MAX_CLOSER_NODES];
int closer_count = 0;
if (req.has_target) {
closer_count = _table.closest_addresses(req.target, closer, DHT_MAX_CLOSER_NODES);
}
_send_response(req.tid, req.from, _has_local_addr, nullptr,
closer, closer_count, 0, nullptr, 0);
// Add requesting node to our table.
if (req.has_id) {
KademliaNode kn;
kn.id = req.from_id;
kn.address = req.from;
kn.added_at = _last_tick_ms;
_table.add(kn);
}
break;
}
default:
// Unknown internal command — ignore.
break;
}
}
// ---- UDP send ----
void DhtRpc::_udp_send(const uint8_t *data, size_t len, const IPv4Address &to) {
if (!_socket) return;
auto *send_req = new UdpSendReq();
memcpy(send_req->data, data, len);
send_req->buf = uv_buf_init(reinterpret_cast<char *>(send_req->data), static_cast<unsigned int>(len));
struct sockaddr_in dest;
ipv4_to_sockaddr(to, dest);
int rc = udx_socket_send(
&send_req->req, _socket, &send_req->buf, 1,
reinterpret_cast<const struct sockaddr *>(&dest),
on_udp_send_complete);
if (rc < 0) {
delete send_req;
}
}
// ---- Inflight management ----
DhtInflight *DhtRpc::_alloc_inflight(uint16_t tid, const IPv4Address &to,
bool internal, uint32_t command,
const uint8_t *target, uint32_t tag) {
if (_inflight_count >= DHT_MAX_INFLIGHT) return nullptr;
DhtInflight &inf = _inflight[_inflight_count++];
memset(&inf, 0, sizeof(inf));
inf.tid = tid;
inf.to = to;
inf.internal = internal;
inf.command = command;
if (target) {
memcpy(inf.target, target, 32);
inf.has_target = true;
}
inf.retries_left = 5;
inf.sent_count = 1;
inf.tag = tag;
inf.active = true;
inf.buffer = nullptr;
inf.buffer_len = 0;
inf.sent_at_ms = _last_tick_ms ? _last_tick_ms : 1;
return &inf;
}
DhtInflight *DhtRpc::_find_inflight(uint16_t tid) {
for (int i = 0; i < _inflight_count; i++) {
if (_inflight[i].active && _inflight[i].tid == tid) {
return &_inflight[i];
}
}
return nullptr;
}
void DhtRpc::_free_inflight(DhtInflight *inf) {
delete[] inf->buffer;
inf->buffer = nullptr;
inf->active = false;
// Compact: swap with last active.
int idx = static_cast<int>(inf - _inflight);
if (idx < _inflight_count - 1) {
_inflight[idx] = _inflight[_inflight_count - 1];
}
_inflight_count--;
}
// ---- Iterative query ----
void DhtRpc::_add_query_peer(DhtQuery *q, const NodeId &id, const IPv4Address &addr) {
// Check if already in the working set.
for (int i = 0; i < q->closest_count; i++) {
if (ipv4_equals(q->closest[i].address, addr)) return;
}
// Check if closer than the worst in the working set.
if (q->closest_count >= DHT_QUERY_K * 3) {
// Working set full — only add if closer than worst unqueried.
// Simple: just don't add.
DHT_LOG("[DhtRpc] _add_query_peer: qid=%u FULL, not adding %d.%d.%d.%d:%d\n",
q->id, addr.ip[0], addr.ip[1], addr.ip[2], addr.ip[3], addr.port);
return;
}
DHT_LOG("[DhtRpc] _add_query_peer: qid=%u adding %d.%d.%d.%d:%d (count=%d)\n",
q->id, addr.ip[0], addr.ip[1], addr.ip[2], addr.ip[3], addr.port,
q->closest_count + 1);
DhtQueryPeer &p = q->closest[q->closest_count++];
p.id = id;
p.address = addr;
p.queried = false;
p.responded = false;
}
void DhtRpc::_advance_query(DhtQuery *q) {
if (q->complete) return;
int concurrency = DHT_QUERY_CONCURRENCY;
// Send requests to unqueried peers (closest first).
while (q->inflight < concurrency) {
// Find the closest unqueried peer.
int best = -1;
for (int i = 0; i < q->closest_count; i++) {
if (q->closest[i].queried) continue;
if (best < 0 || xor_distance_compare(q->target,
q->closest[i].id.bytes, q->closest[best].id.bytes) < 0) {
best = i;
}
}
if (best < 0) break; // No more unqueried peers.
// Check if this peer is still "closer" than the worst responded.
if (q->responded_count >= DHT_QUERY_K) {
int cmp = xor_distance_compare(q->target,
q->closest[best].id.bytes,
q->responded[q->responded_count - 1].id.bytes);
if (cmp >= 0) {
DHT_LOG("[DhtRpc] _advance_query: qid=%u SKIP %d.%d.%d.%d:%d (not closer)\n",
q->id,
q->closest[best].address.ip[0], q->closest[best].address.ip[1],
q->closest[best].address.ip[2], q->closest[best].address.ip[3],
q->closest[best].address.port);
q->closest[best].queried = true; // Mark as skipped.
continue;
}
}
q->closest[best].queried = true;
q->inflight++;
uint16_t tid = _next_tid();
_alloc_inflight(tid, q->closest[best].address,
q->internal, q->command,
q->target, q->tag);
_send_request(tid, q->closest[best].address, nullptr,
q->internal, q->command, q->target,
q->value, q->value_len);
}
// Check if query is complete (no inflight and no unqueried peers).
if (q->inflight == 0) {
bool has_unqueried = false;
for (int i = 0; i < q->closest_count; i++) {
if (!q->closest[i].queried) {
// Only count if it's closer than our worst responded.
if (q->responded_count < DHT_QUERY_K) {
has_unqueried = true;
break;
}
int cmp = xor_distance_compare(q->target,
q->closest[i].id.bytes,
q->responded[q->responded_count - 1].id.bytes);
if (cmp < 0) {
has_unqueried = true;
break;
}
}
}
if (!has_unqueried) {
_complete_query(q);
}
}
}
void DhtRpc::_complete_query(DhtQuery *q) {
// If this query has commit=true and we haven't committed yet, enter commit phase.
if (q->commit && !q->committing && q->responded_count > 0) {
DHT_DISC("[DhtRpc] _complete_query: entering commit phase qid=%u responded=%d\n",
q->id, q->responded_count);
_commit_query(q);
return;
}
DHT_DISC("[DhtRpc] _complete_query: finishing qid=%u commit=%d committing=%d responded=%d\n",
q->id, q->commit, q->committing, q->responded_count);
q->complete = true;
if (on_query_complete) on_query_complete(q->id);
// Remove from active queries.
for (int i = 0; i < _query_count; i++) {
if (_queries[i] == q) {
_queries[i] = _queries[_query_count - 1];
_queries[_query_count - 1] = nullptr;
_query_count--;
delete q;
return;
}
}
}
void DhtRpc::_commit_query(DhtQuery *q) {
// Commit phase: send command WITH token to each closest responded node.
q->committing = true;
q->commit_inflight = 0;
q->commit_total = 0;
q->commit_responses = 0;
uint32_t cmd = q->has_commit_override ? q->commit_command : q->command;
const uint8_t *val = q->has_commit_override
? q->commit_value_buf.data() : q->value;
size_t val_len = q->has_commit_override
? q->commit_value_buf.size() : q->value_len;
DHT_DISC("[DhtRpc] _commit_query: qid=%u cmd=%u responded=%d has_fn=%d\n",
q->id, cmd, q->responded_count, q->commit_value_fn ? 1 : 0);
for (int i = 0; i < q->responded_count; i++) {
DhtQueryPeer &rp = q->responded[i];
DHT_DISC("[DhtRpc] responded[%d] has_token=%d addr=%d.%d.%d.%d:%d\n",
i, rp.has_token,
rp.address.ip[0], rp.address.ip[1],
rp.address.ip[2], rp.address.ip[3],
rp.address.port);
if (!rp.has_token) continue; // No token = can't commit to this node.
// Determine commit value: per-node callback or static buffer.
const uint8_t *commit_val;
size_t commit_val_len;
std::vector<uint8_t> per_node_val;
if (q->commit_value_fn) {
// Compute the node's ID from its address (matches JS peer.id()).
NodeId node_id = cenc::node_id_from_address(rp.address);
per_node_val = q->commit_value_fn(q->target, rp.token, node_id.bytes, rp.address);
commit_val = per_node_val.data();
commit_val_len = per_node_val.size();
} else {
commit_val = val;
commit_val_len = val_len;
}
uint16_t tid = _next_tid();
// Commit phase always uses internal=false — commits are user-facing
// commands (ANNOUNCE, MUTABLE_PUT, etc.), not routing commands.
_alloc_inflight(tid, rp.address, false, cmd,
q->target, q->id); // tag = query id for routing commit responses.
_send_request(tid, rp.address, rp.token,
false, cmd, q->target,
commit_val, commit_val_len);
q->commit_inflight++;
q->commit_total++;
}
DHT_DISC("[DhtRpc] _commit_query: sent %d commit requests\n", q->commit_total);
// If no nodes had tokens, complete immediately.
if (q->commit_inflight == 0) {
q->committing = false;
q->complete = true;
if (on_query_complete) on_query_complete(q->id);
for (int i = 0; i < _query_count; i++) {
if (_queries[i] == q) {
_queries[i] = _queries[_query_count - 1];
_queries[_query_count - 1] = nullptr;
_query_count--;
delete q;
return;
}
}
}
}
DhtQuery *DhtRpc::_find_query(uint32_t id) {
for (int i = 0; i < _query_count; i++) {
if (_queries[i] && _queries[i]->id == id) return _queries[i];
}
return nullptr;
}
// ---- Token management ----
void DhtRpc::_init_secrets() {
if (_secrets_initialized) return;
randombytes_buf(_secrets[0], 32);
randombytes_buf(_secrets[1], 32);
_secrets_initialized = true;
}
void DhtRpc::_rotate_secrets() {
if (!_secrets_initialized) return;
// Swap: secret[0] becomes the old one, secret[1] becomes new.
uint8_t tmp[32];
memcpy(tmp, _secrets[0], 32);
memcpy(_secrets[0], _secrets[1], 32);
memcpy(_secrets[1], tmp, 32);
// Derive new secret from old.
crypto_generichash(_secrets[1], 32, _secrets[1], 32, nullptr, 0);
sodium_memzero(tmp, 32);
}
void DhtRpc::generate_token(const IPv4Address &addr, uint8_t *out) {
_init_secrets();
// Token = BLAKE2b(key=secret[1], data=IP_string).
// Matches JS: crypto_generichash(token, Buffer.from(addr.host), secret[i])
char host_str[16];
snprintf(host_str, sizeof(host_str), "%d.%d.%d.%d",
addr.ip[0], addr.ip[1], addr.ip[2], addr.ip[3]);
size_t host_len = strlen(host_str);
crypto_generichash(out, 32,
reinterpret_cast<const uint8_t *>(host_str), host_len,
_secrets[1], 32);
}
bool DhtRpc::validate_token(const IPv4Address &addr, const uint8_t *token) {
_init_secrets();
char host_str[16];
snprintf(host_str, sizeof(host_str), "%d.%d.%d.%d",
addr.ip[0], addr.ip[1], addr.ip[2], addr.ip[3]);
size_t host_len = strlen(host_str);
// Check against both secrets (current and previous).
uint8_t expected[32];
for (int i = 0; i < 2; i++) {
crypto_generichash(expected, 32,
reinterpret_cast<const uint8_t *>(host_str), host_len,
_secrets[i], 32);
if (sodium_memcmp(expected, token, 32) == 0) {
sodium_memzero(expected, 32);
return true;
}
}
sodium_memzero(expected, 32);
return false;
}
// ---- Public send_request wrapper ----
void DhtRpc::send_request(uint32_t command, const uint8_t *target,
const uint8_t *value, size_t value_len,
const IPv4Address &to, bool internal,
uint32_t tag, const uint8_t *token) {
uint16_t tid = _next_tid();
// Register inflight.
if (_inflight_count < DHT_MAX_INFLIGHT) {
DhtInflight &inf = _inflight[_inflight_count++];
memset(&inf, 0, sizeof(inf));
inf.tid = tid;
inf.to = to;
inf.tag = tag;
inf.active = true;
inf.retries_left = 5;
}
_send_request(tid, to, token, internal, command, target, value, value_len);
}