#ifndef HYPERSWARM_DHT_RPC_H
#define HYPERSWARM_DHT_RPC_H
// DHT-RPC client — wire-compatible with holepunchto/dht-rpc.
// Client-only: bootstrap, PING, FIND_NODE, and iterative queries.
// Uses raw UDP via udx_socket_send(), ticked from the libuv event loop.
#include "compact_encoding.h"
#include "kademlia_table.h"
#include "nat_sampler.h"
#include <cstdint>
#include <cstring>
#include <functional>
#include <vector>
extern "C" {
#include <udx.h>
#include <uv.h>
}
// ---- Constants ----
static constexpr uint8_t DHT_VERSION = 0x03;
static constexpr uint8_t DHT_REQUEST_ID = (0x00 << 4) | DHT_VERSION; // 0x03
static constexpr uint8_t DHT_RESPONSE_ID = (0x01 << 4) | DHT_VERSION; // 0x13
// Flags (shared for request and response, different semantics for bit 2 and 3).
static constexpr uint8_t DHT_FLAG_HAS_ID = 0x01;
static constexpr uint8_t DHT_FLAG_TOKEN = 0x02;
static constexpr uint8_t DHT_FLAG_INTERNAL = 0x04; // Request: internal command.
static constexpr uint8_t DHT_FLAG_CLOSER_NODES = 0x04; // Response: closer nodes present.
static constexpr uint8_t DHT_FLAG_TARGET = 0x08; // Request: target hash present.
static constexpr uint8_t DHT_FLAG_ERROR = 0x08; // Response: error code present.
static constexpr uint8_t DHT_FLAG_VALUE = 0x10; // Value/payload present.
// Internal commands (dht-rpc/lib/commands.js).
static constexpr uint32_t DHT_CMD_PING = 0;
static constexpr uint32_t DHT_CMD_PING_NAT = 1;
static constexpr uint32_t DHT_CMD_FIND_NODE = 2;
static constexpr uint32_t DHT_CMD_DOWN_HINT = 3;
static constexpr uint32_t DHT_CMD_DELAYED_PING = 4;
// Default concurrency for iterative queries.
static constexpr int DHT_QUERY_CONCURRENCY = 10;
static constexpr int DHT_QUERY_K = 20;
static constexpr int DHT_MAX_INFLIGHT = 256;
static constexpr int DHT_MAX_CLOSER_NODES = 20;
static constexpr int DHT_MAX_BOOTSTRAP_NODES = 8;
static constexpr int DHT_INFLIGHT_TIMEOUT_MS = 2000;
static constexpr int DHT_SECRET_ROTATE_INTERVAL = 10; // Drain cycles (750ms each).
// ---- Structures ----
// Decoded DHT response.
struct DhtResponse {
uint16_t tid;
IPv4Address from; // Sender (from UDP recvfrom).
IPv4Address to; // "to" field in the message.
NodeId from_id; // Validated node ID (zeroed if not present/invalid).
bool has_id;
uint8_t token[32];
bool has_token;
IPv4Address closer_nodes[DHT_MAX_CLOSER_NODES];
int closer_count;
uint32_t error;
const uint8_t *value;
size_t value_len;
uint32_t rtt_ms;
};
// Decoded DHT request (for incoming).
struct DhtRequest {
uint16_t tid;
IPv4Address from;
IPv4Address to;
NodeId from_id;
bool has_id;
uint8_t token[32];
bool has_token;
bool internal;
uint32_t command;
uint8_t target[32];
bool has_target;
const uint8_t *value;
size_t value_len;
};
// Inflight request tracking.
struct DhtInflight {
uint16_t tid;
uint64_t sent_at_ms;
IPv4Address to;
bool internal;
uint32_t command;
uint8_t target[32];
bool has_target;
int retries_left;
int sent_count;
uint32_t tag; // Caller-defined tag for routing responses.
bool active;
// Encoded request buffer (cached for retries).
uint8_t *buffer;
size_t buffer_len;
};
// Bootstrap node descriptor.
struct BootstrapNode {
IPv4Address address;
bool resolved; // True if IP is known.
};
// ---- Iterative query state ----
struct DhtQueryPeer {
NodeId id;
IPv4Address address;
bool queried;
bool responded;
uint8_t token[32]; // Token received in response (for commit phase).
bool has_token;
};
struct DhtQuery {
uint32_t id;
uint8_t target[32];
bool internal;
uint32_t command;
const uint8_t *value;
size_t value_len;
uint32_t tag;
// Commit-phase overrides (for split traversal/commit commands).
// If commit_command != 0, use these instead of command/value during commit.
// Matches JS pattern: immutablePut traverses with IMMUTABLE_GET, commits with IMMUTABLE_PUT.
uint32_t commit_command;
std::vector<uint8_t> commit_value_buf; // Owns the commit value data.
bool has_commit_override;
// Per-node commit value callback (e.g., for announce per-node signatures).
// If set, called for each responding node during commit to generate the value.
// Arguments: target (32B), token (32B), node_id (32B from node_id_from_address).
// Returns: the commit value for this specific node.
std::function<std::vector<uint8_t>(const uint8_t *target,
const uint8_t *token,
const uint8_t *node_id,
const IPv4Address &from)> commit_value_fn;
DhtQueryPeer closest[DHT_QUERY_K * 3]; // Working set.
int closest_count;
DhtQueryPeer responded[DHT_QUERY_K]; // Closest that responded.
int responded_count;
int inflight;
int successes;
int errors;
bool complete;
bool commit; // If true, send a commit (non-internal) after query.
bool committing; // True during commit phase (sending with tokens).
int commit_inflight; // Number of inflight commit requests.
int commit_total; // Total commit requests sent.
int commit_responses; // Commit responses received.
};
// ---- DhtRpc class ----
class DhtRpc {
public:
DhtRpc();
~DhtRpc();
// Initialize with a UDP socket and event loop (must be called from UDX thread).
void init(udx_socket_t *socket, uv_loop_t *loop);
// Set bootstrap nodes. Format: "IP:PORT" strings.
void set_bootstrap(const char **nodes, int count);
// Generate our node ID from our bound address.
void set_local_address(const IPv4Address &addr);
// Start bootstrap: FIND_NODE(self) to bootstrap nodes.
void bootstrap();
// Send a PING to a specific address.
void ping(const IPv4Address &to, uint32_t tag = 0);
// Send a FIND_NODE for a target.
void find_node(const uint8_t *target, const IPv4Address &to, uint32_t tag = 0);
// Start an iterative query (FIND_NODE or custom command).
uint32_t query(const uint8_t *target, bool internal, uint32_t command,
const uint8_t *value = nullptr, size_t value_len = 0,
uint32_t tag = 0, bool commit = false);
// Start a query with separate commit-phase command/value.
// Matches JS pattern: traverse with GET command, commit with PUT command.
uint32_t 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 = 0);
// Start a query with a per-node commit value callback.
// The callback generates a different commit value for each responding node,
// needed for operations like announce where the signature includes per-node data.
uint32_t 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,
// (target, token, node_id, from) -> the value to commit at THAT node.
// ⛔ `from` was missing and that is why nothing could record which nodes
// actually hold our record: the commit reply is the only place the
// holder's address appears. _store_announced_nodes had to guess with
// table().closest() and zero the token. See PARITY-hyperswarm.md §15.
std::function<std::vector<uint8_t>(const uint8_t *, const uint8_t *,
const uint8_t *, const IPv4Address &)> commit_fn,
uint32_t tag = 0);
// Process an incoming UDP message. Called from socket recv callback.
void on_message(const uint8_t *data, size_t len, const struct sockaddr_in *from);
// Tick: process timeouts, retries, advance queries. Called from timer.
void tick(uint64_t now_ms);
// Access the routing table.
KademliaTable &table() { return _table; }
const KademliaTable &table() const { return _table; }
bool is_bootstrapped() const { return _bootstrapped; }
// NAT / firewall detection.
uint32_t local_firewall() const { return _nat_sampler.firewall(); }
const NatSampler &nat_sampler() const { return _nat_sampler; }
// Set real network interface addresses (LAN IPs from uv_interface_addresses).
// These are included in handshake NoisePayload.addresses4 so remote peers
// can find LAN shortcuts — matching JS Holepuncher.localAddresses() behavior.
void set_interface_addresses(const IPv4Address *addrs, int count) {
_iface_addr_count = (count > 8) ? 8 : count;
for (int i = 0; i < _iface_addr_count; i++) _iface_addrs[i] = addrs[i];
}
// Get all known addresses for handshake payload, deduped.
// Order matches JS reference (connect.js:391-394, server.js:270-277):
// 1. NatSampler external addresses (like JS dht.remoteAddress())
// 2. Real interface addresses (LAN IPs — like JS localAddresses())
// 3. Bind address fallback (only if nothing else available)
int get_all_addresses(IPv4Address *out, int max_count) const {
int count = 0;
// 1. NatSampler external addresses.
for (int i = 0; i < _nat_sampler.address_count() && count < max_count; i++) {
const IPv4Address &a = _nat_sampler.addresses()[i];
if (a.ip[0] == 0 || a.port == 0) continue;
out[count++] = a;
}
// 2. Real interface addresses (LAN IPs).
for (int i = 0; i < _iface_addr_count && count < max_count; i++) {
bool dup = false;
for (int j = 0; j < count; j++) {
if (memcmp(out[j].ip, _iface_addrs[i].ip, 4) == 0 &&
out[j].port == _iface_addrs[i].port) { dup = true; break; }
}
if (!dup) out[count++] = _iface_addrs[i];
}
// 3. Bind address fallback (only if nothing else available).
if (count == 0 && _has_local_addr && count < max_count) {
out[count++] = _local_addr;
}
return count;
}
// Token management (for responding to requests as a DHT node).
void generate_token(const IPv4Address &addr, uint8_t *out);
bool validate_token(const IPv4Address &addr, const uint8_t *token);
// Send a custom request (used by DhtConnector for PEER_HOLEPUNCH).
// token: if non-null, included in the request (required for commit phase).
void 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 = 0, const uint8_t *token = nullptr);
// Callbacks.
std::function<void()> on_bootstrapped;
std::function<void(const DhtResponse &, uint32_t tag)> on_response;
std::function<void(const DhtRequest &)> on_request;
std::function<void(uint32_t query_id)> on_query_complete;
friend class HyperDht;
private:
// Encode and send a request.
void _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);
// Encode and send a response (also used by HyperDht for handshake replies).
void _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);
// Low-level UDP send.
void _udp_send(const uint8_t *data, size_t len, const IPv4Address &to);
// Decode incoming messages.
bool _decode_response(const uint8_t *data, size_t len,
const struct sockaddr_in *from, DhtResponse &out);
bool _decode_request(const uint8_t *data, size_t len,
const struct sockaddr_in *from, DhtRequest &out);
// Handle decoded response: update table, match inflight, advance queries.
void _handle_response(const DhtResponse &res);
// Handle decoded request: respond to PING, FIND_NODE, etc.
void _handle_request(const DhtRequest &req);
// Inflight management.
DhtInflight *_alloc_inflight(uint16_t tid, const IPv4Address &to,
bool internal, uint32_t command,
const uint8_t *target, uint32_t tag);
DhtInflight *_find_inflight(uint16_t tid);
void _free_inflight(DhtInflight *inf);
// Iterative query advancement.
void _advance_query(DhtQuery *q);
void _add_query_peer(DhtQuery *q, const NodeId &id, const IPv4Address &addr);
void _complete_query(DhtQuery *q);
void _commit_query(DhtQuery *q); // Commit phase: send with tokens to closest responded.
DhtQuery *_find_query(uint32_t id);
// Secret rotation for token generation.
void _rotate_secrets();
void _init_secrets();
// Transaction ID counter.
uint16_t _next_tid();
// State.
udx_socket_t *_socket;
uv_loop_t *_loop;
KademliaTable _table;
IPv4Address _local_addr;
bool _has_local_addr;
// Real network interface addresses (LAN IPs from uv_interface_addresses).
IPv4Address _iface_addrs[8];
int _iface_addr_count;
BootstrapNode _bootstrap_nodes[DHT_MAX_BOOTSTRAP_NODES];
int _bootstrap_count;
bool _bootstrapped;
bool _bootstrapping;
int _bootstrap_pending; // Inflight bootstrap requests.
DhtInflight _inflight[DHT_MAX_INFLIGHT];
int _inflight_count;
uint16_t _tid_counter;
DhtQuery *_queries[16]; // Max concurrent iterative queries.
int _query_count;
uint32_t _query_id_counter;
// Token secrets (two rotating 32-byte secrets).
uint8_t _secrets[2][32];
bool _secrets_initialized;
int _secret_rotate_counter;
uint64_t _last_tick_ms;
// NAT sampler for firewall detection.
NatSampler _nat_sampler;
public:
// UDP send request (public for static callback access).
struct UdpSendReq {
udx_socket_send_t req;
uv_buf_t buf;
uint8_t data[1200]; // Max DHT message size (announce + signature can exceed 512).
};
};
#endif // HYPERSWARM_DHT_RPC_H