From ab541c95d1ae0083c851ce968385fa83317a25ec Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 16 Sep 2026 23:20:23 +0100 Subject: [PATCH] feat(core): dial remote connections over ws or wss URLs connectToRemote's host argument now also accepts a ws or wss URL, dialling a WebSocket-served hub (e.g. the mesh.exadev.io cloudflare-hub, which answers wss only) instead of raw TLS -- the port is meaningless in URL form. Any other URL scheme is refused up front rather than surfacing as an opaque DNS error from the TLS dial treating the whole URL as a hostname. The URL branch drives the identical session + connect_request flow the TLS branch uses, proven by an integration test against a minimal in-test fake hub speaking the real protocol. The dial is agent-comms' own edge adapter (src/core/ws-dial.ts, the same layering its TLS adapter occupies) rather than a wire-mesh-node dependency: wire-mesh-node is not published to npm as a consumable package (a placeholder 0.0.0). It frames identically -- one CBOR frame per binary WebSocket message -- so a hub sees the same bytes from this client as from any other. Closes #149 --- src/core/connection-approval.ts | 2 +- src/core/tool.ts | 5 +- src/core/wire-mesh-transport.ts | 18 ++- src/core/ws-dial.ts | 139 +++++++++++++++++ src/test/ws-dial.test.ts | 107 +++++++++++++ .../wss-remote-connect.integration.test.ts | 144 ++++++++++++++++++ 6 files changed, 410 insertions(+), 5 deletions(-) create mode 100644 src/core/ws-dial.ts create mode 100644 src/test/ws-dial.test.ts create mode 100644 src/test/wss-remote-connect.integration.test.ts diff --git a/src/core/connection-approval.ts b/src/core/connection-approval.ts index f6c7eff7..5b24452e 100644 --- a/src/core/connection-approval.ts +++ b/src/core/connection-approval.ts @@ -98,7 +98,7 @@ export class ConnectionApproval { })); } - /** Initiate an outbound connection to a remote coordinator requiring approval. Fires the connect_request and returns immediately. The connection completes asynchronously when the coordinator accepts or rejects. */ + /** Initiate an outbound connection to a remote coordinator requiring approval. Fires the connect_request and returns immediately. The connection completes asynchronously when the coordinator accepts or rejects. `host` is either a hostname (TLS-dialled with `port`) or a ws:// / wss:// URL (WebSocket-dialled to a hub such as mesh.exadev.io; the port is ignored). */ async connectToRemote(host: string, port: number): Promise { const peerId = this.deps.getPeerId(); const agent = this.deps.agents.get(peerId); diff --git a/src/core/tool.ts b/src/core/tool.ts index 71993a9a..5d6e5e9e 100644 --- a/src/core/tool.ts +++ b/src/core/tool.ts @@ -603,7 +603,10 @@ export class CommsTool { const connectToRemote = this.store.connectToRemote.bind(this.store); return tryMeshAction("connect", async () => { await connectToRemote(action.host, action.port); - return `Connection request sent to ${action.host}:${String(action.port)}.`; + const target = /^wss?:\/\//.test(action.host) + ? action.host + : `${action.host}:${String(action.port)}`; + return `Connection request sent to ${target}.`; }); } diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 1306c0e5..cd64911b 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -9,6 +9,7 @@ */ import { createTlsTransport } from "wire-mesh-core/adapters/tls-transport"; +import { connectWsUrl } from "./ws-dial.js"; import { acceptMeshSession, type AcceptedMeshSession, @@ -797,9 +798,20 @@ export class WireMeshTransport implements MeshTransport { name: string, fingerprint: string, ): Promise { - const connection = await this.wireTransport.connect( - `${host}:${String(port)}`, - ); + // A ws:// or wss:// URL in the host position dials a WebSocket-served + // hub (e.g. the mesh.exadev.io cloudflare-hub) instead of raw TLS -- + // the port is meaningless in URL form, so callers pass 0. Any other URL + // scheme is refused here rather than surfacing as an opaque DNS error + // from the TLS dial treating the whole URL as a hostname. + const isWsUrl = /^wss?:\/\//.test(host); + if (!isWsUrl && host.includes("://")) { + throw new Error( + `expected a hostname or a ws:// / wss:// URL, got "${host}"`, + ); + } + const connection = isWsUrl + ? await connectWsUrl(host) + : await this.wireTransport.connect(`${host}:${String(port)}`); const identity = await this.identityReady; const session = await acceptMeshSession(connection, identity, [DOMAIN], { onFrame: async (conn, frame) => this.handleDataFrame(conn, frame), diff --git a/src/core/ws-dial.ts b/src/core/ws-dial.ts new file mode 100644 index 00000000..b491e50d --- /dev/null +++ b/src/core/ws-dial.ts @@ -0,0 +1,139 @@ +// A ws-library dial adapter for outbound remote connections to a WebSocket-served hub (e.g. the mesh.exadev.io cloudflare-hub, which answers wss:// only): one CBOR frame per binary WebSocket message, the identical framing wire-mesh-node's own WebSocket transport uses, so a hub sees the same bytes from this client as from any other. Agent-comms' own edge adapter rather than a wire-mesh-node dependency because wire-mesh-node is not published to npm as a consumable package (a placeholder 0.0.0), while the dial is genuinely this app's concern at its own edge -- the same layering its TLS transport adapter already occupies. + +import { WebSocket as WsSocket } from "ws"; +import { cdeDecodeOptions, cdeEncodeOptions, decode, encode } from "cbor2"; +import { frameSchema, type Frame } from "wire-mesh-core/generated/protocol"; +import type { Connection } from "wire-mesh-core/ports/transport"; + +const CONNECT_TIMEOUT_MS = 10_000; +// RFC 6455 close codes, named rather than bare: 1000 normal closure, 1002 protocol error. +const CLOSE_NORMAL = 1000; +const CLOSE_PROTOCOL_ERROR = 1002; + +/** Rejects addresses that are neither ws:// nor wss:// URLs -- a wrong-scheme dial would otherwise surface as an opaque socket error. */ +function assertWsUrl(url: string): void { + if (!/^wss?:\/\/./.test(url)) { + throw new Error(`expected a ws:// or wss:// URL, got "${url}"`); + } +} + +export async function connectWsUrl(url: string): Promise { + assertWsUrl(url); + return new Promise((resolve, reject) => { + const socket = new WsSocket(url); + socket.binaryType = "arraybuffer"; + const timer = setTimeout(() => { + socket.terminate(); + reject( + new Error( + `ws dial timed out after ${String(CONNECT_TIMEOUT_MS)}ms: ${url}`, + ), + ); + }, CONNECT_TIMEOUT_MS); + socket.once("open", () => { + clearTimeout(timer); + resolve(wrapSocket(socket)); + }); + socket.once("error", (error) => { + clearTimeout(timer); + reject(error instanceof Error ? error : new Error(String(error))); + }); + }); +} + +function wrapSocket(socket: WsSocket): Connection { + const pending: Frame[] = []; + const waiters: { + resolve: (result: IteratorResult) => void; + reject: (error: unknown) => void; + }[] = []; + let ended = false; + let failure: Error | null = null; + + function endAll(): void { + ended = true; + for (const waiter of waiters.splice(0)) { + waiter.resolve({ value: undefined, done: true }); + } + } + + function failAll(error: Error): void { + failure = error; + ended = true; + for (const waiter of waiters.splice(0)) { + waiter.reject(error); + } + } + + socket.on("message", (data) => { + if (!(data instanceof ArrayBuffer)) { + failAll(new Error("expected a binary WebSocket message")); + socket.close(CLOSE_PROTOCOL_ERROR, "protocol error"); + return; + } + let frame: Frame; + try { + const decoded: unknown = decode(new Uint8Array(data), cdeDecodeOptions); + const parsed = frameSchema.safeParse(decoded); + if (!parsed.success) { + // A decodable-but-unrecognised frame is dropped, keeping the connection -- version negotiation exists to tolerate it. + return; + } + frame = parsed.data; + } catch (error) { + failAll( + error instanceof Error + ? error + : new Error(`frame body failed to decode: ${String(error)}`), + ); + socket.close(CLOSE_PROTOCOL_ERROR, "protocol error"); + return; + } + const waiter = waiters.shift(); + if (waiter) { + waiter.resolve({ value: frame, done: false }); + } else { + pending.push(frame); + } + }); + socket.on("close", endAll); + socket.on("error", () => { + // The receive stream's failure signal; the raw error is not otherwise actionable here. + }); + + const receiveStream: AsyncIterable = { + [Symbol.asyncIterator]() { + return { + async next(): Promise> { + const next = pending.shift(); + if (next !== undefined) { + return { value: next, done: false }; + } + if (failure !== null) { + throw failure; + } + if (ended) { + return { value: undefined, done: true }; + } + return new Promise>((resolve, reject) => { + waiters.push({ resolve, reject }); + }); + }, + }; + }, + }; + + return { + async send(frame: Frame): Promise { + if (ended) { + throw new Error("connection is closed"); + } + socket.send(new Uint8Array(encode(frame, cdeEncodeOptions))); + }, + receive: () => receiveStream, + close: async () => { + socket.close(CLOSE_NORMAL); + return Promise.resolve(); + }, + }; +} diff --git a/src/test/ws-dial.test.ts b/src/test/ws-dial.test.ts new file mode 100644 index 00000000..97e1853f --- /dev/null +++ b/src/test/ws-dial.test.ts @@ -0,0 +1,107 @@ +import { createServer } from "node:http"; +import { WebSocketServer, type WebSocket as WsSocket } from "ws"; +import { afterEach, describe, expect, it } from "vitest"; +import { cdeDecodeOptions, cdeEncodeOptions, decode, encode } from "cbor2"; +import type { Frame } from "wire-mesh-core/generated/protocol"; +import { connectWsUrl } from "../core/ws-dial.js"; + +const SHUTDOWN_GRACE_MS = 250; + +interface TestServer { + url: string; + close: () => Promise; + received: Frame[]; +} + +/** A real local ws server echoing every received frame back, so the dial round trip is exercised against a genuine socket. */ +async function echoServer(): Promise { + const received: Frame[] = []; + const http = createServer(); + const wss = new WebSocketServer({ server: http }); + wss.on("connection", (socket: WsSocket) => { + socket.on("message", (data) => { + const decodedFrame: unknown = decode( + new Uint8Array(data as ArrayBuffer), + cdeDecodeOptions, + ); + const frame = decodedFrame as Frame; + received.push(frame); + socket.send(new Uint8Array(encode(frame, cdeEncodeOptions))); + }); + }); + await new Promise((resolve) => { + http.listen(0, "127.0.0.1", () => { + resolve(); + }); + }); + const address = http.address(); + if (address === null || typeof address === "string") { + throw new Error("expected a TCP listen address"); + } + return { + url: `ws://127.0.0.1:${String(address.port)}/`, + received, + close: async () => + new Promise((resolve) => { + // wss.close() alone waits for connected clients, which would + // deadlock a test asserting behaviour AFTER the server side hangs + // up -- terminate them first, then close the listener. + for (const client of wss.clients) { + client.terminate(); + } + wss.close(); + http.close(() => { + resolve(); + }); + setTimeout(resolve, SHUTDOWN_GRACE_MS); + }), + }; +} + +const servers: TestServer[] = []; + +afterEach(async () => { + for (const server of servers.splice(0)) { + await server.close(); + } +}); + +describe("connectWsUrl", () => { + it("round-trips a frame against a real ws server", async () => { + const server = await echoServer(); + servers.push(server); + const connection = await connectWsUrl(server.url); + const ping: Frame = { type: "ping" }; + await connection.send(ping); + + const iterator = connection.receive()[Symbol.asyncIterator](); + const first = await iterator.next(); + expect(first.done).toBe(false); + expect(first.value).toEqual(ping); + expect(server.received).toEqual([ping]); + + await connection.close(); + }); + + it("rejects a non-ws/wss URL outright, without a socket attempt", async () => { + await expect(connectWsUrl("ftp://example.com/")).rejects.toThrow( + /ws:\/\/ or wss:\/\//, + ); + await expect(connectWsUrl("example.com:1234")).rejects.toThrow( + /ws:\/\/ or wss:\/\//, + ); + }); + + it("ends its receive stream when the server closes the socket", async () => { + const server = await echoServer(); + const connection = await connectWsUrl(server.url); + await connection.send({ type: "ping" }); + const iterator = connection.receive()[Symbol.asyncIterator](); + await iterator.next(); // the echo + // Close the server first: wss.close() itself waits for client sockets, + // so closing the connection first would deadlock the shutdown. + await server.close(); + const afterClose = await iterator.next(); + expect(afterClose.done).toBe(true); + }); +}); diff --git a/src/test/wss-remote-connect.integration.test.ts b/src/test/wss-remote-connect.integration.test.ts new file mode 100644 index 00000000..a4441fe2 --- /dev/null +++ b/src/test/wss-remote-connect.integration.test.ts @@ -0,0 +1,144 @@ +// Integration: connectToRemote against a ws:// URL, through a minimal in-test fake hub speaking the real protocol (handshake reply, connect_request answered with manage-ok) -- proving the URL branch drives the identical session + connect_request flow the TLS branch uses, without standing up a TLS stack or a deployed hub. + +import { createServer, type Server } from "node:http"; +import { WebSocketServer, type WebSocket as WsSocket } from "ws"; +import { afterEach, describe, expect, it } from "vitest"; +import { cdeDecodeOptions, cdeEncodeOptions, decode, encode } from "cbor2"; +import type { Frame } from "wire-mesh-core/generated/protocol"; +import { generateIdentity } from "../core/identity.js"; +import { WireMeshTransport } from "../core/wire-mesh-transport.js"; +import type { TransportEvents } from "../core/transport.js"; + +const SHUTDOWN_GRACE_MS = 250; +const GARBAGE_PORT = 0; // meaningless in URL form -- exactly what the branch should tolerate + +function inertEvents(): TransportEvents { + return { + onMessage: () => undefined, + onPeerConnected: () => undefined, + onPeerDisconnected: () => undefined, + onIntroduction: () => undefined, + onConnectionRequest: () => undefined, + onPeerList: () => undefined, + onPeerJoined: () => undefined, + onBecomeCoordinator: () => undefined, + onRevocationAnnounce: () => undefined, + onPresenceAdvert: () => undefined, + }; +} + +/** The minimal fake hub: replies to a handshake with a compatible one, answers every manage-request with manage-ok. Records every frame for assertions. */ +async function fakeHub(): Promise<{ + url: string; + received: Frame[]; + close: () => Promise; +}> { + const received: Frame[] = []; + const http: Server = createServer(); + const wss = new WebSocketServer({ server: http }); + wss.on("connection", (socket: WsSocket) => { + socket.on("message", (data) => { + const decodedFrame: unknown = decode( + new Uint8Array(data as ArrayBuffer), + cdeDecodeOptions, + ); + const frame = decodedFrame as Frame; + received.push(frame); + if (frame.type === "handshake") { + const answer: Frame = { + type: "handshake", + version: frame.version, + domains: frame.domains, + }; + socket.send(new Uint8Array(encode(answer, cdeEncodeOptions))); + return; + } + if (frame.type === "manage-request") { + const response: Frame = { + type: "manage-response", + "request-id": frame["request-id"], + outcome: { result: "ok" }, + }; + socket.send(new Uint8Array(encode(response, cdeEncodeOptions))); + } + // Gossip and everything else: tolerated, unanswered -- the hub's own forwarding is not this test's subject. + }); + }); + await new Promise((resolve) => { + http.listen(0, "127.0.0.1", () => { + resolve(); + }); + }); + const address = http.address(); + if (address === null || typeof address === "string") { + throw new Error("expected a TCP listen address"); + } + return { + url: `ws://127.0.0.1:${String(address.port)}/`, + received, + close: async () => + new Promise((resolve) => { + // wss.close() alone waits for connected clients; terminate them first so shutdown cannot deadlock on the very connection under test. + for (const client of wss.clients) { + client.terminate(); + } + wss.close(); + http.close(() => { + resolve(); + }); + setTimeout(resolve, SHUTDOWN_GRACE_MS); + }), + }; +} + +const cleanups: (() => Promise)[] = []; + +afterEach(async () => { + for (const close of cleanups.splice(0)) { + await close(); + } +}); + +describe("connectToRemote with a ws:// URL", () => { + it("dials the hub, handshakes, and sends connect_request through the same flow as TLS", async () => { + const hub = await fakeHub(); + cleanups.push(hub.close); + const identity = generateIdentity(); + const transport = new WireMeshTransport(inertEvents(), identity); + + await transport.connectToRemote( + hub.url, + GARBAGE_PORT, + "peer-id", + 0, + "test-agent", + "", + ); + + // The fake hub saw the full flow the TLS branch also drives: this side's handshake, then the connect_request manage-request. + const sawHandshake = hub.received.some((f) => f.type === "handshake"); + const connectRequest = hub.received.find( + (f) => f.type === "manage-request", + ); + expect(sawHandshake).toBe(true); + expect(connectRequest).toBeDefined(); + + await transport.shutdown(); + }); + + it("rejects a non-ws/wss URL host outright rather than dialling garbage", async () => { + const identity = generateIdentity(); + const transport = new WireMeshTransport(inertEvents(), identity); + await expect( + transport.connectToRemote( + "ftp://example.com/", + GARBAGE_PORT, + "peer-id", + 0, + "test-agent", + "", + ), + ).rejects.toThrow(/hostname or a ws:\/\/ \/ wss:\/\//); + await transport.shutdown(); + }); +});