3c79be0 · hyperswarm-peer

Changed files (5) +139 -3

src/udx_transport.cpp +22 -3

…	…	@@ -985,6 +985,10 @@ void UdxTransport::_on_stream_close(udx_stream_t *stream, int status) {
985	985
986	986	peer->connected = false;
987	987
	988	// A retired peer (dedup, disconnect_peer) was already reported; its peer id
	989	// may now belong to the connection that replaced it. Say nothing.
	990	if (peer->doomed) return;
	991
988	992	UdxTransport transport = static_cast<UdxTransport >(stream->socket->data);
989	993	if (!transport) {
990	994	DHT_LOG("[UdxTransport] stream_close: socket->data is NULL! Cannot propagate disconnect.\n");
…	…	@@ -1006,8 +1010,12 @@ void UdxTransport::_on_stream_close(udx_stream_t *stream, int status) {
1006	1010	}
1007	1011
1008	1012	void UdxTransport::_on_stream_finalize(udx_stream_t *stream) {
1009		// Stream fully cleaned up — safe to free PeerConnection if needed.
1010		(void)stream;
	1013	// libudx has closed every handle in the stream: only now may its
	1014	// PeerConnection go, if it was retired.
	1015	PeerConnection peer = static_cast<PeerConnection >(stream->data);
	1016	if (!peer) return;
	1017	peer->finalized = true;
	1018	if (peer->doomed) delete peer;
1011	1019	}
1012	1020
1013	1021	void UdxTransport::_on_socket_recv(udx_socket_t *socket, ssize_t read_len,
…	…	@@ -1094,6 +1102,7 @@ PeerConnection *UdxTransport::_add_peer(int32_t peer_id, uint32_t local_sid,
1094	1102	udx_stream_init(&_udx, &peer->stream, local_sid,
1095	1103	_on_stream_close, _on_stream_finalize);
1096	1104	peer->stream.data = peer;
	1105	peer->stream_initialized = true;
1097	1106
1098	1107	// Connect stream to remote — use birthday winner socket if available.
1099	1108	udx_socket_t *use_socket = socket_override ? socket_override : &_socket;
…	…	@@ -1149,8 +1158,18 @@ void UdxTransport::_remove_peer(int32_t peer_id) {
1149	1158	_peers_by_pubkey.erase(pk_it);
1150	1159	}
1151	1160	}
1152		delete peer;
1153	1161	_peers.erase(it);
	1162	// Delete now only if libudx is done with the stream; otherwise retire it
	1163	// and let _on_stream_finalize delete it (see PeerConnection::doomed).
	1164	if (!peer->stream_initialized \|\| peer->finalized) {
	1165	delete peer;
	1166	return;
	1167	}
	1168	peer->doomed = true;
	1169	if (peer->connected) {
	1170	udx_stream_destroy(&peer->stream);
	1171	peer->connected = false;
	1172	}
1154	1173	}
1155	1174	}
1156	1175

src/udx_transport.h +10 -0

…	…	@@ -107,6 +107,16 @@ struct PeerConnection {
107	107	// True when this connection belongs to the stream consumer (no internal mux).
108	108	bool stream_consumer = false;
109	109
	110	// ⛔ LIFETIME. udx_stream_t (held by value above) embeds libuv handles that
	111	// udx_stream_destroy closes ASYNCHRONOUSLY, signalling completion via the
	112	// stream's on_finalize. A PeerConnection whose stream is initialised must
	113	// therefore not be deleted before finalize — _remove_peer RETIRES it (doomed)
	114	// and _on_stream_finalize deletes it. Deleting at once left libuv walking its
	115	// closing queue through freed memory: uv__finish_close's assertion.
	116	bool stream_initialized = false; // udx_stream_init ran
	117	bool finalized = false; // libudx finished with the stream's handles
	118	bool doomed = false; // removed from _peers; delete on finalize
	119
110	120	// Receive buffer for length-prefixed framing (matches JS SecretStream's uint24 LE).
111	121	// UDX reliable streams are byte streams — multiple writes can coalesce into one read.
112	122	std::vector<uint8_t> recv_buf;

test/build_tests.sh +11 -0

…	…	@@ -118,6 +118,17 @@ g++ -o test_stream_consumer \
118	118	$UDX_OBJS $LDFLAGS
119	119	echo " OK: test_stream_consumer"
120	120
	121	# ---- test_peer_lifecycle: a PeerConnection outlives its udx stream's libuv handles ----
	122	echo "Building test_peer_lifecycle..."
	123	g++ $CXXFLAGS -o $OBJ/test_peer_lifecycle.o test_peer_lifecycle.cpp
	124	g++ -o test_peer_lifecycle \
	125	$OBJ/test_peer_lifecycle.o $OBJ/noise_ik.o $OBJ/noise_handshake.o \
	126	$OBJ/noise_stream.o $OBJ/dht_rpc.o $OBJ/holepuncher.o \
	127	$OBJ/hyperdht.o $OBJ/dht_connector.o $OBJ/udx_transport.o \
	128	$OBJ/swarm_manager.o $OBJ/protomux.o \
	129	$UDX_OBJS $LDFLAGS
	130	echo " OK: test_peer_lifecycle"
	131
121	132	# ---- test_interop_client ----
122	133	echo "Building test_interop_client..."
123	134	g++ $CXXFLAGS -o $OBJ/test_interop_client.o test_interop_client.cpp

test/test_peer_lifecycle.cpp +90 -0

…	…	@@ -0,0 +1,90 @@
	1	// Gate: a PeerConnection must outlive its udx stream's libuv handles.
	2	//
	3	// udx_stream_t embeds three libuv handles (timer, refill_pacing_timer,
	4	// pending_packet_prepare). udx_stream_destroy closes them ASYNCHRONOUSLY and says so
	5	// through the stream's on_finalize (libudx udx.c:355-361, 480-482). The transport used to
	6	// `delete` the PeerConnection — which holds the udx_stream_t by value — right after
	7	// udx_stream_destroy, on disconnect_peer and on connection dedup. libuv then walked its
	8	// closing queue through freed memory: uv__finish_close's assertion
	9	// (`handle->flags & UV_HANDLE_CLOSING`), seen by keetd's M3d run on the public DHT.
	10	//
	11	// Drives the deterministic path, disconnect_peer, three times on two meshed transports.
	12	// Build with SAN=address to see the use-after-free directly; a plain build can pass by
	13	// luck (freed memory not yet reused), which is why the ASan run is the gate.
	14	//
	15	// node ../integration/layer2_testnet.js &
	16	// SAN=address ./build_tests.sh && ./test_peer_lifecycle [host:port]
	17	#include "noise_ik.h"
	18	#include "udx_transport.h"
	19
	20	#include <sodium.h>
	21
	22	#include <atomic>
	23	#include <chrono>
	24	#include <cstdio>
	25	#include <cstring>
	26	#include <fstream>
	27	#include <mutex>
	28	#include <string>
	29	#include <thread>
	30	#include <unistd.h>
	31
	32	static int passed = 0, failed = 0;
	33	static void check (bool ok, const char *what) { (ok ? passed : failed)++; printf(" %s %s\n", ok ? "PASS" : "FAIL", what); }
	34	static void sleep_ms (int ms) { std::this_thread::sleep_for(std::chrono::milliseconds(ms)); }
	35	template <typename F> static bool wait_for (F f, int ms) {
	36	for (int i = 0; i < ms / 10; i++) { if (f()) return true; sleep_ms(10); }
	37	return f();
	38	}
	39
	40	int main (int argc, char **argv) {
	41	if (sodium_init() < 0) return 2;
	42	std::string bs;
	43	if (argc > 1) bs = argv[1];
	44	else { std::ifstream f("/tmp/layer_bootstrap.txt"); if (f) std::getline(f, bs); }
	45	if (bs.find(',') != std::string::npos) bs = bs.substr(0, bs.find(','));
	46	while (!bs.empty() && isspace((unsigned char)bs.back())) bs.pop_back();
	47	if (bs.empty()) { printf("FAIL: no bootstrap (node ../integration/layer2_testnet.js &)\n"); return 1; }
	48
	49	UdxTransport a, b;
	50	std::mutex m;
	51	int32_t aPeer = 0;
	52	std::atomic<int> aReady{ 0 }, bReady{ 0 }, bGone{ 0 };
	53	a.set_mux_callbacks([&] (int32_t id, Protomux *) { std::lock_guard<std::mutex> l(m); aPeer = id; aReady++; }, nullptr);
	54	b.set_mux_callbacks([&] (int32_t, Protomux *) { bReady++; }, [&] (int32_t) { bGone++; });
	55
	56	uint8_t topic[32];
	57	char seed[64];
	58	snprintf(seed, sizeof(seed), "peer-lifecycle/%d", (int)getpid());
	59	crypto_generichash(topic, 32, (const uint8_t *)seed, strlen(seed), nullptr, 0);
	60	const char *bsv[1] = { bs.c_str() };
	61	Ed25519KeyPair ka = NoiseIK::generate_keypair(), kb = NoiseIK::generate_keypair();
	62	a.start_dht(ka, bsv, 1);
	63	b.start_dht(kb, bsv, 1);
	64	check(wait_for([&] { return a.is_dht_active() && b.is_dht_active(); }, 15000), "both DHTs active");
	65	a.swarm_join(topic, true, true);
	66	b.swarm_join(topic, true, true);
	67
	68	for (int round = 1; round <= 3; round++) {
	69	const int want = round;
	70	const bool up = wait_for([&] { return aReady >= want && bReady >= want; }, 40000);
	71	char what[80];
	72	snprintf(what, sizeof(what), "round %d: meshed", round);
	73	check(up, what);
	74	if (!up) break;
	75	int32_t pid;
	76	{ std::lock_guard<std::mutex> l(m); pid = aPeer; }
	77	const int gone = bGone;
	78	a.disconnect_peer(pid, true); // destroy + remove: the path that freed too early
	79	snprintf(what, sizeof(what), "round %d: the remote sees the disconnect", round);
	80	check(wait_for([&] { return bGone > gone; }, 30000), what);
	81	sleep_ms(500); // let the loop run its closing handles
	82	snprintf(what, sizeof(what), "round %d: both transports still running", round);
	83	check(a.is_running() && b.is_running(), what);
	84	}
	85
	86	a.stop();
	87	b.stop();
	88	printf("\n%s test_peer_lifecycle: %d/%d\n", failed ? "FAIL" : "PASS", passed, passed + failed);
	89	return failed ? 1 : 0;
	90	}

test/test_session_channel.cpp +6 -0

…	…	@@ -86,6 +86,12 @@ struct Node {
86	86
87	87	std::vector<uint8_t> topic_id;
88	88
	89	// Stop the transport BEFORE the members its callbacks use are destroyed. `t`
	90	// is declared first, so it would otherwise be destroyed last: its thread's
	91	// final stream teardown fired on_mux_gone into a map already freed (ASan:
	92	// heap-use-after-free in the gone lambda, on every exit path).
	93	~Node() { t.stop(); }
	94
89	95	void wire(const uint8_t *topic) {
90	96	topic_id.assign(topic, topic + 32);
91	97

hyperswarm-peer commits / 3c79be0

Changed files (5) +139 -3