// E2E Interop Client: Connect to a JS HyperDHT server.
//
// Usage: ./test_interop_client <bootstrap_host:port> <server_pk_hex>
//
// Flow:
// 1. Init libuv + libudx + udx_socket
// 2. Init DhtRpc, bootstrap to provided node
// 3. Init HyperDht + DhtConnector
// 4. DhtConnector::connect(remote_pk)
// 5. On connected: report success
// 6. Exit 0 success / 1 failure
//
// Build: see build_tests.sh (test_interop_client target)
#include "compact_encoding.h"
#include "dht_connector.h"
#include "dht_messages.h"
#include "dht_rpc.h"
#include "holepuncher.h"
#include "hyperdht.h"
#include "noise_ik.h"
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <sodium.h>
extern "C" {
#include <udx.h>
#include <uv.h>
}
// ---- Global state ----
static uv_loop_t g_loop;
static udx_t g_udx;
static udx_socket_t g_socket;
static uv_timer_t g_tick_timer;
static uv_timer_t g_timeout_timer;
static DhtRpc *g_rpc = nullptr;
static HyperDht *g_dht = nullptr;
static DhtConnector *g_connector = nullptr;
static bool g_bootstrapped = false;
static bool g_connected = false;
static bool g_failed = false;
static int g_exit_code = 1;
// ---- Hex decode ----
static bool hex_decode(const char *hex, uint8_t *out, int len) {
for (int i = 0; i < len; i++) {
unsigned int byte;
if (sscanf(hex + i * 2, "%02x", &byte) != 1) return false;
out[i] = static_cast<uint8_t>(byte);
}
return true;
}
// ---- Callbacks ----
static void on_socket_recv(udx_socket_t *socket, ssize_t read_len,
const uv_buf_t *buf, const struct sockaddr *from) {
if (read_len <= 0) return;
const uint8_t *data = (const uint8_t *)buf->base;
size_t len = (size_t)read_len;
// DHT-RPC messages.
if (g_rpc && len >= 4) {
uint8_t type_byte = data[0];
if (type_byte == DHT_REQUEST_ID || type_byte == DHT_RESPONSE_ID) {
g_rpc->on_message(data, len, (const struct sockaddr_in *)from);
return;
}
}
// Holepunch probes.
if (g_connector && len == 1 && data[0] == 0x00) {
const struct sockaddr_in *from_in = (const struct sockaddr_in *)from;
IPv4Address addr;
memcpy(addr.ip, &from_in->sin_addr.s_addr, 4);
addr.port = ntohs(from_in->sin_port);
g_connector->on_probe_received(addr);
}
}
static void on_tick(uv_timer_t *handle) {
if (!g_rpc) return;
uint64_t now = uv_now(&g_loop);
g_rpc->tick(now);
if (g_dht) g_dht->tick(now);
if (g_connector) g_connector->tick(now);
}
static void on_timeout(uv_timer_t *handle) {
fprintf(stderr, "Timeout — no connection established\n");
g_exit_code = 1;
uv_stop(&g_loop);
}
static void cleanup_and_exit() {
uv_timer_stop(&g_tick_timer);
uv_timer_stop(&g_timeout_timer);
uv_stop(&g_loop);
}
// ---- Main ----
int main(int argc, char **argv) {
if (argc < 3) {
fprintf(stderr, "Usage: %s <bootstrap_host:port> <server_pk_hex>\n", argv[0]);
return 1;
}
if (sodium_init() < 0) {
fprintf(stderr, "sodium_init failed\n");
return 1;
}
// Parse args.
const char *bootstrap_str = argv[1];
const char *pk_hex = argv[2];
uint8_t remote_pk[32];
if (strlen(pk_hex) != 64 || !hex_decode(pk_hex, remote_pk, 32)) {
fprintf(stderr, "Invalid public key hex (need 64 chars)\n");
return 1;
}
// Parse comma-separated bootstrap nodes.
char bootstrap_buf[512];
strncpy(bootstrap_buf, bootstrap_str, sizeof(bootstrap_buf) - 1);
bootstrap_buf[sizeof(bootstrap_buf) - 1] = '\0';
const char *boot_ptrs[8];
int boot_count = 0;
char *tok = strtok(bootstrap_buf, ",");
while (tok && boot_count < 8) {
boot_ptrs[boot_count++] = tok;
tok = strtok(nullptr, ",");
}
printf("bootstrap: %s (%d nodes)\n", bootstrap_str, boot_count);
printf("remote pk: %.16s...\n", pk_hex);
// Init libuv.
uv_loop_init(&g_loop);
udx_init(&g_loop, &g_udx, nullptr);
// Init socket.
udx_socket_init(&g_udx, &g_socket, nullptr);
g_socket.data = nullptr;
struct sockaddr_in bind_addr;
uv_ip4_addr("0.0.0.0", 0, &bind_addr);
udx_socket_bind(&g_socket, (struct sockaddr *)&bind_addr, 0);
struct sockaddr_in local_addr;
int addr_len = sizeof(local_addr);
udx_socket_getsockname(&g_socket, (struct sockaddr *)&local_addr, &addr_len);
int port = ntohs(local_addr.sin_port);
printf("bound to port %d\n", port);
// Start recv.
udx_socket_recv_start(&g_socket, on_socket_recv);
// Timer.
uv_timer_init(&g_loop, &g_tick_timer);
uv_timer_start(&g_tick_timer, on_tick, 0, 1);
// Timeout (10s for interop).
uv_timer_init(&g_loop, &g_timeout_timer);
uv_timer_start(&g_timeout_timer, on_timeout, 10000, 0);
// Generate our Ed25519 keypair.
Ed25519KeyPair kp;
crypto_sign_keypair(kp.public_key, kp.secret_key);
// Init DHT-RPC.
g_rpc = new DhtRpc();
g_rpc->init(&g_socket, &g_loop);
g_rpc->set_bootstrap(boot_ptrs, boot_count);
IPv4Address local;
local.ip[0] = 127; local.ip[1] = 0; local.ip[2] = 0; local.ip[3] = 1;
local.port = port;
g_rpc->set_local_address(local);
// Collect real interface addresses so handshake payload includes LAN IPs.
uv_interface_address_t *ifaces = nullptr;
int iface_count = 0;
uv_interface_addresses(&ifaces, &iface_count);
IPv4Address lan_addrs[8];
int lan_count = 0;
if (ifaces) {
for (int i = 0; i < iface_count && lan_count < 8; i++) {
if (ifaces[i].is_internal) continue;
if (ifaces[i].address.address4.sin_family != AF_INET) continue;
memcpy(lan_addrs[lan_count].ip, &ifaces[i].address.address4.sin_addr.s_addr, 4);
lan_addrs[lan_count].port = port;
lan_count++;
}
uv_free_interface_addresses(ifaces, iface_count);
}
if (lan_count == 0) { lan_addrs[0] = local; lan_count = 1; }
g_rpc->set_interface_addresses(lan_addrs, lan_count);
printf("interface addresses: %d (first: %d.%d.%d.%d:%d)\n", lan_count,
lan_addrs[0].ip[0], lan_addrs[0].ip[1], lan_addrs[0].ip[2], lan_addrs[0].ip[3], lan_addrs[0].port);
// Init HyperDht.
g_dht = new HyperDht();
g_dht->init(g_rpc, kp);
// Init DhtConnector.
g_connector = new DhtConnector();
g_connector->init(g_dht, &g_socket, &g_loop);
g_connector->set_local_addresses(lan_addrs, lan_count);
// On connected.
g_connector->on_connected = [](uint32_t conn_id, const DhtConnection &conn) {
printf("connected to server via DHT (conn_id=%u)\n", conn_id);
printf("remote address: %d.%d.%d.%d:%d\n",
conn.punched_address.ip[0], conn.punched_address.ip[1],
conn.punched_address.ip[2], conn.punched_address.ip[3],
conn.punched_address.port);
g_connected = true;
g_exit_code = 0;
cleanup_and_exit();
};
g_connector->on_failed = [](uint32_t conn_id, const uint8_t *remote_pk, const char *error) {
printf("connection failed: %s\n", error);
g_failed = true;
g_exit_code = 1;
cleanup_and_exit();
};
// Debug: chain peer found callback.
auto orig_peer_found = g_dht->on_peer_found;
g_dht->on_peer_found = [orig_peer_found](uint32_t query_id, const FoundPeer &fp) {
char pk_hex[65];
for (int i = 0; i < 32; i++) sprintf(pk_hex + i*2, "%02x", fp.public_key[i]);
pk_hex[64] = 0;
printf("peer found: %.16s... relays=%d relay_node=%d.%d.%d.%d:%d\n",
pk_hex, fp.relay_count,
fp.relay_node.ip[0], fp.relay_node.ip[1],
fp.relay_node.ip[2], fp.relay_node.ip[3], fp.relay_node.port);
if (orig_peer_found) orig_peer_found(query_id, fp);
};
// Debug: chain handshake complete callback.
auto orig_hs_complete = g_dht->on_handshake_complete;
g_dht->on_handshake_complete = [orig_hs_complete](uint32_t hs_id, const DhtHandshake &hs) {
printf("handshake complete! id=%u complete=%d\n", hs_id, hs.complete ? 1 : 0);
if (hs.has_remote_payload) {
printf(" remote payload: version=%u error=%u firewall=%u\n",
hs.remote_payload.version, hs.remote_payload.error,
hs.remote_payload.firewall);
printf(" has_udx=%d has_holepunch=%d addresses4=%d\n",
hs.remote_payload.has_udx ? 1 : 0,
hs.remote_payload.has_holepunch ? 1 : 0,
hs.remote_payload.addresses4_count);
if (hs.remote_payload.has_udx) {
printf(" udx: id=%u seq=%u reusable=%d\n",
hs.remote_payload.udx.id, hs.remote_payload.udx.seq,
hs.remote_payload.udx.reusable_socket ? 1 : 0);
}
if (hs.remote_payload.addresses4_count > 0) {
printf(" addr[0]: %d.%d.%d.%d:%d\n",
hs.remote_payload.addresses4[0].ip[0],
hs.remote_payload.addresses4[0].ip[1],
hs.remote_payload.addresses4[0].ip[2],
hs.remote_payload.addresses4[0].ip[3],
hs.remote_payload.addresses4[0].port);
}
} else {
printf(" NO remote payload!\n");
}
if (orig_hs_complete) orig_hs_complete(hs_id, hs);
};
// Chain debug logging onto HyperDht's response callback.
auto original_on_response = g_rpc->on_response;
g_rpc->on_response = [original_on_response](const DhtResponse &res, uint32_t tag) {
printf(" response from %d.%d.%d.%d:%d has_id=%d closer=%d value=%zu tag=0x%x\n",
res.from.ip[0], res.from.ip[1], res.from.ip[2], res.from.ip[3], res.from.port,
res.has_id ? 1 : 0, res.closer_count, res.value_len, tag);
// Forward to HyperDht's handler.
if (original_on_response) original_on_response(res, tag);
};
// Bootstrap callback.
g_rpc->on_bootstrapped = [&]() {
g_bootstrapped = true;
printf("bootstrap complete (table has %d nodes)\n", g_rpc->table().size());
printf("firewall: %u\n", g_rpc->local_firewall());
// Now connect to the server.
printf("connecting to remote peer...\n");
g_connector->connect(remote_pk);
};
// Bootstrap.
printf("bootstrapping...\n");
g_rpc->bootstrap();
// Run event loop.
uv_run(&g_loop, UV_RUN_DEFAULT);
// Cleanup.
delete g_connector;
delete g_dht;
delete g_rpc;
uv_close((uv_handle_t *)&g_tick_timer, nullptr);
uv_close((uv_handle_t *)&g_timeout_timer, nullptr);
udx_socket_close(&g_socket);
uv_run(&g_loop, UV_RUN_DEFAULT);
uv_loop_close(&g_loop);
printf("exit code: %d\n", g_exit_code);
return g_exit_code;
}