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(); + }); +});