From 364327876c14e384c9fd6c0cb449257df8848b7a Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Fri, 11 Sep 2026 20:51:33 +0100 Subject: [PATCH 1/2] fix(core): report the OS-assigned port from becomeCoordinator and addListener Both stored the literal port argument they were called with rather than reading the actual bound port back from the listener, exactly the bug already fixed for TlsTransport (3876df5) but never carried over to WireMeshTransport's own independent implementation. Only matters when called with port 0 for an OS-assigned free port -- listListeners() would silently report 0 instead of the real port. Added a shared listenerPort() helper (mirroring startDataServer's own already-correct pattern) and used it in both places. Migrated become-coordinator-actual-port.integration.test.ts's own regression test onto WireMeshTransport and extended it to cover addListener too. Also drops connect_accepted, connect_rejected, and pong from the MeshMessage union: none has had a real constructor since the substrate swap (connect_request's own accept/reject rides sendManageRequest's request/response round trip directly, and peer identity now comes from the TLS-authenticated connection itself, never a self-reported pong), so route()'s own defensive drop case for them is now just dead cases for variants that no longer exist. --- src/core/wire-mesh-transport.ts | 21 +++--- src/core/wire-protocol.ts | 5 +- ...oordinator-actual-port.integration.test.ts | 64 +++++++++++++------ 3 files changed, 59 insertions(+), 31 deletions(-) diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 425a0b4..723f73d 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -49,7 +49,7 @@ export const DOMAIN = "exadev.io/agent-comms-v1"; /** One verb for the entire MeshMessage union: this phase carries every message opaquely rather than modelling each arm as its own verb, which is core/room's own job once P3 gives this substrate real room semantics. */ export const FRAME_VERB = "exadev.io/agent-comms-v1:frame"; -/** No finer-grained authorisation model exists above "you're a TLS-authenticated member of this mesh" at this phase -- exactly today's TlsTransport model, which does no per-message authorisation either. */ +/** No finer-grained authorisation model exists above "you're an approved member of this mesh" at this phase -- once a session is past the quarantine gate (see handleAcceptedConnection), it's trusted for every message this domain carries. */ export const FRAME_SCOPE: Readonly = { kind: "agent-comms-mesh", }; @@ -72,6 +72,12 @@ function extractMessage(command: ManageCommand): MeshMessage | undefined { return isMeshMessage(message) ? message : undefined; } +/** Reads the port a listener actually bound, from its own reported address -- never the port it was asked to bind, which is 0 whenever the caller wanted the OS to assign a free one. Bookkeeping that stores the requested port instead silently reports 0 for every OS-assigned listener. */ +function listenerPort(listener: Readonly): number { + const port = listener.address.split(":").pop(); + return port === undefined ? 0 : Number(port); +} + // --------------------------------------------------------------------------- // Internal session bookkeeping // --------------------------------------------------------------------------- @@ -335,10 +341,8 @@ export class WireMeshTransport implements MeshTransport { this.events.onBecomeCoordinator(message.peerList); return; } - case "connect_request": - case "connect_accepted": - case "connect_rejected": { - // connect_request only ever reaches route() if a session was somehow promoted without going through consumeQuarantined's own handling of it -- can't happen given every requiresApproval accept path routes through consumeQuarantined first, kept here only so an unrecognised-in-context method fails closed rather than falling to the default onMessage case below. connect_accepted/connect_rejected are never constructed by this transport at all -- connectToRemote's own sendManageRequest outcome carries that meaning directly. + case "connect_request": { + // Only ever reaches route() if a session was somehow promoted without going through consumeQuarantined's own handling of it -- can't happen given every requiresApproval accept path routes through consumeQuarantined first, kept here only so an unrecognised-in-context method fails closed rather than falling to the default onMessage case below. return; } default: { @@ -383,8 +387,7 @@ export class WireMeshTransport implements MeshTransport { void this.handleAcceptedConnection(connection, undefined, true, false); }, ); - const port = this.dataListener.address.split(":").pop(); - this._dataPort = port === undefined ? 0 : Number(port); + this._dataPort = listenerPort(this.dataListener); } // ----------------------------------------------------------------------- @@ -444,7 +447,7 @@ export class WireMeshTransport implements MeshTransport { listener, policy: "full", host, - port, + port: listenerPort(listener), isDefault: true, }); this.defaultListenerId = id; @@ -634,7 +637,7 @@ export class WireMeshTransport implements MeshTransport { listener, policy, host, - port, + port: listenerPort(listener), isDefault: false, }); return id; diff --git a/src/core/wire-protocol.ts b/src/core/wire-protocol.ts index dd1c7cc..cd606a2 100644 --- a/src/core/wire-protocol.ts +++ b/src/core/wire-protocol.ts @@ -3,7 +3,7 @@ * * Transport-agnostic: carries the protocol contract between peers without * depending on net.Socket or any specific transport implementation. Any - * MeshTransport implementation (TlsTransport is the one production + * MeshTransport implementation (WireMeshTransport is the one production * transport today) uses these types. */ @@ -72,13 +72,10 @@ export type MeshMessage = name: string; fingerprint: string; } - | { method: "connect_accepted"; peerId: string; dataPort: number } - | { method: "connect_rejected"; peerId: string; reason: string } | { method: "peer_list"; peers: PeerInfo[] } | { method: "peer_joined"; peer: PeerInfo } | { method: "peer_left"; peerId: string } | { method: "become_coordinator"; peerList: PeerInfo[] } - | { method: "pong"; peerId: string } // Federation wire messages (coordinator-to-coordinator only) | { method: "fed_handshake"; meshId: string; name: string; version: string } | { method: "fed_ack"; meshId: string; name: string; version: string } diff --git a/src/test/become-coordinator-actual-port.integration.test.ts b/src/test/become-coordinator-actual-port.integration.test.ts index be240a7..ac4cb6a 100644 --- a/src/test/become-coordinator-actual-port.integration.test.ts +++ b/src/test/become-coordinator-actual-port.integration.test.ts @@ -1,13 +1,15 @@ /** - * becomeCoordinator actual-port regression test (#42) — becomeCoordinator(host, 0) must report the OS-assigned port it actually bound, not the literal 0 it was called with, on every transport whose listener bookkeeping goes through listListeners(). + * becomeCoordinator/addListener actual-port regression test (#42) — becomeCoordinator(host, 0) and addListener(host, 0, policy) must report the OS-assigned port they actually bound, not the literal 0 they were called with, on every transport whose listener bookkeeping goes through listListeners(). * - * Run: node dist/test/become-coordinator-actual-port.integration.test.js [test-name] With no argument, every scenario runs in order. + * Run directly with a specific scenario name as an argument, or with none to run every scenario in order. */ -import * as tls from "node:tls"; import * as assert from "node:assert/strict"; -import { TlsTransport } from "../core/tls-transport.js"; +import { createTlsTransport } from "@exadev/wire-mesh-core/adapters/tls-transport"; +import { acceptMeshSession } from "@exadev/wire-mesh-core/domain/mesh-session"; +import { WireMeshTransport, DOMAIN } from "../core/wire-mesh-transport.js"; import { generateIdentity } from "../core/identity.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; import type { TransportEvents } from "../core/transport.js"; function noopEvents(): TransportEvents { @@ -23,27 +25,51 @@ function noopEvents(): TransportEvents { }; } -async function testTlsReportsActualPort(): Promise { +/** Connects a real client session to the given port to confirm it's genuinely live, not merely non-zero. */ +async function connectAndClose(port: number): Promise { + const clientIdentity = generateIdentity(); + const clientTransport = createTlsTransport({ + certificatePem: clientIdentity.certificate, + privateKeyPem: clientIdentity.privateKey, + }); + const connection = await clientTransport.connect(`127.0.0.1:${String(port)}`); + const clientIdentityPort = await toIdentityPort(clientIdentity); + const session = await acceptMeshSession(connection, clientIdentityPort, [ + DOMAIN, + ]); + await session.close(); +} + +async function testBecomeCoordinatorReportsActualPort(): Promise { const identity = generateIdentity(); - const transport = new TlsTransport(noopEvents(), identity); + const transport = new WireMeshTransport(noopEvents(), identity); await transport.becomeCoordinator("127.0.0.1", 0); const [listener] = transport.listListeners(); assert.ok(listener); assert.notStrictEqual(listener.port, 0); - await new Promise((resolve, reject) => { - const socket = tls.connect( - { host: "127.0.0.1", port: listener.port, rejectUnauthorized: false }, - () => { - socket.destroy(); - resolve(); - }, - ); - socket.once("error", reject); - }); + await connectAndClose(listener.port); + console.log( + ` ✓ becomeCoordinator reported and bound the same port (${listener.port})`, + ); + + await transport.shutdown(); +} + +async function testAddListenerReportsActualPort(): Promise { + const identity = generateIdentity(); + const transport = new WireMeshTransport(noopEvents(), identity); + await transport.becomeCoordinator("127.0.0.1", 0); + + const id = await transport.addListener("127.0.0.1", 0, "observe"); + const added = transport.listListeners().find((l) => l.id === id); + assert.ok(added); + assert.notStrictEqual(added.port, 0); + + await connectAndClose(added.port); console.log( - ` ✓ TlsTransport reported and bound the same port (${listener.port})`, + ` ✓ addListener reported and bound the same port (${added.port})`, ); await transport.shutdown(); @@ -56,7 +82,9 @@ async function testTlsReportsActualPort(): Promise { const testName = process.argv[2]; const tests: Record Promise> = { - "tls-reports-actual-port": testTlsReportsActualPort, + "become-coordinator-reports-actual-port": + testBecomeCoordinatorReportsActualPort, + "add-listener-reports-actual-port": testAddListenerReportsActualPort, }; const selected = From 1ec34c714de2618df93dccf36ca760a5d694614d Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Fri, 11 Sep 2026 20:51:57 +0100 Subject: [PATCH 2/2] refactor(core): delete TlsTransport now that every bridge runs on WireMeshTransport TlsTransport was agent-comms' own hand-rolled MeshTransport implementation over X.509 certificates and newline-delimited JSON framing -- the substrate every bridge ran on before the P2 migration. With all six bridges cut over to WireMeshTransport (over @exadev/wire-mesh-core) and its own connection-approval quarantine gate closing the security gap that substrate swap introduced, nothing constructs a TlsTransport any more. test-transport.ts's wireTestTransport and wireWireMeshTestTransport wired two different transports onto a test MeshStore for exactly this migration period; with only one transport left, they collapse into one function (kept the wireTestTransport name, the one every existing test already called). Every test that built its own peer directly against TlsTransport (broadcast-window, downtime-replay.integration, identity-restart) now does the same against WireMeshTransport, with fingerprint-based identity comparisons swapped for deviceId, matching the canonical identity createBridgeMesh already uses in production. listener-policy.integration.test.ts's one test that spoke TlsTransport's own wire format directly (a raw tls.connect() writing newline-delimited JSON) is rewritten against a real WireMeshTransport client session instead, and no longer reaches into MeshStore's internals through an unsafe cast to observe the result -- it constructs its own transport directly with a hand-rolled TransportEvents object instead, the same pattern become-coordinator-actual-port.integration.test.ts already used. peer-id-verification.integration.test.ts is deleted outright rather than migrated: the vulnerability class it guarded against (a peer's self-reported peerId not matching the certificate it authenticated with) cannot occur under WireMeshTransport by construction, since peer identity is never self-reported in the first place -- it's read directly from the TLS-authenticated connection's own peerDeviceId. wire-mesh-transport.integration.test.ts and wire-mesh-transport-approval.integration.test.ts were written during the migration specifically to prove WireMeshTransport reached parity with TlsTransport's own behaviour. Now that wireTestTransport wires WireMeshTransport for every test, they duplicate mesh-e2e.integration.test.ts and approval.integration.test.ts/listener-policy.integration.test.ts scenario-for-scenario. Deleted, after moving the one genuinely unique test in the approval file over: the quarantine-gate regression test proving a forged state_update from an unapproved connection is refused. Fixed along the way: approval.integration.test.ts's own "mesh_pending lists pending connections" test shut down storeB immediately once its own assertions were satisfied, without waiting for storeB's own fire-and-forget connectToRemote continuation (the mesh-formation handshake) to actually settle. Under WireMeshTransport's real round-trip timing this let the test's own shutdown race ahead of that continuation, which then tried to track a session into an already-shut-down transport and never closed it -- the same convergence wait "accept establishes the peer connection" already uses fixes it. --- src/core/mesh-store.ts | 4 +- src/core/tls-transport.ts | 977 ------------------ src/test/approval.integration.test.ts | 84 +- src/test/bridge-mesh.test.ts | 2 +- src/test/broadcast-window.integration.test.ts | 13 +- src/test/downtime-replay.integration.test.ts | 14 +- src/test/identity-restart.integration.test.ts | 16 +- src/test/listener-policy.integration.test.ts | 117 +-- src/test/mesh-smoke.runner.ts | 9 +- .../peer-id-verification.integration.test.ts | 246 ----- src/test/test-transport.ts | 19 +- src/test/tls-transport.integration.test.ts | 198 ---- ...esh-transport-approval.integration.test.ts | 290 ------ .../wire-mesh-transport.integration.test.ts | 130 --- 14 files changed, 177 insertions(+), 1942 deletions(-) delete mode 100644 src/core/tls-transport.ts delete mode 100644 src/test/peer-id-verification.integration.test.ts delete mode 100644 src/test/tls-transport.integration.test.ts delete mode 100644 src/test/wire-mesh-transport-approval.integration.test.ts delete mode 100644 src/test/wire-mesh-transport.integration.test.ts diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 382eeb7..c686c7d 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -7,7 +7,7 @@ * Delivery events are pushed directly over the transport — no polling, * no filesystem. * - * Transport is set via setTransport() (e.g. TlsTransport for encrypted + * Transport is set via setTransport() (e.g. WireMeshTransport for encrypted * connections) before init() or any other transport-using method is called * -- there is no default, since every real bridge builds its own transport * from this store's own events getter, which needs the store to already @@ -181,7 +181,7 @@ export class MeshStore implements CommsStore { ); } - /** Sets the transport (e.g. TlsTransport for encrypted connections). Must be called before init() or any other transport-using method. */ + /** Sets the transport (e.g. WireMeshTransport for encrypted connections). Must be called before init() or any other transport-using method. */ setTransport(transport: MeshTransport): void { this.transport = transport; } diff --git a/src/core/tls-transport.ts b/src/core/tls-transport.ts deleted file mode 100644 index acb221b..0000000 --- a/src/core/tls-transport.ts +++ /dev/null @@ -1,977 +0,0 @@ -/** - * TlsTransport — TLS-encrypted transport for the peer mesh, the one - * production MeshTransport implementation. - * - * All connections are wrapped in TLS with certificate fingerprint - * authentication. Each peer generates an ECDSA P-256 keypair and self-signed - * X.509 certificate on startup. The peer ID is the certificate fingerprint — - * verifying a peer's identity is simply checking that the presented - * certificate's fingerprint matches the known ID. - * - * This is the Syncthing trust model: no CA, no PKI, just certificate pinning. - * Overlay networks (Tailscale, WireGuard) are still recommended for NAT - * traversal but are not required for security — TLS handles encryption and - * authentication at the protocol level. - * - * Implements the same MeshTransport interface every transport does. Same - * wire protocol, same event callbacks — MeshStore only ever sees the - * interface, never a concrete transport. - */ - -import * as net from "node:net"; -import * as tls from "node:tls"; -import { encode, isMeshMessage, MessageBuffer } from "./wire-protocol.js"; -import { attachSocketHandshake } from "./handshake.js"; -import type { MeshMessage, PeerInfo } from "./wire-protocol.js"; -import type { - ConnectionHandle, - ListenerInfo, - ListenerPolicy, - TransportEvents, -} from "./transport.js"; -import type { PeerIdentity } from "./identity.js"; -import { fingerprintDer } from "./identity.js"; -import { nanoid } from "./nanoid.js"; - -// --------------------------------------------------------------------------- -// Constants -// --------------------------------------------------------------------------- - -const COORDINATOR_HOST = "127.0.0.1"; -const CONNECT_TIMEOUT_MS = 1000; - -/** - * Wrap tls.createServer with a retry for intermittent OpenSSL ASN.1 races - * that occur when parallel test workers create TLS servers simultaneously. - * Up to 3 attempts before propagating the error. - */ -function retryCreateTlsServer( - options: tls.TlsOptions, - callback: (socket: tls.TLSSocket) => void, -): tls.Server { - let attempts = 0; - const maxAttempts = 3; - const tryCreate = (): tls.Server => { - attempts++; - try { - return tls.createServer(options, callback); - } catch { - if (attempts < maxAttempts) return tryCreate(); - throw new Error( - `tls.createServer failed after ${String(attempts)} attempts`, - ); - } - }; - return tryCreate(); -} - -// --------------------------------------------------------------------------- -// Async socket write helper (not exported) -// --------------------------------------------------------------------------- - -/** - * Safety valve so a dial that never completes cannot grow its outbound queue - * unbounded; oldest entries are dropped first (#23). - */ -const MAX_PENDING_PER_PEER = 100; - -function writeAsync( - socket: net.Socket | tls.TLSSocket, - data: string, -): Promise { - return new Promise((resolve, reject) => { - if (socket.destroyed) { - resolve(); - return; - } - socket.write(data, "utf-8", (err) => { - if (err) reject(err); - else resolve(); - }); - }); -} - -/** - * writeAsync(), but a failed write (the peer disconnected, or its own shutdown destroyed the socket mid-write) resolves instead of rejecting -- cleanup is already handled by the socket's own close/error listeners wherever it's registered, exactly the same ordinary-and-expected outcome broadcast() already treats this way. A caller that genuinely needs to know whether the write landed gets that back as a boolean rather than an exception, since a write racing a legitimate connection teardown is not itself a program error and must never surface as an unhandled rejection. - */ -async function writeBestEffort( - socket: net.Socket | tls.TLSSocket, - data: string, -): Promise { - return writeAsync(socket, data).then( - () => true, - () => false, - ); -} - -// --------------------------------------------------------------------------- -// TlsTransport -// --------------------------------------------------------------------------- - -export class TlsTransport { - // -- Data server (accepts incoming peer data connections over TLS) -- - private dataServer: tls.Server | undefined; - private _dataPort = 0; - - // -- Coordinator listeners (multiple adapters) -- - private coordinatorListeners = new Map< - string, - { - server: tls.Server; - policy: ListenerPolicy; - host: string; - port: number; - isDefault: boolean; - } - >(); - /** The default localhost listener ID (set once during becomeCoordinator). */ - private defaultListenerId: string | undefined; - private _isCoordinator = false; - - // -- Coordinator client socket (TLS connection to the coordinator) -- - private coordinatorSocket: tls.TLSSocket | undefined; - - // -- Messages queued for peers whose dial is still in flight (#23) -- - private pendingOutbound = new Map(); - - // -- Coordinator introduction handshake (resolved on the peer list) -- - private resolveCoordinatorHandshake: (() => void) | undefined; - private coordinatorHandshakeTimer: ReturnType | undefined; - - // -- Peer data connections (peer ID → socket + buffer) -- - private peerConnections = new Map< - string, - { socket: tls.TLSSocket; buffer: MessageBuffer } - >(); - - // -- All sockets accepted by the data server (for shutdown cleanup) -- - private dataServerSockets = new Set(); - - // -- All sockets accepted by the coordinator server (for shutdown cleanup) -- - private coordinatorServerSockets = new Set(); - - // -- Coordinator introduction connections (handle ID → socket) -- - private introConnections = new Map(); - - // -- Pending connections awaiting approval (handle ID → socket + request info) -- - private pendingConnections = new Map< - string, - { - socket: tls.TLSSocket; - peerId: string; - dataPort: number; - name: string; - fingerprint: string; - policy: ListenerPolicy; - } - >(); - - // -- Outbound connectToPeer sockets still mid-dial, not yet in peerConnections or already failed — tracked so shutdown() can destroy an in-flight dial rather than leaving it running in the background indefinitely, long after this transport is gone -- - private pendingConnectSockets = new Set(); - - // -- Peers this side has dialled out to (in flight or established) — deliberately separate from peerConnections, which also holds sockets THIS side merely accepted an inbound dial from and identified via pong. A peer can legitimately be dialling us at the same time we need to dial them (mesh formation establishes one connection in each direction so each side's acceptor can push its own state), so connectToPeer's own "don't dial twice" guard must not be satisfied by an inbound connection that happens to share the same peer ID. -- - private outboundDials = new Set(); - - // -- Shutdown sentinel — prevents callbacks after shutdown() -- - private shutDown = false; - - // -- This peer's ID (set during connectToCoordinator or becomeCoordinator) -- - private _peerId = ""; - - private readonly events: TransportEvents; - private readonly identity: PeerIdentity; - - constructor(events: TransportEvents, identity: PeerIdentity) { - this.events = events; - this.identity = identity; - } - - // -- Public getters for interface properties -- - - get dataPort(): number { - return this._dataPort; - } - - get isCoordinator(): boolean { - return this._isCoordinator; - } - - get hasCoordinatorConnection(): boolean { - return ( - this.coordinatorSocket !== undefined && !this.coordinatorSocket.destroyed - ); - } - - // ----------------------------------------------------------------------- - // TLS options - // ----------------------------------------------------------------------- - - private get tlsOptions(): tls.TlsOptions { - return { - key: this.identity.privateKey, - cert: this.identity.certificate, - // Do not reject unauthorized — we do our own fingerprint verification after the TLS handshake completes, in verifyClaimedPeerId(). - rejectUnauthorized: false, - requestCert: true, - }; - } - - private get connectOptions(): tls.ConnectionOptions { - return { - key: this.identity.privateKey, - cert: this.identity.certificate, - rejectUnauthorized: false, - // Don't verify server cert via CA — verify via fingerprint pinning - }; - } - - // ----------------------------------------------------------------------- - // Peer identity verification - // ----------------------------------------------------------------------- - - /** - * Verify a connected socket's presented certificate fingerprint matches the peer ID it claims via `introduce`/`pong` in the wire protocol, destroying the socket and returning false on any mismatch (including no certificate presented at all). - * - * Peer IDs are minted as the fingerprint of the peer's own certificate (identity.ts's `generateIdentity()`, wired up by every bridge as `store.peerId = identity.fingerprint`), so a claimed peer ID that doesn't match the certificate actually presented on this connection means the socket is not who it says it is — regardless of what it typed into the wire message. - */ - private verifyClaimedPeerId( - socket: tls.TLSSocket, - claimedPeerId: string, - ): boolean { - const cert = socket.getPeerCertificate(); - // Node's types declare every PeerCertificate field non-optional, but the documented runtime behaviour when the peer presents no certificate at all is an empty object — not null/undefined, and not a Buffer-typed `raw`. Detect that real shape rather than trusting the declared type. - if (Object.keys(cert).length === 0) { - socket.destroy(); - this.events.onError?.( - new Error( - `Rejected connection claiming peer ID ${claimedPeerId}: no certificate presented`, - ), - ); - return false; - } - const actualFingerprint = fingerprintDer(cert.raw); - if (actualFingerprint !== claimedPeerId) { - socket.destroy(); - this.events.onError?.( - new Error( - `Rejected connection claiming peer ID ${claimedPeerId}: presented certificate fingerprint is ${actualFingerprint}`, - ), - ); - return false; - } - return true; - } - - // ----------------------------------------------------------------------- - // MeshTransport — Data server - // ----------------------------------------------------------------------- - - async startDataServer(): Promise { - await new Promise((resolve, reject) => { - this.dataServer = retryCreateTlsServer(this.tlsOptions, (socket) => { - this.handleIncomingDataConnection(socket); - }); - this.dataServer.listen(0, COORDINATOR_HOST, () => { - const addr = this.dataServer?.address(); - if (typeof addr === "object" && addr !== null) { - this._dataPort = addr.port; - } - resolve(); - }); - this.dataServer.on("error", reject); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Coordinator connection (client side) - // ----------------------------------------------------------------------- - - async connectToCoordinator( - host: string, - port: number, - peerId: string, - localDataPort: number, - ): Promise { - this._peerId = peerId; - await new Promise((resolve, reject) => { - const socket = tls.connect({ ...this.connectOptions, host, port }, () => { - this.coordinatorSocket = socket; - - // Wire up the protocol handshake first (client role) so the introduction below lands after it on the wire, not before. - const buffer = new MessageBuffer(); - attachSocketHandshake( - socket, - "client", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (isMeshMessage(item)) { - this.dispatchCoordinatorClientMessage(item); - } - } - }, - (error) => this.events.onError?.(error), - ); - - // Send introduction - const intro: MeshMessage = { - method: "introduce", - peerId, - dataPort: localDataPort, - }; - socket.write(encode(intro)); - - socket.on("error", () => { - /* ignore late errors on coordinator connection */ - }); - - clearTimeout(timer); - // Resolve on the coordinator's peer list rather than on sending the - // introduction: MeshStore.init() then returns only after the post-join - // peer dials have started, so the first broadcasts are queued for the - // dialling peers instead of dropped (#23). A timeout keeps today's - // degraded behaviour for a coordinator that never answers. - this.resolveCoordinatorHandshake = resolve; - this.coordinatorHandshakeTimer = setTimeout(() => { - this.resolveCoordinatorHandshake = undefined; - resolve(); - }, CONNECT_TIMEOUT_MS); - }); - - const timer = setTimeout(() => { - socket.destroy(); - reject(new Error("Coordinator connection timeout")); - }, CONNECT_TIMEOUT_MS); - - socket.on("error", (err) => { - clearTimeout(timer); - socket.destroy(); - reject(err); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Connect to remote coordinator with approval - // ----------------------------------------------------------------------- - - async connectToRemote( - host: string, - port: number, - peerId: string, - localDataPort: number, - name: string, - fingerprint: string, - ): Promise { - this._peerId = peerId; - await new Promise((resolve, reject) => { - const socket = tls.connect({ ...this.connectOptions, host, port }, () => { - this.coordinatorSocket = socket; - - clearTimeout(timer); - - // Wire up the protocol handshake first (client role) so connect_request below lands after it on the wire, not before. - const buffer = new MessageBuffer(); - let approved = false; - attachSocketHandshake( - socket, - "client", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (!isMeshMessage(item)) continue; - - if (!approved) { - if (item.method === "connect_accepted") { - approved = true; - resolve(); - } else if (item.method === "connect_rejected") { - socket.destroy(); - reject(new Error(`Connection rejected: ${item.reason}`)); - return; - } - } else { - this.dispatchCoordinatorClientMessage(item); - } - } - }, - (error) => this.events.onError?.(error), - ); - - // Send connect_request instead of introduce - const req: MeshMessage = { - method: "connect_request", - peerId, - dataPort: localDataPort, - name, - fingerprint, - }; - socket.write(encode(req)); - socket.on("error", () => { - /* ignore late errors on coordinator connection */ - }); - }); - - const timer = setTimeout(() => { - socket.destroy(); - reject(new Error("Remote connection timeout")); - }, CONNECT_TIMEOUT_MS); - - socket.on("error", (err) => { - clearTimeout(timer); - socket.destroy(); - reject(err); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Become coordinator (server side) - // ----------------------------------------------------------------------- - - async becomeCoordinator(host: string, port: number): Promise { - const id = nanoid(8); - await new Promise((resolve, reject) => { - const server = retryCreateTlsServer(this.tlsOptions, (socket) => { - this.handleCoordinatorServerConnection(socket, "full"); - }); - - server.listen(port, host, () => { - const addr = server.address(); - const actualPort = - typeof addr === "object" && addr !== null ? addr.port : port; - this._isCoordinator = true; - this.coordinatorListeners.set(id, { - server, - policy: "full", - host, - port: actualPort, - isDefault: true, - }); - this.defaultListenerId = id; - resolve(); - }); - - server.on("error", (err: unknown) => { - reject(err instanceof Error ? err : new Error(String(err))); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Multi-listener management - // ----------------------------------------------------------------------- - - async addListener( - host: string, - port: number, - policy: ListenerPolicy, - ): Promise { - if (!this._isCoordinator) { - throw new Error("Only the coordinator can add listeners"); - } - - const id = nanoid(8); - await new Promise((resolve, reject) => { - const server = retryCreateTlsServer(this.tlsOptions, (socket) => { - this.handleCoordinatorServerConnection(socket, policy); - }); - - server.listen(port, host, () => { - const addr = server.address(); - const actualPort = - typeof addr === "object" && addr !== null ? addr.port : port; - this.coordinatorListeners.set(id, { - server, - policy, - host, - port: actualPort, - isDefault: false, - }); - resolve(); - }); - - server.on("error", (err: unknown) => { - reject(err instanceof Error ? err : new Error(String(err))); - }); - }); - return id; - } - - removeListener(id: string): Promise { - const listener = this.coordinatorListeners.get(id); - if (!listener) { - throw new Error(`Listener ${id} not found`); - } - if (listener.isDefault) { - throw new Error("Cannot remove the default localhost listener"); - } - - listener.server.unref(); - listener.server.close(); - this.coordinatorListeners.delete(id); - - return Promise.resolve(); - } - - listListeners(): ListenerInfo[] { - const result: ListenerInfo[] = []; - for (const [id, listener] of this.coordinatorListeners) { - result.push({ - id, - host: listener.host, - port: listener.port, - policy: listener.policy, - isDefault: listener.isDefault, - }); - } - return result; - } - - // ----------------------------------------------------------------------- - // MeshTransport — Connect to a peer's data server - // ----------------------------------------------------------------------- - - async connectToPeer(peer: PeerInfo, ownPeerId: string): Promise { - if (this.shutDown || this.outboundDials.has(peer.id)) return; - this.outboundDials.add(peer.id); - - // Queue broadcasts until the connection registers: messages sent in the dial window previously had nowhere to go and were silently dropped. - this.pendingOutbound.set(peer.id, this.pendingOutbound.get(peer.id) ?? []); - - await new Promise((resolve) => { - const socket = tls.connect( - { ...this.connectOptions, host: COORDINATOR_HOST, port: peer.port }, - () => { - this.pendingConnectSockets.delete(socket); - if (this.shutDown) { - this.pendingOutbound.delete(peer.id); - socket.destroy(); - resolve(); - return; - } - if (!this.verifyClaimedPeerId(socket, peer.id)) { - this.pendingOutbound.delete(peer.id); - this.outboundDials.delete(peer.id); - resolve(); - return; - } - - const buffer = new MessageBuffer(); - this.peerConnections.set(peer.id, { socket, buffer }); - - // Wire up the protocol handshake first (client role) so everything below lands after it on the wire, not before. - attachSocketHandshake( - socket, - "client", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (isMeshMessage(item)) { - const handle: ConnectionHandle = { id: peer.id }; - this.dispatchDataMessage(handle, item); - } - } - }, - (error) => this.events.onError?.(error), - ); - - // Identify ourselves - const pong: MeshMessage = { method: "pong", peerId: ownPeerId }; - socket.write(encode(pong)); - - void this.flushPending(peer.id, socket); - - resolve(); - }, - ); - // Tracked from the moment tls.connect() returns (before the connect callback above ever fires) so shutdown() can destroy a dial that's still mid-handshake — otherwise it keeps running in the background with nothing to cancel it, eventually erroring out (or not) long after this transport, and the test or process that started it, are gone. - this.pendingConnectSockets.add(socket); - - const handle: ConnectionHandle = { id: peer.id }; - let disconnected = false; - - const onDisconnect = (): void => { - if (disconnected) return; - disconnected = true; - this.pendingConnectSockets.delete(socket); - this.outboundDials.delete(peer.id); - // Only remove peerConnections' entry if it still points at THIS socket -- the peer may also have dialled us independently (a second, inbound connection for the same peer ID), and that connection's own close must never delete a live entry this one already replaced, or vice versa. - const current = this.peerConnections.get(peer.id); - const wasConnected = current?.socket === socket; - if (wasConnected) this.peerConnections.delete(peer.id); - if (wasConnected && !this.shutDown) { - this.events.onPeerDisconnected(handle); - } - }; - - socket.on("close", onDisconnect); - socket.on("error", (err) => { - if (!this.shutDown) { - this.events.onError?.( - new Error( - `connectToPeer(${peer.id}, port ${String(peer.port)}) socket error: ${err.message}`, - ), - ); - } - this.pendingOutbound.delete(peer.id); - onDisconnect(); - socket.destroy(); - resolve(); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Send / broadcast - // ----------------------------------------------------------------------- - - async send(handle: ConnectionHandle, message: MeshMessage): Promise { - // Check data connections first - const peerConn = this.peerConnections.get(handle.id); - if (peerConn) { - await writeBestEffort(peerConn.socket, encode(message)); - return; - } - - // Check coordinator introduction connections - const introSocket = this.introConnections.get(handle.id); - if (introSocket) { - await writeBestEffort(introSocket, encode(message)); - return; - } - - throw new Error(`No connection for handle ${handle.id}`); - } - - async acceptConnection(handle: ConnectionHandle): Promise { - const pending = this.pendingConnections.get(handle.id); - if (!pending) { - throw new Error(`No pending connection for handle ${handle.id}`); - } - this.pendingConnections.delete(handle.id); - - const { socket, peerId, dataPort, policy } = pending; - - // Move to introConnections so send() can reach this peer - this.introConnections.set(peerId, socket); - - // Send acceptance to the connecting peer -- best-effort: the peer having already vanished (or this side's own concurrent shutdown destroying the socket mid-write) is an ordinary disconnect, not a program error, and must not surface as an unhandled rejection once onIntroduction's own downstream state-sync work below is already in flight. - const accepted: MeshMessage = { - method: "connect_accepted", - peerId: this._peerId, - dataPort: this._dataPort, - }; - await writeBestEffort(socket, encode(accepted)); - - // Fire onIntroduction so MeshStore processes the new peer normally - const connHandle: ConnectionHandle = { id: peerId, policy }; - this.events.onIntroduction(connHandle, { peerId, dataPort }); - } - - async rejectConnection( - handle: ConnectionHandle, - reason: string, - ): Promise { - const pending = this.pendingConnections.get(handle.id); - if (!pending) { - throw new Error(`No pending connection for handle ${handle.id}`); - } - this.pendingConnections.delete(handle.id); - - const { socket } = pending; - - const rejected: MeshMessage = { - method: "connect_rejected", - peerId: handle.id, - reason, - }; - await writeBestEffort(socket, encode(rejected)); - socket.destroy(); - } - - async broadcast(message: MeshMessage): Promise { - const data = encode(message); - const writes: Promise[] = []; - for (const [, peer] of this.peerConnections) { - writes.push(writeBestEffort(peer.socket, data).then(() => undefined)); - } - for (const queue of this.pendingOutbound.values()) { - queue.push(message); - if (queue.length > MAX_PENDING_PER_PEER) queue.shift(); - } - await Promise.all(writes); - } - - /** Send messages queued while the peer's connection was being dialled. */ - private async flushPending( - peerId: string, - socket: tls.TLSSocket, - ): Promise { - const queue = this.pendingOutbound.get(peerId); - this.pendingOutbound.delete(peerId); - if (queue === undefined) return; - for (const message of queue) { - const sent = await writeBestEffort(socket, encode(message)); - if (!sent) return; // connection is dying; close/error listeners clean up - } - } - - // ----------------------------------------------------------------------- - // MeshTransport — Shutdown / unref - // ----------------------------------------------------------------------- - - shutdown(): Promise { - this.shutDown = true; - if (this.coordinatorHandshakeTimer !== undefined) { - clearTimeout(this.coordinatorHandshakeTimer); - this.coordinatorHandshakeTimer = undefined; - } - this.resolveCoordinatorHandshake = undefined; - - // Destroy the coordinator client socket - this.coordinatorSocket?.unref(); - this.coordinatorSocket?.destroy(); - this.coordinatorSocket = undefined; - - // Destroy all identified peer connections - for (const [, peer] of this.peerConnections) { - peer.socket.unref(); - peer.socket.destroy(); - } - this.peerConnections.clear(); - - // Destroy all data server accepted sockets (including unidentified) - for (const socket of this.dataServerSockets) { - socket.unref(); - socket.destroy(); - } - this.dataServerSockets.clear(); - - // Destroy all coordinator server accepted sockets - for (const socket of this.coordinatorServerSockets) { - socket.unref(); - socket.destroy(); - } - this.coordinatorServerSockets.clear(); - - // Clear introduction connection tracking - this.introConnections.clear(); - - // Clear pending connections - for (const [, pending] of this.pendingConnections) { - pending.socket.unref(); - pending.socket.destroy(); - } - this.pendingConnections.clear(); - - // Destroy any outbound connectToPeer dials still mid-flight — without this they keep retrying/erroring out in the background indefinitely, long after this transport (and whatever test or process started it) is gone. - for (const socket of this.pendingConnectSockets) { - socket.unref(); - socket.destroy(); - } - this.pendingConnectSockets.clear(); - - // Close data server - this.dataServer?.unref(); - this.dataServer?.close(); - this.dataServer = undefined; - - // Close coordinator listener servers - for (const [, listener] of this.coordinatorListeners) { - listener.server.unref(); - listener.server.close(); - } - this.coordinatorListeners.clear(); - this.defaultListenerId = undefined; - - this._isCoordinator = false; - - return Promise.resolve(); - } - - unref(): void { - this.dataServer?.unref(); - for (const [, listener] of this.coordinatorListeners) { - listener.server.unref(); - } - this.coordinatorSocket?.unref(); - } - - // ----------------------------------------------------------------------- - // Internal — Coordinator client message dispatch - // ----------------------------------------------------------------------- - - private dispatchCoordinatorClientMessage(msg: MeshMessage): void { - if (this.shutDown) return; - - if (msg.method === "peer_list") { - // Fire onPeerList first: it starts the post-join dials (and their - // broadcast queues) before init() resolves. - this.events.onPeerList(msg.peers); - this.completeCoordinatorHandshake(); - } else if (msg.method === "peer_joined") { - this.events.onPeerJoined(msg.peer); - } else if (msg.method === "become_coordinator") { - this.events.onBecomeCoordinator(msg.peerList); - } - } - - /** Complete connectToCoordinator's handshake after the peer list arrives. */ - private completeCoordinatorHandshake(): void { - if (this.coordinatorHandshakeTimer !== undefined) { - clearTimeout(this.coordinatorHandshakeTimer); - this.coordinatorHandshakeTimer = undefined; - } - this.resolveCoordinatorHandshake?.(); - this.resolveCoordinatorHandshake = undefined; - } - - // ----------------------------------------------------------------------- - // Internal — Data message dispatch - // ----------------------------------------------------------------------- - - private dispatchDataMessage( - handle: ConnectionHandle, - msg: MeshMessage, - ): void { - if (this.shutDown) return; - - if (msg.method === "become_coordinator") { - this.events.onBecomeCoordinator(msg.peerList); - } else { - this.events.onMessage(handle, msg); - } - } - - // ----------------------------------------------------------------------- - // Internal — Coordinator server connection handling - // ----------------------------------------------------------------------- - - /** - * Accepts a new connection on a coordinator server. Reads - * introduce messages and fires onIntroduction so MeshStore can - * respond with the peer list and broadcast the arrival. - * The connection is tagged with the listener's policy. - */ - private handleCoordinatorServerConnection( - socket: tls.TLSSocket, - policy: ListenerPolicy, - ): void { - if (this.shutDown) { - socket.destroy(); - return; - } - - this.coordinatorServerSockets.add(socket); - socket.on("close", () => this.coordinatorServerSockets.delete(socket)); - socket.on("error", () => this.coordinatorServerSockets.delete(socket)); - - const buffer = new MessageBuffer(); - attachSocketHandshake( - socket, - "server", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (!isMeshMessage(item)) continue; - - if (item.method === "introduce") { - if (!this.verifyClaimedPeerId(socket, item.peerId)) continue; - const handle: ConnectionHandle = { id: item.peerId, policy }; - this.introConnections.set(handle.id, socket); - this.events.onIntroduction(handle, { - peerId: item.peerId, - dataPort: item.dataPort, - }); - } else if (item.method === "connect_request") { - const handle: ConnectionHandle = { id: item.peerId, policy }; - this.pendingConnections.set(handle.id, { - socket, - peerId: item.peerId, - dataPort: item.dataPort, - name: item.name, - fingerprint: item.fingerprint, - policy, - }); - this.events.onConnectionRequest(handle, { - peerId: item.peerId, - dataPort: item.dataPort, - name: item.name, - fingerprint: item.fingerprint, - }); - } - } - }, - (error) => this.events.onError?.(error), - ); - } - - // ----------------------------------------------------------------------- - // Internal — Incoming data connection handling - // ----------------------------------------------------------------------- - - private handleIncomingDataConnection(socket: tls.TLSSocket): void { - if (this.shutDown) { - socket.destroy(); - return; - } - - this.dataServerSockets.add(socket); - socket.on("close", () => this.dataServerSockets.delete(socket)); - - const buffer = new MessageBuffer(); - let remotePeerId: string | undefined; - let disconnected = false; - - attachSocketHandshake( - socket, - "server", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (isMeshMessage(item)) { - if (item.method === "pong") { - const peerId = item.peerId; - if (!this.verifyClaimedPeerId(socket, peerId)) continue; - remotePeerId = peerId; - if (!this.peerConnections.has(peerId)) { - this.peerConnections.set(peerId, { socket, buffer }); - } - void this.flushPending(peerId, socket); - const handle: ConnectionHandle = { id: peerId }; - const info: PeerInfo = { - id: peerId, - port: 0, - startedAt: new Date().toISOString(), - }; - this.events.onPeerConnected(handle, info); - } else if (remotePeerId !== undefined) { - const handle: ConnectionHandle = { id: remotePeerId }; - this.dispatchDataMessage(handle, item); - } - } - } - }, - (error) => this.events.onError?.(error), - ); - - const onDisconnect = (): void => { - if (disconnected) return; - disconnected = true; - if (remotePeerId !== undefined) { - // Only remove peerConnections' entry if it still points at THIS socket -- an outbound dial to the same peer ID can register its own, separate connection there, and this accepted socket closing must never delete that live entry. - const current = this.peerConnections.get(remotePeerId); - if (current?.socket === socket) { - this.peerConnections.delete(remotePeerId); - } - if (!this.shutDown) { - this.events.onPeerDisconnected({ id: remotePeerId }); - } - } - }; - - socket.on("close", onDisconnect); - socket.on("error", onDisconnect); - } -} diff --git a/src/test/approval.integration.test.ts b/src/test/approval.integration.test.ts index 02ab221..877c8f4 100644 --- a/src/test/approval.integration.test.ts +++ b/src/test/approval.integration.test.ts @@ -9,10 +9,19 @@ import * as net from "node:net"; import * as assert from "node:assert/strict"; import { test, describe } from "node:test"; +import { createTlsTransport } from "@exadev/wire-mesh-core/adapters/tls-transport"; +import { acceptMeshSession } from "@exadev/wire-mesh-core/domain/mesh-session"; import { MeshStore } from "../core/mesh-store.js"; import { CommsTool } from "../core/tool.js"; import { buildAction } from "../core/bridge.js"; -import type { DeliveryEvent } from "../core/types.js"; +import { generateIdentity } from "../core/identity.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; +import { + DOMAIN, + FRAME_SCOPE, + buildCommand, +} from "../core/wire-mesh-transport.js"; +import type { DeliveryEvent, AgentIdentity } from "../core/types.js"; import { waitFor, wireTestTransport } from "./test-transport.js"; /** Find a free port on localhost by binding to port 0. */ @@ -338,6 +347,12 @@ describe("connection approval", () => { () => storeA.listPendingConnections().length === 0, "pending connection is cleared after accept", ); + + // connectToRemote (storeB) and acceptConnection's own introduction handling (storeA) both continue asynchronously after this point (state_sync's own connectToPeer round trip) -- shutting down before that settles races storeB's still-in-flight session setup against its own teardown, leaving a session that gets tracked into an already-shut-down transport and never closed. Wait for real convergence first, the same way "accept establishes the peer connection" above already does. + await waitFor( + () => storeB.serialise().agents[storeA.peerId] !== undefined, + "connector sees the coordinator agent", + ); } finally { await storeB.shutdown(); await storeA.shutdown(); @@ -573,4 +588,71 @@ describe("connection approval", () => { await storeA.shutdown(); } }); + + void test("a message other than introduce/connect_request from an unapproved connection is refused, not routed", async () => { + const portA = await uniquePort(); + const storeA = new MeshStore(portA); + wireTestTransport(storeA); + try { + await storeA.init(); + await storeA.registerAgent({ + name: "coordinator", + harness: "test", + cwd: "/test/a", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // A hostile client that completes a TLS handshake (proving only which key it holds, not that a human approved it) and skips connect_request entirely, going straight for a forged state_update. + const attackerIdentity = generateIdentity(); + const attackerTransport = createTlsTransport({ + certificatePem: attackerIdentity.certificate, + privateKeyPem: attackerIdentity.privateKey, + }); + const connection = await attackerTransport.connect( + `127.0.0.1:${String(portA)}`, + ); + const attackerIdentityPort = await toIdentityPort(attackerIdentity); + const session = await acceptMeshSession( + connection, + attackerIdentityPort, + [DOMAIN], + ); + + const forgedAgent: AgentIdentity = { + id: "forged-attacker-agent", + version: 1, + name: "forged", + harness: "test", + cwd: "/forged", + pid: 1, + startedAt: new Date().toISOString(), + visibility: "visible", + status: "active", + tags: [], + subscribedRooms: [], + }; + const outcome = await session.sendManageRequest( + buildCommand({ + method: "state_update", + patch: { type: "agent_upsert", agent: forgedAgent }, + }), + FRAME_SCOPE, + ); + + assert.equal( + outcome.result, + "error", + "an unapproved connection's message must be refused, not routed", + ); + assert.equal( + storeA.serialise().agents[forgedAgent.id], + undefined, + "a forged state_update from an unapproved connection must never reach mesh-store's own state", + ); + } finally { + await storeA.shutdown(); + } + }); }); diff --git a/src/test/bridge-mesh.test.ts b/src/test/bridge-mesh.test.ts index c2ca148..329111f 100644 --- a/src/test/bridge-mesh.test.ts +++ b/src/test/bridge-mesh.test.ts @@ -37,7 +37,7 @@ void test("createBridgeMesh sets peerId to deviceIdToHex(identity.deviceId), not } }); -void test("createBridgeMesh wires a WireMeshTransport, not TlsTransport", async () => { +void test("createBridgeMesh wires a WireMeshTransport", async () => { const slot = tempSlot("test-harness"); const { store } = createBridgeMesh(slot); try { diff --git a/src/test/broadcast-window.integration.test.ts b/src/test/broadcast-window.integration.test.ts index 7c17eac..30c0d6e 100644 --- a/src/test/broadcast-window.integration.test.ts +++ b/src/test/broadcast-window.integration.test.ts @@ -1,23 +1,24 @@ /** - * Integration test for issue #23: state patches broadcast before the TLS data connections are established must not be silently lost. + * Integration test for issue #23: state patches broadcast before the mesh data connections are established must not be silently lost. * - * Every production bridge calls registerAgent() immediately after store.init() returns, while the fire-and-forget peer dials are still in flight. Broadcasts landing in that window used to have nowhere to go, so the joining peer stayed invisible in established peers' list_agents until some later patch happened to arrive. The transports now queue broadcasts for dialling peers and flush them when the connection registers. + * Every production bridge calls registerAgent() immediately after store.init() returns, while the fire-and-forget peer dials are still in flight. Broadcasts landing in that window used to have nowhere to go, so the joining peer stayed invisible in established peers' list_agents until some later patch happened to arrive. The transport now queues broadcasts for dialling peers and flushes them when the connection registers. */ import * as assert from "node:assert/strict"; +import { deviceIdToHex } from "@exadev/wire-mesh-core/domain/device-id"; import { MeshStore } from "../core/mesh-store.js"; -import { TlsTransport } from "../core/tls-transport.js"; +import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import { generateIdentity } from "../core/identity.js"; import type { PeerIdentity } from "../core/identity.js"; const TEST_PORT = 19890; const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); -/** A peer wired like a real bridge: TLS transport, fingerprint peer ID. */ +/** A peer wired like a real bridge: WireMeshTransport, device-id peer ID. */ function makePeer(identity: PeerIdentity): MeshStore { const store = new MeshStore(TEST_PORT); - store.peerId = identity.fingerprint; - store.setTransport(new TlsTransport(store.events, identity)); + store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); + store.setTransport(new WireMeshTransport(store.events, identity)); return store; } diff --git a/src/test/downtime-replay.integration.test.ts b/src/test/downtime-replay.integration.test.ts index 4675876..b66e184 100644 --- a/src/test/downtime-replay.integration.test.ts +++ b/src/test/downtime-replay.integration.test.ts @@ -8,8 +8,9 @@ import * as assert from "node:assert/strict"; import * as fs from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; +import { deviceIdToHex } from "@exadev/wire-mesh-core/domain/device-id"; import { MeshStore } from "../core/mesh-store.js"; -import { TlsTransport } from "../core/tls-transport.js"; +import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import { generateIdentity } from "../core/identity.js"; import { loadOrCreateIdentity, @@ -27,11 +28,11 @@ interface Peer { deliveries: DeliveryEvent[]; } -/** A peer wired like a real bridge: TLS transport, fingerprint peer ID. */ +/** A peer wired like a real bridge: WireMeshTransport, device-id peer ID. */ function makePeer(identity: PeerIdentity): Peer { const store = new MeshStore(TEST_PORT); - store.peerId = identity.fingerprint; - store.setTransport(new TlsTransport(store.events, identity)); + store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); + store.setTransport(new WireMeshTransport(store.events, identity)); const deliveries: DeliveryEvent[] = []; return { store, deliveries }; } @@ -109,7 +110,10 @@ async function main(): Promise { // B restarts in the same slot: same identity, same agent ID. const identityB2 = loadOrCreateIdentity(slot); - assert.equal(identityB2.fingerprint, identityB.fingerprint); + assert.equal( + deviceIdToHex(Uint8Array.from(identityB2.deviceId)), + deviceIdToHex(Uint8Array.from(identityB.deviceId)), + ); const b2 = makePeer(identityB2); b2.store.onDelivery = (_id, ev) => { b2.deliveries.push(ev); diff --git a/src/test/identity-restart.integration.test.ts b/src/test/identity-restart.integration.test.ts index e60209f..c51f186 100644 --- a/src/test/identity-restart.integration.test.ts +++ b/src/test/identity-restart.integration.test.ts @@ -1,15 +1,16 @@ /** * Integration test for issue #14: a bridge restarted with a persisted identity must keep its agent ID, room membership, and push delivery. * - * Before persistent identities, every restart generated a fresh TLS certificate, so the fingerprint-derived agent ID changed and peers kept targeting the old ID: their messages queued for an agent whose local handler (gated on agentId === peerId) never fired again. + * Before persistent identities, every restart generated a fresh keypair, so the device-id-derived agent ID changed and peers kept targeting the old ID: their messages queued for an agent whose local handler (gated on agentId === peerId) never fired again. */ import * as assert from "node:assert/strict"; import * as fs from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; +import { deviceIdToHex } from "@exadev/wire-mesh-core/domain/device-id"; import { MeshStore } from "../core/mesh-store.js"; -import { TlsTransport } from "../core/tls-transport.js"; +import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import { generateIdentity } from "../core/identity.js"; import type { PeerIdentity } from "../core/identity.js"; import { @@ -27,11 +28,11 @@ interface Peer { deliveries: DeliveryEvent[]; } -/** A peer wired like a real bridge: TLS transport, fingerprint peer ID. */ +/** A peer wired like a real bridge: WireMeshTransport, device-id peer ID. */ function makePeer(identity: PeerIdentity): Peer { const store = new MeshStore(TEST_PORT); - store.peerId = identity.fingerprint; - store.setTransport(new TlsTransport(store.events, identity)); + store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); + store.setTransport(new WireMeshTransport(store.events, identity)); const deliveries: DeliveryEvent[] = []; return { store, deliveries }; } @@ -98,7 +99,10 @@ async function main(): Promise { // A restarts in the same slot: same key material, same agent ID. const identityA2 = loadOrCreateIdentity(slot); - assert.equal(identityA2.fingerprint, identityA.fingerprint); + assert.equal( + deviceIdToHex(Uint8Array.from(identityA2.deviceId)), + deviceIdToHex(Uint8Array.from(identityA.deviceId)), + ); const a2 = makePeer(identityA2); a2.store.onDelivery = (_id, ev) => { a2.deliveries.push(ev); diff --git a/src/test/listener-policy.integration.test.ts b/src/test/listener-policy.integration.test.ts index b7534bd..1c43892 100644 --- a/src/test/listener-policy.integration.test.ts +++ b/src/test/listener-policy.integration.test.ts @@ -7,11 +7,19 @@ */ import * as net from "node:net"; -import * as tls from "node:tls"; +import { createTlsTransport } from "@exadev/wire-mesh-core/adapters/tls-transport"; +import { acceptMeshSession } from "@exadev/wire-mesh-core/domain/mesh-session"; import { MeshStore } from "../core/mesh-store.js"; import { CommsTool } from "../core/tool.js"; import { buildAction } from "../core/bridge.js"; import { generateIdentity } from "../core/identity.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; +import { + DOMAIN, + FRAME_SCOPE, + buildCommand, + WireMeshTransport, +} from "../core/wire-mesh-transport.js"; import type { ConnectionHandle, TransportEvents } from "../core/transport.js"; import * as assert from "node:assert/strict"; import { test, describe } from "node:test"; @@ -329,74 +337,63 @@ describe("listener policy", () => { }); void test("connections via non-default listener carry policy in handle", async () => { - const store = new MeshStore(TEST_PORT); - wireTestTransport(store); + // Constructed directly rather than through MeshStore: MeshStore.events is a getter that builds a fresh TransportEvents object on every access, so there's no way to observe what the transport itself passed to onIntroduction without either reaching into the transport by an unsafe cast or, as here, supplying our own TransportEvents object the transport calls directly. + let receivedPolicy: string | undefined = "not-called"; + const events: TransportEvents = { + onMessage: () => undefined, + onPeerConnected: () => undefined, + onPeerDisconnected: () => undefined, + onIntroduction: (handle) => { + receivedPolicy = handle.policy; + }, + onConnectionRequest: () => undefined, + onPeerList: () => undefined, + onPeerJoined: () => undefined, + onBecomeCoordinator: () => undefined, + }; + const identity = generateIdentity(); + const transport = new WireMeshTransport(events, identity); try { - await store.init(); + await transport.becomeCoordinator("127.0.0.1", 0); - // Add an observe listener - const listenerId = await store.addListener("127.0.0.1", 0, "observe"); - const listeners = store.listListeners(); + const listenerId = await transport.addListener("127.0.0.1", 0, "observe"); + const listeners = transport.listListeners(); const observeListener = listeners.find((l) => l.id === listenerId); assert.ok(observeListener, "Should find the observe listener"); - // Connect to the observe listener and send an introduce message over a real TLS client connection -- the listener is a TlsTransport server, which requires an actual TLS handshake before any application bytes are readable, and separately verifies the introduce message's claimed peerId against the client certificate's own fingerprint, so the probe needs a real generated identity, not an arbitrary string. The transport should tag the connection handle with policy="observe" - // - // We intercept at the transport.events level because store.events is a getter that creates a fresh object each call. + // Connect to the observe listener and send an introduce message over a real WireMeshTransport client session -- the transport should tag the resulting connection handle with policy="observe". const probeIdentity = generateIdentity(); - const receivedHandle = await new Promise<{ - policy: string | undefined; - } | null>((resolve) => { - const timeout = setTimeout(() => resolve(null), 3000); - - const transport = ( - store as unknown as { transport: { events: TransportEvents } } - ).transport; - // Captured by value (bound, so eslint doesn't flag an unsafely-detached method reference), not by a closure that re-reads transport.events.onIntroduction at call time -- a lazy re-lookup would resolve to the wrapper itself once the assignment below replaces it, recursing forever on the very first introduction. - const originalOnIntroduction = transport.events.onIntroduction.bind( - transport.events, - ); - transport.events.onIntroduction = (handle, msg) => { - // Capture the handle's policy - resolve({ policy: handle.policy }); - clearTimeout(timeout); - // Call the original handler - originalOnIntroduction(handle, msg); - }; - - const socket = tls.connect({ - port: observeListener.port, - host: "127.0.0.1", - key: probeIdentity.privateKey, - cert: probeIdentity.certificate, - rejectUnauthorized: false, - }); - - socket.on("secureConnect", () => { - socket.write( - JSON.stringify({ - method: "introduce", - peerId: probeIdentity.fingerprint, - dataPort: 19999, - }) + "\n", - ); - }); - - socket.on("error", () => { - socket.destroy(); - clearTimeout(timeout); - resolve(null); - }); + const probeTransport = createTlsTransport({ + certificatePem: probeIdentity.certificate, + privateKeyPem: probeIdentity.privateKey, }); - - assert.ok(receivedHandle, "Should receive an introduction"); - assert.equal( - receivedHandle.policy, - "observe", - "Handle should carry observe policy", + const connection = await probeTransport.connect( + `127.0.0.1:${String(observeListener.port)}`, ); + const probeIdentityPort = await toIdentityPort(probeIdentity); + const session = await acceptMeshSession(connection, probeIdentityPort, [ + DOMAIN, + ]); + try { + const outcome = await session.sendManageRequest( + buildCommand({ + method: "introduce", + peerId: "probe", + dataPort: 19999, + }), + FRAME_SCOPE, + ); + assert.equal(outcome.result, "ok", "introduce should be accepted"); + assert.equal( + receivedPolicy, + "observe", + "Handle should carry observe policy", + ); + } finally { + await session.close(); + } } finally { - await store.shutdown(); + await transport.shutdown(); } }); }); diff --git a/src/test/mesh-smoke.runner.ts b/src/test/mesh-smoke.runner.ts index 5c2c29d..c37a1e8 100644 --- a/src/test/mesh-smoke.runner.ts +++ b/src/test/mesh-smoke.runner.ts @@ -1,7 +1,7 @@ /** * Multi-process smoke test for the MeshStore mesh. * - * Spawns two separate Node.js processes running MeshStore instances over TlsTransport, verifies they discover each other via the coordinator, exchange messages, and receive push delivery. Each spawned process requires() compiled dist/ output directly (the point is exercising the real built artifact across a genuine process boundary, not re-testing TS source logic already covered elsewhere), so this runs via its own pnpm test:smoke script rather than the plain pnpm test glob -- unlike every other *.test.ts file, it needs a build to have happened first. + * Spawns two separate Node.js processes running MeshStore instances over WireMeshTransport, verifies they discover each other via the coordinator, exchange messages, and receive push delivery. Each spawned process requires() compiled dist/ output directly (the point is exercising the real built artifact across a genuine process boundary, not re-testing TS source logic already covered elsewhere), so this runs via its own pnpm test:smoke script rather than the plain pnpm test glob -- unlike every other *.test.ts file, it needs a build to have happened first. * * Usage: pnpm test:smoke */ @@ -109,14 +109,15 @@ function buildScript(name: string, actions: string): string { return [ `const { MeshStore } = require("./dist/core/mesh-store.js");`, `const { CommsTool } = require("./dist/core/tool.js");`, - `const { TlsTransport } = require("./dist/core/tls-transport.js");`, + `const { WireMeshTransport } = require("./dist/core/wire-mesh-transport.js");`, `const { generateIdentity } = require("./dist/core/identity.js");`, + `const { deviceIdToHex } = require("@exadev/wire-mesh-core/domain/device-id");`, `function log(msg) { process.stdout.write(JSON.stringify(msg) + "\\n"); }`, `(async () => {`, ` const store = new MeshStore(${String(SMOKE_PORT)});`, ` const identity = generateIdentity();`, - ` store.peerId = identity.fingerprint;`, - ` store.setTransport(new TlsTransport(store.events, identity));`, + ` store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId));`, + ` store.setTransport(new WireMeshTransport(store.events, identity));`, ` const tool = new CommsTool(store);`, ` const deliveries = [];`, ` store.onDelivery = (_id, event) => {`, diff --git a/src/test/peer-id-verification.integration.test.ts b/src/test/peer-id-verification.integration.test.ts deleted file mode 100644 index 0b43431..0000000 --- a/src/test/peer-id-verification.integration.test.ts +++ /dev/null @@ -1,246 +0,0 @@ -/** - * Peer ID verification integration test (#40) — a socket claiming a peer ID that doesn't match the certificate it actually presents must be rejected, on every path where TlsTransport learns a remote peer's identity from a self-reported wire message: the coordinator's `introduce` handler, the data server's `pong` handler, and the client's own `connectToPeer` dial. - * - * Run: node dist/test/peer-id-verification.integration.test.js [test-name] With no argument, every scenario runs in order. - */ - -import * as net from "node:net"; -import * as tls from "node:tls"; -import * as assert from "node:assert/strict"; -import { TlsTransport } from "../core/tls-transport.js"; -import { generateIdentity } from "../core/identity.js"; -import { encode } from "../core/wire-protocol.js"; -import type { PeerInfo } from "../core/wire-protocol.js"; -import type { TransportEvents } from "../core/transport.js"; - -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -function allocFreePort(): Promise { - return new Promise((resolve, reject) => { - const server = net.createServer(); - server.listen(0, "127.0.0.1", () => { - const addr = server.address(); - const port = addr && typeof addr === "object" ? addr.port : 0; - server.close(() => resolve(port)); - }); - server.on("error", reject); - }); -} - -function noopEvents(overrides: Partial = {}): TransportEvents { - return { - onMessage: () => undefined, - onPeerConnected: () => undefined, - onPeerDisconnected: () => undefined, - onIntroduction: () => undefined, - onConnectionRequest: () => undefined, - onPeerList: () => undefined, - onPeerJoined: () => undefined, - onBecomeCoordinator: () => undefined, - ...overrides, - }; -} - -async function testSpoofedIntroduceRejected(): Promise { - const identityCoordinator = generateIdentity(); - const identityAttacker = generateIdentity(); - - let introduced = false; - let sawError = false; - const transport = new TlsTransport( - noopEvents({ - onIntroduction: () => { - introduced = true; - }, - onError: () => { - sawError = true; - }, - }), - identityCoordinator, - ); - const port = await allocFreePort(); - await transport.becomeCoordinator("127.0.0.1", port); - - // The attacker connects with its own genuine certificate, but sends an `introduce` claiming the coordinator's own peer ID — self-reported identity that doesn't match the certificate on this connection. - const socket = tls.connect({ - key: identityAttacker.privateKey, - cert: identityAttacker.certificate, - host: "127.0.0.1", - port, - rejectUnauthorized: false, - }); - - await new Promise((resolve, reject) => { - socket.once("secureConnect", () => { - socket.write( - encode({ - method: "introduce", - peerId: identityCoordinator.fingerprint, - dataPort: 12345, - }), - ); - resolve(); - }); - socket.once("error", reject); - }); - - await sleep(300); - - assert.strictEqual( - introduced, - false, - "onIntroduction must not fire for a spoofed peer ID", - ); - assert.ok(sawError, "the rejection should be reported via onError"); - assert.ok(socket.destroyed, "the spoofing socket should be destroyed"); - - await transport.shutdown(); - console.log(" ✓ introduce with mismatched certificate is rejected"); -} - -async function testSpoofedPongRejected(): Promise { - const identityListener = generateIdentity(); - const identityAttacker = generateIdentity(); - - let connected = false; - let sawError = false; - const transport = new TlsTransport( - noopEvents({ - onPeerConnected: () => { - connected = true; - }, - onError: () => { - sawError = true; - }, - }), - identityListener, - ); - await transport.startDataServer(); - - // The attacker connects to the data server with its own certificate, but sends a `pong` claiming an arbitrary, unrelated peer ID. - const socket = tls.connect({ - key: identityAttacker.privateKey, - cert: identityAttacker.certificate, - host: "127.0.0.1", - port: transport.dataPort, - rejectUnauthorized: false, - }); - - await new Promise((resolve, reject) => { - socket.once("secureConnect", () => { - socket.write(encode({ method: "pong", peerId: "NOT-MY-CERTIFICATE" })); - resolve(); - }); - socket.once("error", reject); - }); - - await sleep(300); - - assert.strictEqual( - connected, - false, - "onPeerConnected must not fire for a spoofed peer ID", - ); - assert.ok(sawError, "the rejection should be reported via onError"); - assert.ok(socket.destroyed, "the spoofing socket should be destroyed"); - - await transport.shutdown(); - console.log(" ✓ pong with mismatched certificate is rejected"); -} - -async function testConnectToPeerCertMismatchRejected(): Promise { - // The real peer B is listening under its own genuine identity... - const identityB = generateIdentity(); - const transportB = new TlsTransport(noopEvents(), identityB); - await transportB.startDataServer(); - - // ...but the peer list entry a compromised or misbehaving coordinator could hand to a dialling client claims a completely different ID for that same host:port. - const claimedPeer: PeerInfo = { - id: "CLAIMED-BUT-WRONG-ID", - port: transportB.dataPort, - startedAt: new Date().toISOString(), - }; - - let sawError = false; - const identityDialer = generateIdentity(); - const transportDialer = new TlsTransport( - noopEvents({ - onError: () => { - sawError = true; - }, - }), - identityDialer, - ); - - await transportDialer.connectToPeer(claimedPeer, identityDialer.fingerprint); - await sleep(200); - - assert.ok( - sawError, - "the rejection should be reported via onError when the dialled peer's certificate doesn't match the claimed ID", - ); - await assert.rejects( - () => - transportDialer.send( - { id: claimedPeer.id }, - { method: "pong", peerId: identityDialer.fingerprint }, - ), - /No connection for handle/, - "no peer connection should have been registered under the falsely claimed ID", - ); - - await transportDialer.shutdown(); - await transportB.shutdown(); - console.log( - " ✓ connectToPeer rejects a certificate that doesn't match the claimed peer ID", - ); -} - -// --------------------------------------------------------------------------- -// Runner -// --------------------------------------------------------------------------- - -const testName = process.argv[2]; - -const tests: Record Promise> = { - "spoofed-introduce-rejected": testSpoofedIntroduceRejected, - "spoofed-pong-rejected": testSpoofedPongRejected, - "connect-to-peer-cert-mismatch-rejected": - testConnectToPeerCertMismatchRejected, -}; - -const selected = - testName === undefined - ? Object.entries(tests) - : Object.entries(tests).filter(([name]) => name === testName); -if (selected.length === 0) { - console.error(`Unknown test: ${testName}`); - console.error(`Available: ${Object.keys(tests).join(", ")}`); - process.exit(1); -} - -async function run(): Promise { - for (const [name, fn] of selected) { - console.log(`Running ${name}:`); - await fn(); - } - - const maxWait = 2000; - const start = Date.now(); - while ( - (( - process as unknown as { _getActiveHandles?: () => unknown[] } - )._getActiveHandles?.()?.length ?? 0) > 0 && - Date.now() - start < maxWait - ) { - await new Promise((resolve) => setTimeout(resolve, 50)); - } - process.exit(0); -} - -run().catch((err: unknown) => { - console.error(`FAIL [${testName ?? "all"}]:`, err); - process.exit(1); -}); diff --git a/src/test/test-transport.ts b/src/test/test-transport.ts index 8b5f01a..d3747e2 100644 --- a/src/test/test-transport.ts +++ b/src/test/test-transport.ts @@ -1,6 +1,5 @@ -/** Wires a real TlsTransport with a freshly generated identity onto a freshly constructed MeshStore -- the same setTransport() call every production bridge makes immediately after construction. MeshStore has no default transport, so every test that constructs one needs this (or an equivalent explicit setTransport() call) before init() or any other transport-using method runs. */ +/** Wires a real WireMeshTransport with a freshly generated identity onto a freshly constructed MeshStore -- the same setTransport() call every production bridge makes immediately after construction (via createBridgeMesh). MeshStore has no default transport, so every test that constructs one needs this (or an equivalent explicit setTransport() call) before init() or any other transport-using method runs. */ -import { TlsTransport } from "../core/tls-transport.js"; import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import { generateIdentity } from "../core/identity.js"; import { deviceIdToHex } from "@exadev/wire-mesh-core/domain/device-id"; @@ -8,22 +7,10 @@ import type { MeshStore } from "../core/mesh-store.js"; export function wireTestTransport(store: MeshStore): void { const identity = generateIdentity(); - // Every real bridge sets peerId to the identity's own certificate fingerprint before wiring the transport -- TlsTransport's cert-pinning trust model means a peer's advertised ID and the fingerprint the other side actually authenticates the connection against must be the same value, or introduction/state-sync never recognises the peer as itself. - store.peerId = identity.fingerprint; - store.setTransport(new TlsTransport(store.events, identity)); - // Surface transport-level errors instead of leaving them silent — a genuine socket failure during a test run is signal worth seeing even when the test's own assertions still pass, since it can point at a real race the assertions don't happen to catch. - store.onError = (e) => { - console.error(`[transport error, peerId=${store.peerId}]`, e.message); - }; -} - -/** - * Wires a real WireMeshTransport onto a freshly constructed MeshStore -- the P2 substrate swap's own analogue of wireTestTransport above. Sets peerId to deviceIdToHex(identity.deviceId) rather than the certificate fingerprint: production bridges don't cut MeshStore.peerId over to this until step 4 of the migration's own sequencing, but WireMeshTransport's own internal session bookkeeping is already keyed by device-id, so a test exercising it end-to-end through a real MeshStore needs the two identity spaces to actually correlate. - */ -export function wireWireMeshTestTransport(store: MeshStore): void { - const identity = generateIdentity(); + // Every real bridge sets peerId to deviceIdToHex(identity.deviceId) before wiring the transport (createBridgeMesh) -- WireMeshTransport's own session bookkeeping is keyed by device-id, so a peer's advertised ID and the identity the other side actually authenticates the connection against must be the same value, or introduction/state-sync never recognises the peer as itself. store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); store.setTransport(new WireMeshTransport(store.events, identity)); + // Surface transport-level errors instead of leaving them silent — a genuine socket failure during a test run is signal worth seeing even when the test's own assertions still pass, since it can point at a real race the assertions don't happen to catch. store.onError = (e) => { console.error(`[transport error, peerId=${store.peerId}]`, e.message); }; diff --git a/src/test/tls-transport.integration.test.ts b/src/test/tls-transport.integration.test.ts deleted file mode 100644 index bd83d56..0000000 --- a/src/test/tls-transport.integration.test.ts +++ /dev/null @@ -1,198 +0,0 @@ -/** - * TlsTransport integration test — verifies that two MeshStore instances - * can communicate over TLS with certificate pinning. - * - * Run: node dist/test/tls-transport.integration.test.js [test-name] - * With no argument, every scenario runs in order. - */ - -import * as net from "node:net"; -import * as assert from "node:assert/strict"; -import { MeshStore } from "../core/mesh-store.js"; -import { TlsTransport } from "../core/tls-transport.js"; -import { generateIdentity } from "../core/identity.js"; -import type { PeerIdentity } from "../core/identity.js"; - -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -function allocFreePort(): Promise { - return new Promise((resolve, reject) => { - const server = net.createServer(); - server.listen(0, "127.0.0.1", () => { - const addr = server.address(); - const port = addr && typeof addr === "object" ? addr.port : 0; - server.close(() => resolve(port)); - }); - server.on("error", reject); - }); -} - -async function cleanup(...stores: MeshStore[]): Promise { - for (const s of stores) { - await s.shutdown(); - } -} - -async function testTlsPeerCommunication(): Promise { - const port = await allocFreePort(); - const identityA: PeerIdentity = generateIdentity(); - const identityB: PeerIdentity = generateIdentity(); - - // Verify identities are different - assert.notStrictEqual(identityA.fingerprint, identityB.fingerprint); - console.log(" Identity A:", identityA.fingerprint.substring(0, 23) + "..."); - console.log(" Identity B:", identityB.fingerprint.substring(0, 23) + "..."); - - // Create stores with TLS transport - const storeA = new MeshStore(port); - storeA.peerId = identityA.fingerprint; - storeA.setTransport(new TlsTransport(storeA.events, identityA)); - - const deliveriesA: unknown[] = []; - storeA.onDelivery = () => { - deliveriesA.push(1); - }; - await storeA.init(); - await storeA.registerAgent({ - name: "a", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - await sleep(100); - - const storeB = new MeshStore(port); - storeB.peerId = identityB.fingerprint; - storeB.setTransport(new TlsTransport(storeB.events, identityB)); - - const deliveriesB: Record[] = []; - storeB.onDelivery = (_id: string, ev: unknown) => { - deliveriesB.push(ev as Record); - }; - await storeB.init(); - await storeB.registerAgent({ - name: "b", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - await sleep(300); - - // Create room and exchange messages - const room = await storeA.createRoom({ - name: `tls-test-${port}`, - type: "public", - owner: storeA.peerId, - description: "TLS transport test", - }); - await sleep(200); - await storeB.joinRoom(room.id, storeB.peerId); - await sleep(200); - - deliveriesB.length = 0; - await storeA.sendRoomMessage(room.id, storeA.peerId, "Hello over TLS!"); - await sleep(300); - - assert.ok( - deliveriesB.length >= 1, - `B should receive at least 1 delivery, got ${String(deliveriesB.length)}`, - ); - const roomMsg = deliveriesB.find((ev) => ev.type === "room_message"); - assert.ok(roomMsg !== undefined, "Should find room_message event"); - const message = roomMsg.message as Record; - assert.strictEqual(message.content, "Hello over TLS!"); - assert.strictEqual(message.from, identityA.fingerprint); - - console.log(" ✓ Room message received over TLS"); - - // Test DM - deliveriesB.length = 0; - await storeA.sendDm(storeA.peerId, storeB.peerId, "Direct over TLS!"); - await sleep(300); - - const dm = deliveriesB.find((ev) => ev.type === "dm"); - assert.ok(dm !== undefined, "Should find dm event"); - const dmMessage = dm.message as Record; - assert.strictEqual(dmMessage.content, "Direct over TLS!"); - assert.strictEqual(dmMessage.from, identityA.fingerprint); - - console.log(" ✓ DM received over TLS"); - - await cleanup(storeA, storeB); -} - -async function testFingerprintIsPeerId(): Promise { - const identity = generateIdentity(); - const port = await allocFreePort(); - - const store = new MeshStore(port); - store.peerId = identity.fingerprint; - store.setTransport(new TlsTransport(store.events, identity)); - - await store.init(); - const agent = await store.registerAgent({ - name: "fp-test", - harness: "test", - cwd: "/test/fp", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - // The agent ID should be the certificate fingerprint - assert.strictEqual(agent.id, identity.fingerprint); - console.log(" ✓ Peer ID equals certificate fingerprint"); - - await store.shutdown(); -} - -// --------------------------------------------------------------------------- -// Runner -// --------------------------------------------------------------------------- - -const testName = process.argv[2]; - -const tests: Record Promise> = { - "tls-communication": testTlsPeerCommunication, - "tls-fingerprint": testFingerprintIsPeerId, -}; - -const selected = - testName === undefined - ? Object.entries(tests) - : Object.entries(tests).filter(([name]) => name === testName); -if (selected.length === 0) { - console.error(`Unknown test: ${testName}`); - console.error(`Available: ${Object.keys(tests).join(", ")}`); - process.exit(1); -} - -async function run(): Promise { - for (const [name, fn] of selected) { - console.log(`Running ${name}:`); - await fn(); - } - - const maxWait = 2000; - const start = Date.now(); - while ( - (( - process as unknown as { _getActiveHandles?: () => unknown[] } - )._getActiveHandles?.()?.length ?? 0) > 0 && - Date.now() - start < maxWait - ) { - await new Promise((resolve) => setTimeout(resolve, 50)); - } - process.exit(0); -} - -run().catch((err: unknown) => { - console.error(`FAIL [${testName ?? "all"}]:`, err); - process.exit(1); -}); diff --git a/src/test/wire-mesh-transport-approval.integration.test.ts b/src/test/wire-mesh-transport-approval.integration.test.ts deleted file mode 100644 index 3f89521..0000000 --- a/src/test/wire-mesh-transport-approval.integration.test.ts +++ /dev/null @@ -1,290 +0,0 @@ -/** - * Parity test for WireMeshTransport's connection-approval flow (connectToRemote/acceptConnection/rejectConnection) and listener management, mirroring approval.integration.test.ts's own scenarios against the new substrate. - */ - -import * as assert from "node:assert/strict"; -import { test, describe } from "node:test"; -import * as net from "node:net"; -import { createTlsTransport } from "@exadev/wire-mesh-core/adapters/tls-transport"; -import { acceptMeshSession } from "@exadev/wire-mesh-core/domain/mesh-session"; -import { MeshStore } from "../core/mesh-store.js"; -import { generateIdentity } from "../core/identity.js"; -import { toIdentityPort } from "../core/wire-mesh-identity.js"; -import { - DOMAIN, - FRAME_SCOPE, - buildCommand, -} from "../core/wire-mesh-transport.js"; -import type { DeliveryEvent, AgentIdentity } from "../core/types.js"; -import { waitFor, wireWireMeshTestTransport } from "./test-transport.js"; - -function findFreePort(): Promise { - return new Promise((resolve, reject) => { - const server = net.createServer(); - server.listen(0, "127.0.0.1", () => { - const addr = server.address(); - const port = typeof addr === "object" && addr !== null ? addr.port : 0; - server.close(() => resolve(port)); - }); - server.on("error", reject); - }); -} - -/** Find a free port for a single test's use. An alias for findFreePort(): asking the OS for a fresh ephemeral port each call already guarantees distinctness from any other currently-bound port, so no arithmetic offset is layered on top -- a prior +offset scheme could push an already-high OS-assigned port past 65535 and fail with ERR_SOCKET_BAD_PORT. */ -function uniquePort(): Promise { - return findFreePort(); -} - -describe("WireMeshTransport connection approval", () => { - void test("accept establishes the peer connection", async () => { - const portA = await uniquePort(); - const portB = await uniquePort(); - - const storeA = new MeshStore(portA); - wireWireMeshTestTransport(storeA); - const storeB = new MeshStore(portB); - wireWireMeshTestTransport(storeB); - try { - const receivedRequests: Extract< - DeliveryEvent, - { type: "connection_request" } - >[] = []; - storeA.onDelivery = (_id, event) => { - if (event.type === "connection_request") { - receivedRequests.push(event); - } - }; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - storeB.connectToRemote("127.0.0.1", portA); - - await waitFor( - () => receivedRequests.length === 1, - "coordinator receives the connection request", - ); - const request = receivedRequests[0]; - assert.ok(request?.connectionId); - - await storeA.acceptConnection(request.connectionId); - - await waitFor( - () => storeA.serialise().agents[storeB.peerId] !== undefined, - "coordinator sees the connector agent", - ); - await waitFor( - () => storeB.serialise().agents[storeA.peerId] !== undefined, - "connector sees the coordinator agent", - ); - - const agentsA = await storeA.listAgents(storeA.peerId); - const agentsB = await storeB.listAgents(storeB.peerId); - assert.ok( - agentsA.some((a) => a.id === storeB.peerId), - "Coordinator should see the connector agent", - ); - assert.ok( - agentsB.some((a) => a.id === storeA.peerId), - "Connector should see the coordinator agent", - ); - } finally { - await storeB.shutdown(); - await storeA.shutdown(); - } - }); - - void test("reject closes with reason", async () => { - const portA = await uniquePort(); - const portB = await uniquePort(); - - const storeA = new MeshStore(portA); - wireWireMeshTestTransport(storeA); - const storeB = new MeshStore(portB); - wireWireMeshTestTransport(storeB); - try { - const receivedRequests: Extract< - DeliveryEvent, - { type: "connection_request" } - >[] = []; - storeA.onDelivery = (_id, event) => { - if (event.type === "connection_request") { - receivedRequests.push(event); - } - }; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - storeB.connectToRemote("127.0.0.1", portA); - - await waitFor( - () => receivedRequests.length === 1, - "coordinator receives the connection request", - ); - const request = receivedRequests[0]; - assert.ok(request?.connectionId); - - await storeA.rejectConnection(request.connectionId, "unauthorised"); - - await waitFor( - () => storeA.listPendingConnections().length === 0, - "the pending connection clears", - ); - const agentsA = await storeA.listAgents(storeA.peerId); - assert.ok( - !agentsA.some((a) => a.id === storeB.peerId), - "Rejected peer should not appear in agent list", - ); - } finally { - await storeB.shutdown(); - await storeA.shutdown(); - } - }); - - void test("a message other than introduce/connect_request from an unapproved connection is refused, not routed", async () => { - const portA = await uniquePort(); - const storeA = new MeshStore(portA); - wireWireMeshTestTransport(storeA); - try { - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - // A hostile client that completes a TLS handshake (proving only which key it holds, not that a human approved it) and skips connect_request entirely, going straight for a forged state_update. - const attackerIdentity = generateIdentity(); - const attackerTransport = createTlsTransport({ - certificatePem: attackerIdentity.certificate, - privateKeyPem: attackerIdentity.privateKey, - }); - const connection = await attackerTransport.connect( - `127.0.0.1:${String(portA)}`, - ); - const attackerIdentityPort = await toIdentityPort(attackerIdentity); - const session = await acceptMeshSession( - connection, - attackerIdentityPort, - [DOMAIN], - ); - - const forgedAgent: AgentIdentity = { - id: "forged-attacker-agent", - version: 1, - name: "forged", - harness: "test", - cwd: "/forged", - pid: 1, - startedAt: new Date().toISOString(), - visibility: "visible", - status: "active", - tags: [], - subscribedRooms: [], - }; - const outcome = await session.sendManageRequest( - buildCommand({ - method: "state_update", - patch: { type: "agent_upsert", agent: forgedAgent }, - }), - FRAME_SCOPE, - ); - - assert.equal( - outcome.result, - "error", - "an unapproved connection's message must be refused, not routed", - ); - assert.equal( - storeA.serialise().agents[forgedAgent.id], - undefined, - "a forged state_update from an unapproved connection must never reach mesh-store's own state", - ); - } finally { - await storeA.shutdown(); - } - }); -}); - -describe("WireMeshTransport listener management", () => { - void test("addListener creates an additional listener; removeListener removes it", async () => { - const port = await uniquePort(); - const store = new MeshStore(port); - wireWireMeshTestTransport(store); - try { - await store.init(); - const before = store.listListeners(); - assert.strictEqual( - before.length, - 1, - "starts with just the default listener", - ); - assert.ok(before[0]?.isDefault); - - const extraPort = await uniquePort(); - const id = await store.addListener("127.0.0.1", extraPort, "observe"); - const afterAdd = store.listListeners(); - assert.strictEqual(afterAdd.length, 2); - const added = afterAdd.find((l) => l.id === id); - assert.ok(added); - assert.strictEqual(added.policy, "observe"); - assert.strictEqual(added.isDefault, false); - - await store.removeListener(id); - const afterRemove = store.listListeners(); - assert.strictEqual(afterRemove.length, 1); - } finally { - await store.shutdown(); - } - }); - - void test("removeListener rejects removing the default listener", async () => { - const port = await uniquePort(); - const store = new MeshStore(port); - wireWireMeshTestTransport(store); - try { - await store.init(); - const [defaultListener] = store.listListeners(); - assert.ok(defaultListener); - await assert.rejects(() => store.removeListener(defaultListener.id)); - } finally { - await store.shutdown(); - } - }); -}); diff --git a/src/test/wire-mesh-transport.integration.test.ts b/src/test/wire-mesh-transport.integration.test.ts deleted file mode 100644 index 43d3bb8..0000000 --- a/src/test/wire-mesh-transport.integration.test.ts +++ /dev/null @@ -1,130 +0,0 @@ -/** - * End-to-end parity test for WireMeshTransport, mirroring mesh-e2e.integration.test.ts's own scenario (coordinator discovery, room creation, messaging, delivery push) but against the new substrate instead of TlsTransport -- the concrete proof this phase's own charter ("semantics unchanged") holds for the core mesh-formation path, using waitFor polling rather than fixed sleeps for the same reason approval.integration.test.ts already does. - */ - -import * as assert from "node:assert/strict"; -import { test } from "node:test"; -import { MeshStore } from "../core/mesh-store.js"; -import { CommsTool } from "../core/tool.js"; -import { buildAction } from "../core/bridge.js"; -import type { DeliveryEvent } from "../core/types.js"; -import { waitFor, wireWireMeshTestTransport } from "./test-transport.js"; - -async function createStore( - port: number, - name: string, - harness: string, -): Promise<{ store: MeshStore; tool: CommsTool; deliveries: DeliveryEvent[] }> { - const store = new MeshStore(port); - wireWireMeshTestTransport(store); - - const deliveries: DeliveryEvent[] = []; - store.onDelivery = (_agentId, event) => { - deliveries.push(event); - }; - - const tool = new CommsTool(store); - - await store.init(); - await store.registerAgent({ - name, - harness, - cwd: `/test/${name}`, - pid: process.pid, - visibility: "visible", - tags: [], - }); - - return { store, tool, deliveries }; -} - -void test("WireMeshTransport: coordinator discovery, rooms, messaging, and delivery match TlsTransport's own semantics", async () => { - const port = 20_000 + Math.floor(Math.random() * 1000); - const a = await createStore(port, "peer-a", "test-a"); - let b: Awaited> | undefined; - - try { - b = await createStore(port, "peer-b", "test-b"); - - await waitFor( - () => a.store.serialise().agents[b?.store.peerId ?? ""] !== undefined, - "A sees B's agent", - ); - await waitFor( - () => b?.store.serialise().agents[a.store.peerId] !== undefined, - "B sees A's agent", - ); - - const agentsA = await a.store.listAgents(a.store.peerId); - const agentsB = await b.store.listAgents(b.store.peerId); - assert.ok(agentsA.length >= 2, "A should see both agents"); - assert.ok(agentsB.length >= 2, "B should see both agents"); - - const roomId = `test-room-${String(Date.now())}`; - const room = await a.store.createRoom({ - name: roomId, - type: "public", - owner: a.store.peerId, - description: "Test room", - }); - - await waitFor( - () => b?.store.serialise().rooms[room.id] !== undefined, - "B sees the room", - ); - await b.store.joinRoom(room.id, b.store.peerId); - - await waitFor( - () => a.store.serialise().rooms[room.id]?.members.length === 2, - "A sees B join the room", - ); - const roomA = await a.store.getRoom(room.id); - const roomB = await b.store.getRoom(room.id); - assert.strictEqual(roomA?.members.length, 2, "A should see 2 room members"); - assert.strictEqual(roomB?.members.length, 2, "B should see 2 room members"); - - b.deliveries.length = 0; - await a.store.sendRoomMessage(room.id, a.store.peerId, "Hello from A!"); - await waitFor( - () => b?.deliveries.length !== 0, - "B receives the room message", - ); - const roomMsg = b.deliveries[0]; - assert.ok(roomMsg); - assert.strictEqual(roomMsg.type, "room_message"); - assert.strictEqual(roomMsg.message.content, "Hello from A!"); - - b.deliveries.length = 0; - await a.store.sendDm(a.store.peerId, b.store.peerId, "Hey B!"); - await waitFor(() => b?.deliveries.length !== 0, "B receives the DM"); - const dmEvent = b.deliveries[0]; - assert.ok(dmEvent); - assert.strictEqual(dmEvent.type, "dm"); - - const messages = await b.store.readRoomMessages(room.id); - assert.ok(messages.length >= 1, "B should see the message"); - - a.deliveries.length = 0; - const action = buildAction({ - action: "send", - target: room.id, - content: "Hello from B via tool!", - }); - await b.tool.handle( - { - agentId: b.store.peerId, - harness: "test-b", - cwd: "/test/peer-b", - pid: process.pid, - }, - action, - ); - await waitFor( - () => a.deliveries.length !== 0, - "A receives B's tool-sent message", - ); - } finally { - await b?.store.shutdown(); - await a.store.shutdown(); - } -});