From e6464f4245ee385a6d644c85d6143e71affc60a5 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Fri, 11 Sep 2026 20:08:07 +0100 Subject: [PATCH 1/2] fix(core): quarantine WireMeshTransport sessions until connection approval Every listener (the coordinator port, and any addListener-created listener, including a non-loopback one exposed via mesh_listen) promoted an accepted connection to a fully trusted, routed session the instant its TLS handshake completed. TLS only proves which key the far side holds, never that a human approved it as a mesh member -- the only gate that ever existed was intercepting connect_request specifically, but any other message type (state_update, a room send, a DM) reached mesh-store's own onMessage unfiltered, letting an unapproved peer forge arbitrary mesh state before ever requesting a connection. The previous TlsTransport substrate never had this gap: its own coordinator socket only ever parsed introduce/connect_request, dropping anything else, while a separate data connection (reachable only via a port learned through an already-approved introduction) carried full routing. Restore that boundary under the new substrate's unified accept path: a listener-accepted session is quarantined -- restricted to introduce (the pre-existing, ungated coordinator-handoff path, unchanged) or connect_request (now blocking in place on the human decision Promise acceptConnection/ rejectConnection resolve) -- and only promoted to full trust once approved. A session's own data-port dial (the second half of an already-approved mesh pairing) is unaffected, matching its pre-existing trust model exactly. Adds a test that forges a state_update from a connection which never sent connect_request at all, confirming it's refused rather than reaching mesh-store's state. --- src/core/wire-mesh-transport.ts | 139 ++++++++++++++---- ...esh-transport-approval.integration.test.ts | 90 +++++++++++- 2 files changed, 189 insertions(+), 40 deletions(-) diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 9346088..425a0b4 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -43,20 +43,22 @@ import { nanoid } from "./nanoid.js"; const COORDINATOR_HOST = "127.0.0.1"; -/** This project's own namespaced domain (registrant "exadev.io", local name "agent-comms-v1"), matching namespaced-domain-id's "/" shape -- registry/core-domains.md's own recommended pattern for a third party. */ -const DOMAIN = "exadev.io/agent-comms-v1"; +/** This project's own namespaced domain (registrant "exadev.io", local name "agent-comms-v1"), matching namespaced-domain-id's "/" shape -- registry/core-domains.md's own recommended pattern for a third party. Exported for test use only: a security test simulating a hostile client that skips connect_request needs to construct a well-formed frame under the same domain/verb/scope this transport itself listens on, rather than duplicating these as separately-maintained magic strings that could silently drift from the real values. */ +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. */ -const FRAME_VERB = "exadev.io/agent-comms-v1:frame"; +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. */ -const FRAME_SCOPE: Readonly = { kind: "agent-comms-mesh" }; +export const FRAME_SCOPE: Readonly = { + kind: "agent-comms-mesh", +}; // --------------------------------------------------------------------------- // Message carriage // --------------------------------------------------------------------------- -function buildCommand(message: MeshMessage): ManageCommand { +export function buildCommand(message: MeshMessage): ManageCommand { return { verb: FRAME_VERB, params: { message } }; } @@ -74,12 +76,19 @@ function extractMessage(command: ManageCommand): MeshMessage | undefined { // Internal session bookkeeping // --------------------------------------------------------------------------- +/** A human decision on a connect_request: "accept" resumes the still-blocked consumeQuarantined loop as a fully trusted session; "reject" (also used when the requester disconnects before a decision is made) unblocks it to close instead. */ +type ConnectionDecision = "accept" | "reject"; + interface PendingConnection { respond: IncomingManageRequest["respond"]; dataPort: number; name: string; fingerprint: string; policy: ListenerPolicy | undefined; + /** Held so acceptConnection can trackSession it synchronously, before firing onIntroduction -- that event's own handler (mesh-store's handleIntroduction) sends a reply on this same handle immediately, which needs peerSessions already populated. Waiting for consumeQuarantined's own suspended loop to resume and do it would race: resolve() below only wakes that loop on a later microtask tick, after onIntroduction has already fired. */ + session: AcceptedMeshSession; + /** Settles the Promise consumeQuarantined is blocked on for this connect_request -- the mechanism by which acceptConnection/rejectConnection resume a loop suspended mid-iteration, without ever needing to re-obtain (and so needing to reason about the identity of) a second iterator over the same session's incomingManageRequests. */ + resolve: (decision: ConnectionDecision) => void; } interface TrackedListener { @@ -169,6 +178,7 @@ export class WireMeshTransport implements MeshTransport { connection: Readonly, policy: ListenerPolicy | undefined, fireOnPeerConnected: boolean, + requiresApproval: boolean, ): Promise { if (this.shutDown) { await connection.close(); @@ -187,13 +197,21 @@ export class WireMeshTransport implements MeshTransport { await session.close(); return; } - this.trackSession(deviceIdHex, session); // ConnectionHandle.policy is `?: ListenerPolicy`, not `?: ListenerPolicy | undefined` -- under exactOptionalPropertyTypes these are genuinely different types, so the key must be entirely absent rather than present-with-undefined-value when this listener has no policy of its own. const handle: ConnectionHandle = { id: deviceIdHex, ...(policy !== undefined ? { policy } : {}), }; + if (requiresApproval) { + // A listener a stranger can dial cold (the coordinator port, or any addListener-created listener) must not hand out full routing on the strength of a TLS handshake alone -- that only proves which key the far side holds, never that a human has approved them as a mesh member. Quarantine every request from this session until it's either an `introduce` (the pre-existing, ungated coordinator-handoff path, unchanged from before this substrate swap) or an approved `connect_request`. + this.allSessions.add(session); + this.consumeQuarantined(session, handle); + this.watchForDisconnect(session, handle, deviceIdHex); + return; + } + + this.trackSession(deviceIdHex, session); if (fireOnPeerConnected) { const info: PeerInfo = { id: deviceIdHex, @@ -207,6 +225,69 @@ export class WireMeshTransport implements MeshTransport { this.watchForDisconnect(session, handle, deviceIdHex); } + /** Reads a not-yet-trusted session's requests until it's promoted (an `introduce`, handled and trusted immediately, matching the pre-existing coordinator-handoff trust boundary) or a `connect_request` arrives, at which point this same loop iteration blocks on the human decision Promise stored in pendingConnections -- resumed in place by acceptConnection/rejectConnection (or by watchForDisconnect, if the requester disconnects first) -- rather than ever stopping and later re-entering the session's incomingManageRequests from a second call, which would require assuming a fresh access yields a distinct, independently-advancing iterator rather than resuming the one already in progress. Anything else arriving before introduce/connect_request is a protocol violation from an unapproved peer and is refused and closed rather than routed. */ + private consumeQuarantined( + session: AcceptedMeshSession, + handle: ConnectionHandle, + ): void { + void (async () => { + let approved = false; + for await (const request of session.incomingManageRequests) { + if (this.shutDown) break; + if (approved) { + await this.dispatchIncoming(request, handle); + continue; + } + const message = extractMessage(request.command); + if (message === undefined) { + await request.respond({ result: "ok" }).catch(() => undefined); + continue; + } + if (message.method === "introduce") { + this.trackSession(handle.id, session); + this.route(handle, message); + await request.respond({ result: "ok" }).catch(() => undefined); + approved = true; + continue; + } + if (message.method === "connect_request") { + // Held open deliberately -- see this file's own header comment. Answered later by acceptConnection/rejectConnection, not here. + const decision = await new Promise((resolve) => { + this.pendingConnections.set(handle.id, { + respond: request.respond, + dataPort: message.dataPort, + name: message.name, + fingerprint: message.fingerprint, + policy: handle.policy, + session, + resolve, + }); + this.events.onConnectionRequest(handle, { + peerId: handle.id, + dataPort: message.dataPort, + name: message.name, + fingerprint: message.fingerprint, + }); + }); + if (decision === "reject") { + // Rejection can arrive via rejectConnection (session still open) or via shutdown/disconnect (session already closing) -- catch rather than assume which. + await session.close().catch(() => undefined); + return; + } + // acceptConnection has already called trackSession synchronously, before firing onIntroduction -- nothing left to do here beyond trusting subsequent requests on this same session. + approved = true; + continue; + } + // Any other message before introduce/connect_request is a protocol violation from an unapproved peer -- refuse and terminate rather than route it. + await request + .respond({ result: "error", code: "not_approved" }) + .catch(() => undefined); + await session.close().catch(() => undefined); + return; + } + })(); + } + private consumeIncoming( session: AcceptedMeshSession, handle: ConnectionHandle, @@ -228,25 +309,6 @@ export class WireMeshTransport implements MeshTransport { await request.respond({ result: "ok" }).catch(() => undefined); return; } - - if (message.method === "connect_request") { - // Held open deliberately -- see this file's own header comment. Answered later by acceptConnection/rejectConnection, not here. - this.pendingConnections.set(handle.id, { - respond: request.respond, - dataPort: message.dataPort, - name: message.name, - fingerprint: message.fingerprint, - policy: handle.policy, - }); - this.events.onConnectionRequest(handle, { - peerId: handle.id, - dataPort: message.dataPort, - name: message.name, - fingerprint: message.fingerprint, - }); - return; - } - this.route(handle, message); await request.respond({ result: "ok" }).catch(() => undefined); } @@ -273,9 +335,10 @@ export class WireMeshTransport implements MeshTransport { this.events.onBecomeCoordinator(message.peerList); return; } + case "connect_request": case "connect_accepted": case "connect_rejected": { - // Never constructed by this transport -- connectToRemote's own sendManageRequest outcome carries this meaning directly. Dropped rather than treated as an error in case a future peer still sends one (forward compatibility with anything else speaking this same opaque-frame domain). + // 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. return; } default: { @@ -295,6 +358,8 @@ export class WireMeshTransport implements MeshTransport { const wasTracked = this.peerSessions.get(deviceIdHex) === session; if (wasTracked) this.peerSessions.delete(deviceIdHex); this.allSessions.delete(session); + // A requester disconnecting before a human decides must unblock consumeQuarantined's own still-suspended loop iteration -- otherwise that promise, and the closure awaiting it, never settle. + this.pendingConnections.get(deviceIdHex)?.resolve("reject"); this.pendingConnections.delete(deviceIdHex); // A no-op when this session was never a connectToPeer dial (e.g. the coordinator-client or an accepted connection) -- Set.delete on an absent key is always safe. this.dataDials.delete(deviceIdHex); @@ -315,7 +380,7 @@ export class WireMeshTransport implements MeshTransport { this.dataListener = await this.wireTransport.listen( `${COORDINATOR_HOST}:0`, (connection) => { - void this.handleAcceptedConnection(connection, undefined, true); + void this.handleAcceptedConnection(connection, undefined, true, false); }, ); const port = this.dataListener.address.split(":").pop(); @@ -366,7 +431,12 @@ export class WireMeshTransport implements MeshTransport { `${host}:${String(port)}`, (connection) => { const tracked = this.coordinatorListeners.get(id); - void this.handleAcceptedConnection(connection, tracked?.policy, false); + void this.handleAcceptedConnection( + connection, + tracked?.policy, + false, + true, + ); }, ); this._isCoordinator = true; @@ -512,6 +582,8 @@ export class WireMeshTransport implements MeshTransport { } this.pendingConnections.delete(handle.id); await pending.respond({ result: "ok" }); + // Must happen before onIntroduction fires below: mesh-store's own handleIntroduction sends a reply on this exact handle synchronously as part of handling that event, which needs peerSessions already populated -- waiting for consumeQuarantined's own suspended loop to resume (via resolve() below) would race, since that only happens on a later microtask tick. + this.trackSession(handle.id, pending.session); const acceptedHandle: ConnectionHandle = { id: handle.id, ...(pending.policy !== undefined ? { policy: pending.policy } : {}), @@ -520,6 +592,8 @@ export class WireMeshTransport implements MeshTransport { peerId: handle.id, dataPort: pending.dataPort, }); + // Resumes consumeQuarantined's own still-blocked loop iteration so it starts trusting subsequent requests on this same session. + pending.resolve("accept"); } async rejectConnection( @@ -536,9 +610,8 @@ export class WireMeshTransport implements MeshTransport { code: "rejected", message: reason, }); - const session = this.peerSessions.get(handle.id); - this.peerSessions.delete(handle.id); - if (session !== undefined) await session.close(); + // Resumes consumeQuarantined's blocked loop iteration, which closes the session itself -- it was never tracked anywhere else to close it from here. + pending.resolve("reject"); } // ----------------------------------------------------------------------- @@ -554,7 +627,7 @@ export class WireMeshTransport implements MeshTransport { const listener = await this.wireTransport.listen( `${host}:${String(port)}`, (connection) => { - void this.handleAcceptedConnection(connection, policy, false); + void this.handleAcceptedConnection(connection, policy, false, true); }, ); this.coordinatorListeners.set(id, { @@ -599,6 +672,8 @@ export class WireMeshTransport implements MeshTransport { await pending .respond({ result: "error", code: "shutting_down" }) .catch(() => undefined); + // Unblocks consumeQuarantined's own suspended loop iteration for this session -- otherwise it never settles, even though the session itself is about to be closed below via allSessions. + pending.resolve("reject"); } this.pendingConnections.clear(); diff --git a/src/test/wire-mesh-transport-approval.integration.test.ts b/src/test/wire-mesh-transport-approval.integration.test.ts index fd1493e..3f89521 100644 --- a/src/test/wire-mesh-transport-approval.integration.test.ts +++ b/src/test/wire-mesh-transport-approval.integration.test.ts @@ -5,8 +5,17 @@ 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 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, wireWireMeshTestTransport } from "./test-transport.js"; function findFreePort(): Promise { @@ -21,17 +30,15 @@ function findFreePort(): Promise { }); } -let portOffset = 0; -async function uniquePort(): Promise { - portOffset += 10; - const base = await findFreePort(); - return base + portOffset; +/** 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 = portA + 100; + const portB = await uniquePort(); const storeA = new MeshStore(portA); wireWireMeshTestTransport(storeA); @@ -105,7 +112,7 @@ describe("WireMeshTransport connection approval", () => { void test("reject closes with reason", async () => { const portA = await uniquePort(); - const portB = portA + 100; + const portB = await uniquePort(); const storeA = new MeshStore(portA); wireWireMeshTestTransport(storeA); @@ -166,6 +173,73 @@ describe("WireMeshTransport 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); + 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", () => { From c93ae9985e26dc3f8ef097aa7a059f4e11a39328 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Fri, 11 Sep 2026 20:08:21 +0100 Subject: [PATCH 2/2] test: give mesh-smoke's own dist-requiring subprocess a dedicated script Every spawned child in this file require()s compiled dist/ output directly (exercising the actual built artifact across a real process boundary, not TS source), unlike every other *.test.ts file since the switch to running the suite straight from source via tsx. Left matching that general glob, it fails with a confusing MODULE_NOT_FOUND the moment dist/ isn't already present from an earlier, unrelated build. Renamed to mesh-smoke.runner.ts (matching delivery-receipt.runner.ts's own naming for the same reason) so plain pnpm test never picks it up, with its own pnpm test:smoke script that builds first. CI's test job runs it as an explicit step so this coverage isn't silently dropped from CI the same way federation/visibility coverage previously was. --- .github/workflows/ci.yml | 1 + package.json | 3 ++- .../{mesh-smoke.integration.test.ts => mesh-smoke.runner.ts} | 4 +++- 3 files changed, 6 insertions(+), 2 deletions(-) rename src/test/{mesh-smoke.integration.test.ts => mesh-smoke.runner.ts} (95%) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc9682b..9ad782c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -50,6 +50,7 @@ jobs: - run: pnpm install --frozen-lockfile - run: pnpm run build - run: pnpm test + - run: pnpm test:smoke pages: name: GitHub Pages diff --git a/package.json b/package.json index c333013..89c4aaf 100644 --- a/package.json +++ b/package.json @@ -54,7 +54,8 @@ "lint": "eslint .", "test": "tsx --test --test-concurrency=1 'src/test/**/*.test.ts'", "test:delivery": "tsx src/test/delivery-receipt.runner.ts", - "test:all": "pnpm test && pnpm test:delivery", + "test:smoke": "pnpm build && tsx --test src/test/mesh-smoke.runner.ts", + "test:all": "pnpm test && pnpm test:delivery && pnpm test:smoke", "test:frontend": "tsx --test src/bridges/user/web/frontend/test/**/*.unit.test.ts", "test:e2e": "playwright test", "typecheck": "tsc --noEmit" diff --git a/src/test/mesh-smoke.integration.test.ts b/src/test/mesh-smoke.runner.ts similarity index 95% rename from src/test/mesh-smoke.integration.test.ts rename to src/test/mesh-smoke.runner.ts index 9caf51c..5c2c29d 100644 --- a/src/test/mesh-smoke.integration.test.ts +++ b/src/test/mesh-smoke.runner.ts @@ -1,7 +1,9 @@ /** * 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. + * 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. + * + * Usage: pnpm test:smoke */ import * as assert from "node:assert/strict";