diff --git a/apps/desktop/src/lib/trpc/routers/index.ts b/apps/desktop/src/lib/trpc/routers/index.ts index ee3ac6858d7..43226a4f354 100644 --- a/apps/desktop/src/lib/trpc/routers/index.ts +++ b/apps/desktop/src/lib/trpc/routers/index.ts @@ -21,6 +21,7 @@ import { createNotificationsRouter } from "./notifications"; import { createPageContentRouter } from "./page-content"; import { createPermissionsRouter } from "./permissions"; import { createPluginsRouter } from "./plugins"; +import { createPortForwardsRouter } from "./port-forwards"; import { createPortsRouter } from "./ports"; import { createProjectsRouter } from "./projects"; import { createResourceMetricsRouter } from "./resource-metrics"; @@ -54,6 +55,7 @@ export const createAppRouter = (getWindow: () => BrowserWindow | null) => { permissions: createPermissionsRouter(), plugins: createPluginsRouter(), ports: createPortsRouter(), + portForwards: createPortForwardsRouter(), resourceMetrics: createResourceMetricsRouter(), menu: createMenuRouter(), external: createExternalRouter(), diff --git a/apps/desktop/src/lib/trpc/routers/port-forwards/index.ts b/apps/desktop/src/lib/trpc/routers/port-forwards/index.ts new file mode 100644 index 00000000000..953aa53557c --- /dev/null +++ b/apps/desktop/src/lib/trpc/routers/port-forwards/index.ts @@ -0,0 +1 @@ +export { createPortForwardsRouter } from "./port-forwards"; diff --git a/apps/desktop/src/lib/trpc/routers/port-forwards/port-forwards.ts b/apps/desktop/src/lib/trpc/routers/port-forwards/port-forwards.ts new file mode 100644 index 00000000000..231a0cd83d0 --- /dev/null +++ b/apps/desktop/src/lib/trpc/routers/port-forwards/port-forwards.ts @@ -0,0 +1,54 @@ +import { observable } from "@trpc/server/observable"; +import { portForwardManager, setRelayToken } from "main/lib/port-forward"; +import type { PortForward } from "shared/types"; +import { z } from "zod"; +import { publicProcedure, router } from "../.."; + +export const createPortForwardsRouter = () => { + return router({ + setRelayToken: publicProcedure + .input(z.object({ token: z.string().nullable() })) + .mutation(({ input }) => { + setRelayToken(input.token); + }), + + list: publicProcedure.query((): PortForward[] => portForwardManager.list()), + + subscribe: publicProcedure + .input(z.object({ clientId: z.string() })) + .subscription(({ input }) => { + return observable((emit) => { + const onChange = (forwards: PortForward[]) => emit.next(forwards); + portForwardManager.on("change", onChange); + emit.next(portForwardManager.list()); + return () => { + portForwardManager.off("change", onChange); + // The subscription is per window; its teardown is how the + // manager learns a window (and its wanted set) is gone. + void portForwardManager.releaseClient(input.clientId); + }; + }); + }), + + sync: publicProcedure + .input( + z.object({ + clientId: z.string(), + hostUrl: z.string(), + workspaceId: z.string(), + ports: z.array(z.number().int().positive()), + }), + ) + .mutation( + ({ input }): Promise => portForwardManager.sync(input), + ), + + retryEphemeral: publicProcedure + .input(z.object({ id: z.string() })) + .mutation(({ input }) => portForwardManager.retryEphemeral(input.id)), + + killLocalOwner: publicProcedure + .input(z.object({ id: z.string() })) + .mutation(({ input }) => portForwardManager.killLocalOwner(input.id)), + }); +}; diff --git a/apps/desktop/src/main/index.ts b/apps/desktop/src/main/index.ts index 2a30eb5cce9..c88a5f088b2 100644 --- a/apps/desktop/src/main/index.ts +++ b/apps/desktop/src/main/index.ts @@ -43,6 +43,7 @@ import { shutdownTanstackDbPersistence, } from "./lib/persistence/persistence"; import { syncInstalledPluginMcpServers } from "./lib/plugin-installs"; +import { portForwardManager } from "./lib/port-forward"; import { ensureProjectIconsDir, getProjectIconPath } from "./lib/project-icons"; import { runQuitCleanup } from "./lib/quit-sequence"; import { initSentry } from "./lib/sentry"; @@ -273,6 +274,9 @@ app.on("before-quit", async (event) => { } isQuitting = true; + // Local port-forward listeners hold no state worth draining; drop them so + // nothing keeps 127.0.0.1: bound after the app is gone. + portForwardManager.stopAll(); // Snapshot all open windows (bounds + org) before they close, so relaunch // restores them. markAppQuitting() stops per-window close handlers from // shrinking the set as windows close one-by-one. diff --git a/apps/desktop/src/main/lib/host-service-utils.ts b/apps/desktop/src/main/lib/host-service-utils.ts index d8d73d8a5cd..b3150736683 100644 --- a/apps/desktop/src/main/lib/host-service-utils.ts +++ b/apps/desktop/src/main/lib/host-service-utils.ts @@ -47,7 +47,7 @@ function normalizePort(port: number): number | null { return port; } -function canBindPort(port: number): Promise { +export function canBindPort(port: number): Promise { return new Promise((resolve) => { const server = createServer(); const finish = (available: boolean) => { diff --git a/apps/desktop/src/main/lib/port-forward/index.ts b/apps/desktop/src/main/lib/port-forward/index.ts new file mode 100644 index 00000000000..69d890a679d --- /dev/null +++ b/apps/desktop/src/main/lib/port-forward/index.ts @@ -0,0 +1,22 @@ +import { canBindPort } from "../host-service-utils"; +import { portManager } from "../terminal/port-manager"; +import { PortForwardManager } from "./port-forward-manager"; +import { RelayForwardTransport } from "./relay-forward-transport"; + +export { PortForwardManager } from "./port-forward-manager"; +export { RelayForwardTransport } from "./relay-forward-transport"; +export type { ForwardTransport } from "./types"; + +let relayToken: string | null = null; + +export function setRelayToken(token: string | null): void { + relayToken = token; +} + +export const portForwardManager = new PortForwardManager({ + transport: new RelayForwardTransport({ getToken: () => relayToken }), + getLocalPorts: () => portManager.getAllPorts(), + killLocalPort: ({ terminalId, workspaceId, port }) => + portManager.killPort({ terminalId, workspaceId, port }), + canBindPort, +}); diff --git a/apps/desktop/src/main/lib/port-forward/mux-session.test.ts b/apps/desktop/src/main/lib/port-forward/mux-session.test.ts new file mode 100644 index 00000000000..530e3123b80 --- /dev/null +++ b/apps/desktop/src/main/lib/port-forward/mux-session.test.ts @@ -0,0 +1,280 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import type { AddressInfo } from "node:net"; +import { + decodeMuxFrame, + decodeOpenPort, + decodeWindowDelta, + encodeClose, + encodeData, + encodeEof, + encodeHello, + encodeOpened, + encodeOpenFail, + encodePong, + encodeWindow, + MUX_INITIAL_WINDOW_BYTES, + MUX_MAX_DATA_BYTES, + MuxFrameType, + MuxOpenFailCode, +} from "@superset/shared/port-forward-mux"; +import { type WebSocket, WebSocketServer } from "ws"; +import { RelayForwardTransport } from "./relay-forward-transport"; + +/** + * A protocol-faithful mini host: echoes every stream, credits received DATA, + * and respects the desktop's WINDOW credits when echoing back. + */ +interface MiniHostStream { + sendWindow: number; + queue: Buffer[]; + eofPending: boolean; + eofSent: boolean; +} + +function startMiniHost(options?: { + skipHello?: boolean; + closeOnConnect?: boolean; + refusePort?: number; +}) { + const wss = new WebSocketServer({ host: "127.0.0.1", port: 0 }); + const sockets = new Set(); + let connections = 0; + + wss.on("connection", (ws) => { + connections++; + sockets.add(ws); + ws.on("close", () => sockets.delete(ws)); + if (options?.closeOnConnect) { + ws.close(1008, "nope"); + return; + } + if (!options?.skipHello) ws.send(encodeHello()); + + const streams = new Map(); + const drain = (id: number, s: MiniHostStream) => { + while (s.queue.length > 0 && s.sendWindow > 0) { + let chunk = s.queue.shift(); + if (!chunk) break; + const budget = Math.min(s.sendWindow, MUX_MAX_DATA_BYTES); + if (chunk.byteLength > budget) { + s.queue.unshift(chunk.subarray(budget)); + chunk = chunk.subarray(0, budget); + } + s.sendWindow -= chunk.byteLength; + ws.send(encodeData(id, chunk)); + } + if (s.eofPending && s.queue.length === 0 && !s.eofSent) { + s.eofSent = true; + ws.send(encodeEof(id)); + } + }; + + ws.on("message", (raw) => { + const data = Array.isArray(raw) + ? Buffer.concat(raw) + : Buffer.isBuffer(raw) + ? raw + : Buffer.from(raw as ArrayBuffer); + const frame = decodeMuxFrame(data); + if (!frame) return; + switch (frame.type) { + case MuxFrameType.ping: + ws.send(encodePong()); + return; + case MuxFrameType.open: { + const port = decodeOpenPort(frame); + if (port !== null && port === options?.refusePort) { + ws.send( + encodeOpenFail( + frame.streamId, + MuxOpenFailCode.portNotOwned, + "port not owned by workspace", + ), + ); + return; + } + streams.set(frame.streamId, { + sendWindow: MUX_INITIAL_WINDOW_BYTES, + queue: [], + eofPending: false, + eofSent: false, + }); + ws.send(encodeOpened(frame.streamId)); + return; + } + case MuxFrameType.data: { + const s = streams.get(frame.streamId); + if (!s) return; + // Credit first (we consumed it), then echo within window. + ws.send(encodeWindow(frame.streamId, frame.payload.byteLength)); + s.queue.push(Buffer.from(frame.payload)); + drain(frame.streamId, s); + return; + } + case MuxFrameType.window: { + const s = streams.get(frame.streamId); + const delta = decodeWindowDelta(frame); + if (s && delta !== null) { + s.sendWindow += delta; + drain(frame.streamId, s); + } + return; + } + case MuxFrameType.eof: { + const s = streams.get(frame.streamId); + if (!s) return; + s.eofPending = true; + drain(frame.streamId, s); + return; + } + case MuxFrameType.close: { + const s = streams.get(frame.streamId); + if (s) { + streams.delete(frame.streamId); + ws.send(encodeClose(frame.streamId)); + } + return; + } + default: + return; + } + }); + }); + + const port = (wss.address() as AddressInfo).port; + return { + hostUrl: `http://127.0.0.1:${port}`, + connectionCount: () => connections, + close: async () => { + for (const s of sockets) s.terminate(); + // Bun ships its own `ws` shim whose server close callback does not + // fire while a just-terminated client lingers; don't await it. + wss.close(); + await new Promise((r) => setTimeout(r, 25)); + }, + }; +} + +function transportFor(): RelayForwardTransport { + return new RelayForwardTransport({ + getToken: () => "token", + fetchFn: async () => + new Response(JSON.stringify({ ok: true, proto: 2 }), { status: 200 }), + }); +} + +function target(hostUrl: string, remotePort = 3000) { + return { hostUrl, workspaceId: "ws1", remotePort }; +} + +const hosts: { close: () => Promise }[] = []; +afterEach(async () => { + while (hosts.length) await hosts.pop()?.close(); +}); + +function miniHost(options?: Parameters[0]) { + const host = startMiniHost(options); + hosts.push(host); + return host; +} + +describe("RelayForwardTransport over a mux session", () => { + test("a stream opens and echoes bytes both ways", async () => { + const host = miniHost(); + const transport = transportFor(); + const stream = await transport.openStream(target(host.hostUrl)); + stream.write(Buffer.from("marco")); + const echoed = await new Promise((r) => stream.once("data", r)); + expect(echoed.toString()).toBe("marco"); + stream.destroy(); + }); + + test("many streams share one WebSocket connection", async () => { + const host = miniHost(); + const transport = transportFor(); + const a = await transport.openStream(target(host.hostUrl, 3000)); + const b = await transport.openStream(target(host.hostUrl, 4000)); + a.write(Buffer.from("aaa")); + b.write(Buffer.from("bbb")); + const [fromA, fromB] = await Promise.all([ + new Promise((r) => a.once("data", r)), + new Promise((r) => b.once("data", r)), + ]); + expect(fromA.toString()).toBe("aaa"); + expect(fromB.toString()).toBe("bbb"); + expect(host.connectionCount()).toBe(1); + a.destroy(); + b.destroy(); + }); + + test("a transfer larger than the window completes intact", async () => { + const host = miniHost(); + const transport = transportFor(); + const stream = await transport.openStream(target(host.hostUrl)); + const size = MUX_INITIAL_WINDOW_BYTES * 3; + const payload = Buffer.alloc(size); + for (let i = 0; i < size; i++) payload[i] = i % 251; + + const received: Buffer[] = []; + let total = 0; + const done = new Promise((resolve) => { + stream.on("data", (chunk: Buffer) => { + received.push(chunk); + total += chunk.byteLength; + if (total >= size) resolve(); + }); + }); + stream.write(payload); + await done; + expect(Buffer.concat(received).equals(payload)).toBe(true); + stream.destroy(); + }); + + test("OPEN_FAIL surfaces as a rejected openStream", async () => { + const host = miniHost({ refusePort: 5432 }); + const transport = transportFor(); + expect(transport.openStream(target(host.hostUrl, 5432))).rejects.toThrow( + "port not owned by workspace", + ); + }); + + test("an old host closing before HELLO names the real problem", async () => { + const host = miniHost({ closeOnConnect: true }); + const transport = transportFor(); + expect(transport.openStream(target(host.hostUrl))).rejects.toThrow( + "Host does not support port forwarding", + ); + }); + + test("ending the local side sends EOF and the echo finishes", async () => { + const host = miniHost(); + const transport = transportFor(); + const stream = await transport.openStream(target(host.hostUrl)); + const chunks: Buffer[] = []; + const ended = new Promise((r) => stream.on("end", r)); + stream.on("data", (c: Buffer) => chunks.push(c)); + stream.end(Buffer.from("last words")); + await ended; + expect(Buffer.concat(chunks).toString()).toBe("last words"); + }); + + test("a dead session rejects, and the next open dials fresh", async () => { + const host = miniHost(); + const transport = transportFor(); + const stream = await transport.openStream(target(host.hostUrl)); + // The manager always attaches an error handler; without one a dead + // session's destroy(err) would crash instead of closing. + stream.on("error", () => {}); + const closed = new Promise((r) => stream.on("close", r)); + await host.close(); + await closed; + + const revived = miniHost(); + // Same transport, new host on a new port: fresh session per key. + const stream2 = await transport.openStream(target(revived.hostUrl)); + stream2.write(Buffer.from("back")); + const echoed = await new Promise((r) => stream2.once("data", r)); + expect(echoed.toString()).toBe("back"); + stream2.destroy(); + }); +}); diff --git a/apps/desktop/src/main/lib/port-forward/mux-session.ts b/apps/desktop/src/main/lib/port-forward/mux-session.ts new file mode 100644 index 00000000000..aa691aa3173 --- /dev/null +++ b/apps/desktop/src/main/lib/port-forward/mux-session.ts @@ -0,0 +1,378 @@ +import { Duplex } from "node:stream"; +import { + decodeHelloVersion, + decodeMuxFrame, + decodeOpenFail, + decodeWindowDelta, + encodeClose, + encodeData, + encodeEof, + encodeOpen, + encodePing, + encodeWindow, + MUX_INITIAL_WINDOW_BYTES, + MUX_MAX_DATA_BYTES, + MUX_MAX_STREAMS, + MUX_PING_INTERVAL_MS, + MUX_PONG_TIMEOUT_MS, + MUX_PROTOCOL_VERSION, + type MuxFrame, + MuxFrameType, +} from "@superset/shared/port-forward-mux"; +import { WebSocket } from "ws"; + +const HELLO_TIMEOUT_MS = 15_000; +const OPEN_TIMEOUT_MS = 20_000; +/** A session with no live streams closes itself after this long. */ +const IDLE_CLOSE_MS = 60_000; + +interface StreamRecord { + duplex: Duplex; + /** Bytes we may still send toward the host before it credits us. */ + sendWindow: number; + /** Chunks waiting on send-window; the head may be partially sent. */ + sendQueue: { chunk: Buffer; done: (err?: Error | null) => void }[]; + headOffset: number; + /** Received bytes not yet credited back to the host. */ + uncredited: number; + creditScheduled: boolean; + /** push() returned false; credits pause until the consumer reads. */ + readPaused: boolean; + settleOpen: { + resolve: (d: Duplex) => void; + reject: (err: Error) => void; + timer: ReturnType; + } | null; + remoteClosed: boolean; +} + +/** + * The desktop end of one port-forward mux session — one WebSocket through + * the relay per (host, workspace), every forwarded TCP connection a numbered + * stream inside it (@superset/shared/port-forward-mux). Opening a connection + * costs one OPEN frame on this warm pipe instead of a relay dial-back. + */ +export class MuxSession { + /** Resolves once the host's HELLO proves it speaks the protocol. */ + readonly ready: Promise; + private readonly ws: WebSocket; + private readonly streams = new Map(); + private nextStreamId = 1; + private dead = false; + private helloSeen = false; + private lastPongAt = Date.now(); + private pingTimer: ReturnType | null = null; + private idleTimer: ReturnType | null = null; + + constructor( + url: string, + private readonly options: { onClosed: () => void }, + ) { + this.ws = new WebSocket(url); + this.ready = new Promise((resolve, reject) => { + const helloTimer = setTimeout(() => { + this.ws.terminate(); + reject(new Error("Host did not answer")); + }, HELLO_TIMEOUT_MS); + this.ws.once("unexpected-response", (_req, res) => { + clearTimeout(helloTimer); + reject(new Error(describeUpgradeFailure(res.statusCode))); + }); + this.ws.once("error", (err) => { + clearTimeout(helloTimer); + reject(err); + }); + this.ws.on("message", (data) => { + const frame = decodeMuxFrame(toUint8(data)); + if (!frame) return; + if (!this.helloSeen) { + if (frame.type !== MuxFrameType.hello) return; + const version = decodeHelloVersion(frame); + clearTimeout(helloTimer); + if (version !== MUX_PROTOCOL_VERSION) { + this.ws.close(1002, "protocol version mismatch"); + reject(new Error("Host speaks an unsupported forwarding protocol")); + return; + } + this.helloSeen = true; + this.startTimers(); + this.armIdleClose(); + resolve(); + return; + } + this.dispatch(frame); + }); + this.ws.on("close", () => { + clearTimeout(helloTimer); + const reason = this.helloSeen + ? new Error("Forward tunnel disconnected") + : // An older host-service has no /fwd route; the relay pairs + // the stream, then the host's own server refuses it. + new Error( + "Host does not support port forwarding (update the host-service)", + ); + this.destroySession(reason); + reject(reason); + }); + }); + // A session that never becomes ready surfaces through openStream; + // nothing awaits `ready` bare. + this.ready.catch(() => {}); + } + + get isDead(): boolean { + return this.dead; + } + + streamCount(): number { + return this.streams.size; + } + + openStream(port: number): Promise { + if (this.dead) { + return Promise.reject(new Error("Forward tunnel disconnected")); + } + if (this.streams.size >= MUX_MAX_STREAMS) { + return Promise.reject(new Error("Too many concurrent connections")); + } + this.cancelIdleClose(); + const id = this.nextStreamId++; + return new Promise((resolve, reject) => { + const record: StreamRecord = { + duplex: null as unknown as Duplex, + sendWindow: MUX_INITIAL_WINDOW_BYTES, + sendQueue: [], + headOffset: 0, + uncredited: 0, + creditScheduled: false, + readPaused: false, + settleOpen: { + resolve, + reject, + timer: setTimeout(() => { + this.streams.delete(id); + this.armIdleClose(); + reject(new Error("Host did not answer")); + }, OPEN_TIMEOUT_MS), + }, + remoteClosed: false, + }; + record.duplex = this.createDuplex(id, record); + this.streams.set(id, record); + this.ws.send(encodeOpen(id, port)); + }); + } + + close(): void { + this.ws.close(1000, "session closed"); + } + + private createDuplex(id: number, record: StreamRecord): Duplex { + const session = this; + return new Duplex({ + write(chunk: Buffer, _enc, cb) { + record.sendQueue.push({ chunk, done: cb }); + session.pumpSend(id, record); + }, + read() { + record.readPaused = false; + session.scheduleCredit(id, record); + }, + final(cb) { + if (!session.dead && !record.remoteClosed) { + session.ws.send(encodeEof(id)); + } + cb(); + }, + destroy(err, cb) { + if (session.streams.get(id) === record) { + session.streams.delete(id); + if (!session.dead && !record.remoteClosed) { + session.ws.send(encodeClose(id)); + } + session.armIdleClose(); + } + for (const pending of record.sendQueue) { + pending.done(err ?? new Error("stream destroyed")); + } + record.sendQueue = []; + cb(err); + }, + }); + } + + private dispatch(frame: MuxFrame): void { + if (frame.type === MuxFrameType.pong) { + this.lastPongAt = Date.now(); + return; + } + const record = this.streams.get(frame.streamId); + // Frames racing our CLOSE for an already-dropped stream are normal. + if (!record) return; + + switch (frame.type) { + case MuxFrameType.opened: { + if (record.settleOpen) { + clearTimeout(record.settleOpen.timer); + const { resolve } = record.settleOpen; + record.settleOpen = null; + resolve(record.duplex); + } + return; + } + case MuxFrameType.openFail: { + this.streams.delete(frame.streamId); + this.armIdleClose(); + if (record.settleOpen) { + clearTimeout(record.settleOpen.timer); + const { reject } = record.settleOpen; + record.settleOpen = null; + reject(new Error(decodeOpenFail(frame).message)); + } + return; + } + case MuxFrameType.data: { + const chunk = Buffer.from(frame.payload); + record.uncredited += chunk.byteLength; + if (!record.duplex.push(chunk)) { + record.readPaused = true; + } + this.scheduleCredit(frame.streamId, record); + return; + } + case MuxFrameType.window: { + const delta = decodeWindowDelta(frame); + if (delta !== null) { + record.sendWindow += delta; + this.pumpSend(frame.streamId, record); + } + return; + } + case MuxFrameType.eof: { + record.duplex.push(null); + return; + } + case MuxFrameType.close: { + record.remoteClosed = true; + this.streams.delete(frame.streamId); + this.armIdleClose(); + if (record.settleOpen) { + clearTimeout(record.settleOpen.timer); + const { reject } = record.settleOpen; + record.settleOpen = null; + reject(new Error("Connection refused by host")); + return; + } + record.duplex.destroy(); + return; + } + default: + return; + } + } + + private pumpSend(id: number, record: StreamRecord): void { + if (this.dead) return; + while (record.sendQueue.length > 0 && record.sendWindow > 0) { + const head = record.sendQueue[0]; + const remaining = head.chunk.byteLength - record.headOffset; + const len = Math.min(remaining, record.sendWindow, MUX_MAX_DATA_BYTES); + this.ws.send( + encodeData( + id, + head.chunk.subarray(record.headOffset, record.headOffset + len), + ), + ); + record.sendWindow -= len; + record.headOffset += len; + if (record.headOffset === head.chunk.byteLength) { + record.sendQueue.shift(); + record.headOffset = 0; + // Completing the write callback only once the bytes are inside + // the window is what propagates backpressure to the local socket. + head.done(); + } + } + } + + // Credits are flushed once per tick so a burst of DATA frames becomes one + // WINDOW frame. A paused reader holds credits entirely; _read releases. + private scheduleCredit(id: number, record: StreamRecord): void { + if (record.creditScheduled) return; + record.creditScheduled = true; + setImmediate(() => { + record.creditScheduled = false; + if (this.dead || this.streams.get(id) !== record) return; + if (record.readPaused) return; + if (record.uncredited > 0) { + this.ws.send(encodeWindow(id, record.uncredited)); + record.uncredited = 0; + } + }); + } + + private startTimers(): void { + this.pingTimer = setInterval(() => { + if (Date.now() - this.lastPongAt > MUX_PONG_TIMEOUT_MS) { + this.ws.terminate(); + return; + } + if (this.ws.readyState === WebSocket.OPEN) { + this.ws.send(encodePing()); + } + }, MUX_PING_INTERVAL_MS); + this.pingTimer.unref?.(); + } + + private armIdleClose(): void { + if (this.dead || this.streams.size > 0) return; + this.cancelIdleClose(); + this.idleTimer = setTimeout(() => { + if (this.streams.size === 0) this.close(); + }, IDLE_CLOSE_MS); + this.idleTimer.unref?.(); + } + + private cancelIdleClose(): void { + if (this.idleTimer) clearTimeout(this.idleTimer); + this.idleTimer = null; + } + + private destroySession(reason: Error): void { + if (this.dead) return; + this.dead = true; + if (this.pingTimer) clearInterval(this.pingTimer); + this.cancelIdleClose(); + for (const [id, record] of this.streams) { + this.streams.delete(id); + if (record.settleOpen) { + clearTimeout(record.settleOpen.timer); + record.settleOpen.reject(reason); + } else { + record.duplex.destroy(reason); + } + } + this.options.onClosed(); + } +} + +function toUint8(data: Buffer | ArrayBuffer | Buffer[]): Uint8Array { + if (Buffer.isBuffer(data)) return data; + if (Array.isArray(data)) return Buffer.concat(data); + return new Uint8Array(data); +} + +export function describeUpgradeFailure(status: number | undefined): string { + switch (status) { + case 401: + return "Session expired, sign in again"; + case 403: + return "No access to this host"; + case 503: + return "Host is offline"; + case 504: + return "Host did not answer"; + default: + return `Relay refused the stream (${status ?? "unknown"})`; + } +} diff --git a/apps/desktop/src/main/lib/port-forward/port-forward-manager.test.ts b/apps/desktop/src/main/lib/port-forward/port-forward-manager.test.ts new file mode 100644 index 00000000000..460ea9be756 --- /dev/null +++ b/apps/desktop/src/main/lib/port-forward/port-forward-manager.test.ts @@ -0,0 +1,375 @@ +import { describe, expect, test } from "bun:test"; +import net from "node:net"; +import { type Duplex, PassThrough } from "node:stream"; +import type { DetectedPort } from "@superset/port-scanner"; +import type { ForwardTarget } from "shared/types"; +import { PortForwardManager } from "./port-forward-manager"; +import type { ForwardTransport } from "./types"; + +const HOST = "https://relay.test/hosts/org:machine"; + +async function startEcho(): Promise<{ port: number; close: () => void }> { + const server = net.createServer((s) => s.pipe(s)); + const port = await new Promise((resolve) => + server.listen(0, "127.0.0.1", () => + resolve((server.address() as net.AddressInfo).port), + ), + ); + return { port, close: () => server.close() }; +} + +async function freePort(): Promise { + const server = net.createServer(); + return new Promise((resolve) => + server.listen(0, "127.0.0.1", () => { + const { port } = server.address() as net.AddressInfo; + server.close(() => resolve(port)); + }), + ); +} + +/** A transport that opens a TCP connection to an echo server on this machine. */ +function echoTransport(echoPort: number): ForwardTransport & { + opened: ForwardTarget[]; +} { + const opened: ForwardTarget[] = []; + return { + kind: "relay", + opened, + probe: async () => {}, + openStream: async (target) => { + opened.push(target); + const socket = net.connect({ host: "127.0.0.1", port: echoPort }); + await new Promise((r) => socket.once("connect", () => r())); + return socket as Duplex; + }, + }; +} + +function manager( + transport: ForwardTransport, + overrides: Partial[0]> = {}, +) { + return new PortForwardManager({ + transport, + getLocalPorts: () => [], + killLocalPort: async () => ({ success: true }), + canBindPort: async () => true, + ...overrides, + }); +} + +async function roundTrip(port: number, payload: string): Promise { + const socket = net.connect({ host: "127.0.0.1", port }); + await new Promise((r) => socket.once("connect", () => r())); + socket.write(payload); + // TCP may split the echo across chunks; collect until it is all here. + const expected = Buffer.byteLength(payload); + const data = await new Promise((resolve) => { + const chunks: Buffer[] = []; + let total = 0; + socket.on("data", (chunk: Buffer) => { + chunks.push(chunk); + total += chunk.byteLength; + if (total >= expected) resolve(Buffer.concat(chunks)); + }); + }); + socket.destroy(); + return data.toString(); +} + +describe("PortForwardManager", () => { + test("sync listens on the requested port and bridges bytes", async () => { + const echo = await startEcho(); + const transport = echoTransport(echo.port); + const m = manager(transport); + const local = await freePort(); + const [fwd] = await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [local], + }); + expect(fwd?.status).toEqual({ state: "active", localPort: local }); + expect(await roundTrip(local, "ping")).toBe("ping"); + expect(transport.opened[0]).toEqual({ + hostUrl: HOST, + workspaceId: "ws1", + remotePort: local, + }); + m.stopAll(); + echo.close(); + }); + + test("sync for another workspace stops the previous forwards", async () => { + const echo = await startEcho(); + const m = manager(echoTransport(echo.port)); + const a = await freePort(); + await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [a], + }); + const b = await freePort(); + const list = await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws2", + ports: [b], + }); + expect(list.map((f) => f.target.workspaceId)).toEqual(["ws2"]); + expect(await freeToBind(a)).toBe(true); + m.stopAll(); + echo.close(); + }); + + test("a busy local port reports busy with the owner when known", async () => { + const echo = await startEcho(); + const owner: DetectedPort = { + port: echo.port, + pid: 42, + processName: "node", + terminalId: "t1", + workspaceId: "local-ws", + detectedAt: 0, + address: "127.0.0.1", + }; + const m = manager(echoTransport(echo.port), { + getLocalPorts: () => [owner], + }); + const [fwd] = await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [echo.port], + }); + expect(fwd?.status).toEqual({ + state: "busy", + localPort: echo.port, + localOwner: { + pid: 42, + processName: "node", + terminalId: "t1", + workspaceId: "local-ws", + }, + }); + m.stopAll(); + echo.close(); + }); + + test("retryEphemeral moves a busy forward to another local port", async () => { + const echo = await startEcho(); + const m = manager(echoTransport(echo.port)); + const [fwd] = await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [echo.port], + }); + expect(fwd?.status.state).toBe("busy"); + const retried = await m.retryEphemeral(fwd?.id ?? ""); + expect(retried?.status.state).toBe("active"); + if (retried?.status.state !== "active") throw new Error("not active"); + expect(retried.status.localPort).not.toBe(echo.port); + expect(await roundTrip(retried.status.localPort, "x")).toBe("x"); + m.stopAll(); + echo.close(); + }); + + test("killLocalOwner restarts the forward once the port frees", async () => { + const echo = await startEcho(); + let killed = false; + const m = manager(echoTransport(echo.port), { + getLocalPorts: () => [ + { + port: echo.port, + pid: 1, + processName: "node", + terminalId: "t1", + workspaceId: "local-ws", + detectedAt: 0, + address: "127.0.0.1", + }, + ], + killLocalPort: async () => { + killed = true; + echo.close(); + return { success: true }; + }, + canBindPort: freeToBind, + }); + const [fwd] = await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [echo.port], + }); + const result = await m.killLocalOwner(fwd?.id ?? ""); + expect(killed).toBe(true); + expect(result).toEqual({ success: true }); + expect(m.list()[0]?.status).toEqual({ + state: "active", + localPort: echo.port, + }); + m.stopAll(); + }); + + test("a failed probe yields an error status", async () => { + const m = manager({ + kind: "relay", + probe: async () => { + throw new Error("protocol v1"); + }, + openStream: async () => new PassThrough(), + }); + const [fwd] = await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [await freePort()], + }); + expect(fwd?.status).toEqual({ state: "error", message: "protocol v1" }); + m.stopAll(); + }); + + test("stopAll releases every listener", async () => { + const echo = await startEcho(); + const m = manager(echoTransport(echo.port)); + const local = await freePort(); + await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [local], + }); + m.stopAll(); + expect(m.list()).toEqual([]); + expect(await freeToBind(local)).toBe(true); + echo.close(); + }); + + test("a sync during the probe never orphans a listener", async () => { + const local = await freePort(); + let releaseProbe: () => void = () => {}; + const transport: ForwardTransport = { + kind: "relay", + probe: () => + new Promise((resolve) => { + releaseProbe = resolve; + }), + openStream: async () => new PassThrough() as unknown as Duplex, + }; + const m = manager(transport); + const first = m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [local], + }); + // Deselect before the probe answers. + await m.sync({ clientId: "w1", hostUrl: HOST, workspaceId: "", ports: [] }); + releaseProbe(); + await first; + expect(m.list()).toEqual([]); + expect(await freeToBind(local)).toBe(true); + }); + + test("a later successful stream restores active after a failure", async () => { + const echo = await startEcho(); + let fail = true; + const echoing = echoTransport(echo.port); + const transport: ForwardTransport = { + kind: "relay", + probe: async () => {}, + openStream: (target) => { + if (fail) return Promise.reject(new Error("relay hiccup")); + return echoing.openStream(target); + }, + }; + const m = manager(transport); + const local = await freePort(); + await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [local], + }); + // The first connection fails and poisons the status. + const s = net.connect({ host: "127.0.0.1", port: local }); + await new Promise((r) => s.once("close", r)); + expect(m.list()[0]?.status.state).toBe("error"); + // The listener is still bound; the next connection heals it. + fail = false; + expect(await roundTrip(local, "healed")).toBe("healed"); + expect(m.list()[0]?.status).toEqual({ + state: "active", + localPort: local, + }); + m.stopAll(); + echo.close(); + }); + + test("two clients wanting different forwards do not fight", async () => { + const echo = await startEcho(); + const m = manager(echoTransport(echo.port)); + const a = await freePort(); + const b = await freePort(); + await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [a], + }); + await m.sync({ + clientId: "w2", + hostUrl: HOST, + workspaceId: "ws2", + ports: [b], + }); + // Both forwards run; neither sync tore the other down. + expect( + m + .list() + .map((f) => f.target.remotePort) + .sort(), + ).toEqual([a, b].sort()); + // Releasing one client stops only its forward. + await m.releaseClient("w1"); + expect(m.list().map((f) => f.target.remotePort)).toEqual([b]); + expect(await freeToBind(a)).toBe(true); + m.stopAll(); + echo.close(); + }); + + test("a forward both clients want survives either releasing it", async () => { + const echo = await startEcho(); + const m = manager(echoTransport(echo.port)); + const a = await freePort(); + await m.sync({ + clientId: "w1", + hostUrl: HOST, + workspaceId: "ws1", + ports: [a], + }); + await m.sync({ + clientId: "w2", + hostUrl: HOST, + workspaceId: "ws1", + ports: [a], + }); + await m.releaseClient("w1"); + expect(m.list()).toHaveLength(1); + expect(await roundTrip(a, "still here")).toBe("still here"); + m.stopAll(); + echo.close(); + }); +}); + +function freeToBind(port: number): Promise { + return new Promise((resolve) => { + const server = net.createServer(); + server.once("error", () => resolve(false)); + server.listen(port, "127.0.0.1", () => server.close(() => resolve(true))); + }); +} diff --git a/apps/desktop/src/main/lib/port-forward/port-forward-manager.ts b/apps/desktop/src/main/lib/port-forward/port-forward-manager.ts new file mode 100644 index 00000000000..674aa41345e --- /dev/null +++ b/apps/desktop/src/main/lib/port-forward/port-forward-manager.ts @@ -0,0 +1,348 @@ +import { EventEmitter } from "node:events"; +import net from "node:net"; +import type { DetectedPort } from "@superset/port-scanner"; +import type { + ForwardTarget, + LocalPortOwner, + PortForward, + PortForwardStatus, +} from "shared/types"; +import { portForwardId } from "shared/types"; +import type { ForwardTransport } from "./types"; + +interface ForwardEntry { + forward: PortForward; + server: net.Server | null; + sockets: Set; +} + +export interface PortForwardManagerOptions { + transport: ForwardTransport; + /** Local ports the desktop's own port scanner attributes to a workspace. */ + getLocalPorts: () => DetectedPort[]; + killLocalPort: (owner: LocalPortOwner & { port: number }) => Promise<{ + success: boolean; + error?: string; + }>; + canBindPort: (port: number) => Promise; +} + +const MAX_PENDING_FRAMES = 64; +const MAX_PENDING_BYTES = 4 * 1024 * 1024; +const KILL_WAIT_MS = 3_000; +const KILL_POLL_MS = 250; + +/** + * Owns every local listener that forwards to a remote workspace port. Each + * window is a client with its own wanted set (its selected workspace's remote + * ports); the manager keeps exactly the union running, so two windows on + * different workspaces don't tear each other's forwards down. A client's set + * is dropped when its window's subscription goes away. + */ +export class PortForwardManager extends EventEmitter<{ + change: [PortForward[]]; +}> { + private readonly entries = new Map(); + private readonly wantedByClient = new Map< + string, + Map + >(); + + constructor(private readonly options: PortForwardManagerOptions) { + super(); + } + + list(): PortForward[] { + return Array.from(this.entries.values(), (e) => ({ ...e.forward })); + } + + async sync({ + clientId, + hostUrl, + workspaceId, + ports, + }: { + clientId: string; + hostUrl: string; + workspaceId: string; + ports: number[]; + }): Promise { + const wanted = new Map(); + for (const remotePort of new Set(ports)) { + const target = { hostUrl, workspaceId, remotePort }; + wanted.set(portForwardId(target), target); + } + if (wanted.size > 0) { + this.wantedByClient.set(clientId, wanted); + } else { + this.wantedByClient.delete(clientId); + } + return this.reconcile(); + } + + /** A window went away; whatever only it wanted stops. */ + async releaseClient(clientId: string): Promise { + if (!this.wantedByClient.delete(clientId)) return; + await this.reconcile(); + } + + private async reconcile(): Promise { + const union = new Map(); + for (const wanted of this.wantedByClient.values()) { + for (const [id, target] of wanted) union.set(id, target); + } + for (const id of Array.from(this.entries.keys())) { + if (!union.has(id)) this.stop(id); + } + await Promise.all( + Array.from(union.entries()) + .filter(([id]) => !this.entries.has(id)) + .map(([, target]) => this.start(target)), + ); + return this.list(); + } + + async retryEphemeral(id: string): Promise { + const entry = this.entries.get(id); + if (!entry) return null; + if (entry.forward.status.state !== "busy") return { ...entry.forward }; + await this.listen({ entry, localPort: 0 }); + if (this.entries.get(id) !== entry) { + this.closeServer(entry); + return null; + } + return { ...entry.forward }; + } + + async killLocalOwner( + id: string, + ): Promise<{ success: boolean; error?: string }> { + const entry = this.entries.get(id); + if (!entry) return { success: false, error: "Forward not found" }; + const { status } = entry.forward; + if (status.state !== "busy" || !status.localOwner) { + return { success: false, error: "Local process is not known" }; + } + const port = entry.forward.target.remotePort; + const result = await this.options.killLocalPort({ + ...status.localOwner, + port, + }); + if (!result.success) return result; + const deadline = Date.now() + KILL_WAIT_MS; + while (Date.now() < deadline) { + if (this.entries.get(id) !== entry) { + return { success: false, error: "Forward was stopped" }; + } + if (await this.options.canBindPort(port)) { + await this.listen({ entry, localPort: port }); + if (this.entries.get(id) !== entry) this.closeServer(entry); + return { success: true }; + } + await new Promise((r) => setTimeout(r, KILL_POLL_MS)); + } + return { success: false, error: "Local port is still busy" }; + } + + stopAll(): void { + for (const id of Array.from(this.entries.keys())) this.stop(id); + } + + private async start(target: ForwardTarget): Promise { + const id = portForwardId(target); + const entry: ForwardEntry = { + forward: { + id, + target, + status: { state: "error", message: "Starting" }, + transport: this.options.transport.kind, + connections: 0, + }, + server: null, + sockets: new Set(), + }; + this.entries.set(id, entry); + try { + await this.options.transport.probe(target); + } catch (err) { + this.setStatus({ + entry, + status: { state: "error", message: messageOf(err) }, + }); + return; + } + // A sync during the probe may have stopped this forward; binding now + // would orphan a listener nothing owns. + if (this.entries.get(id) !== entry) return; + await this.listen({ entry, localPort: target.remotePort }); + } + + private listen({ + entry, + localPort, + }: { + entry: ForwardEntry; + localPort: number; + }): Promise { + this.closeServer(entry); + return new Promise((resolve) => { + const server = net.createServer((socket) => + this.accept({ entry, socket }), + ); + entry.server = server; + server.once("error", (err: NodeJS.ErrnoException) => { + entry.server = null; + if (err.code === "EADDRINUSE") { + this.setStatus({ + entry, + status: { + state: "busy", + localPort, + localOwner: this.findLocalOwner(localPort), + }, + }); + } else { + this.setStatus({ + entry, + status: { state: "error", message: messageOf(err) }, + }); + } + resolve(); + }); + server.listen(localPort, "127.0.0.1", () => { + const address = server.address(); + const bound = + address && typeof address === "object" ? address.port : localPort; + this.setStatus({ + entry, + status: { state: "active", localPort: bound }, + }); + resolve(); + }); + }); + } + + private accept({ + entry, + socket, + }: { + entry: ForwardEntry; + socket: net.Socket; + }): void { + entry.sockets.add(socket); + entry.forward.connections = entry.sockets.size; + this.emitChange(); + socket.on("close", () => { + entry.sockets.delete(socket); + entry.forward.connections = entry.sockets.size; + this.emitChange(); + }); + socket.on("error", () => socket.destroy()); + + // Bytes the client sends while the relay stream is still opening. A + // consumer must be attached from the first tick: an unread socket drops + // data under Bun, and a browser sends its request right after connect. + const pending: Buffer[] = []; + let pendingBytes = 0; + const buffer = (chunk: Buffer) => { + pending.push(chunk); + pendingBytes += chunk.byteLength; + if ( + pending.length > MAX_PENDING_FRAMES || + pendingBytes > MAX_PENDING_BYTES + ) { + socket.destroy(); + } + }; + socket.on("data", buffer); + + this.options.transport + .openStream(entry.forward.target) + .then((stream) => { + socket.off("data", buffer); + if (socket.destroyed) { + stream.destroy(); + return; + } + // A transient stream failure flips the forward to error; the + // listener is still bound, so the next success restores it. + if ( + entry.forward.status.state === "error" && + this.entries.get(entry.forward.id) === entry && + entry.server + ) { + const address = entry.server.address(); + const bound = + address && typeof address === "object" + ? address.port + : entry.forward.target.remotePort; + this.setStatus({ + entry, + status: { state: "active", localPort: bound }, + }); + } + stream.on("error", () => socket.destroy()); + stream.on("close", () => socket.destroy()); + socket.on("close", () => stream.destroy()); + for (const chunk of pending) stream.write(chunk); + pending.length = 0; + socket.pipe(stream); + stream.pipe(socket); + }) + .catch((err) => { + socket.destroy(); + if (this.entries.get(entry.forward.id) === entry) { + this.setStatus({ + entry, + status: { state: "error", message: messageOf(err) }, + }); + } + }); + } + + private findLocalOwner(port: number): LocalPortOwner | null { + const match = this.options.getLocalPorts().find((p) => p.port === port); + if (!match) return null; + return { + pid: match.pid, + processName: match.processName, + terminalId: match.terminalId, + workspaceId: match.workspaceId, + }; + } + + private stop(id: string): void { + const entry = this.entries.get(id); + if (!entry) return; + this.entries.delete(id); + this.closeServer(entry); + this.emitChange(); + } + + private closeServer(entry: ForwardEntry): void { + for (const socket of entry.sockets) socket.destroy(); + entry.sockets.clear(); + entry.forward.connections = 0; + entry.server?.close(); + entry.server = null; + } + + private setStatus({ + entry, + status, + }: { + entry: ForwardEntry; + status: PortForwardStatus; + }): void { + entry.forward.status = status; + this.emitChange(); + } + + private emitChange(): void { + this.emit("change", this.list()); + } +} + +function messageOf(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} diff --git a/apps/desktop/src/main/lib/port-forward/relay-forward-transport.ts b/apps/desktop/src/main/lib/port-forward/relay-forward-transport.ts new file mode 100644 index 00000000000..a53a9c66528 --- /dev/null +++ b/apps/desktop/src/main/lib/port-forward/relay-forward-transport.ts @@ -0,0 +1,83 @@ +import type { Duplex } from "node:stream"; +import type { ForwardTarget } from "shared/types"; +import { MuxSession } from "./mux-session"; +import type { ForwardTransport } from "./types"; + +export interface RelayForwardTransportOptions { + getToken: () => string | null; + fetchFn?: (input: string) => Promise; +} + +/** + * Forwards over one mux session per (host, workspace), spliced through + * relay2's per-stream dial-back exactly like a terminal stream. The first + * connection after selecting a workspace pays the relay dial; every later + * connection is an OPEN frame on the warm session. Only protocol v2 relays + * can carry binary client frames, so the probe refuses v1 (`/health` + * without `proto: 2`). + */ +export class RelayForwardTransport implements ForwardTransport { + readonly kind = "relay" as const; + private readonly probes = new Map>(); + private readonly sessions = new Map(); + + constructor(private readonly options: RelayForwardTransportOptions) {} + + async probe(target: ForwardTarget): Promise { + const origin = new URL(target.hostUrl).origin; + let pending = this.probes.get(origin); + if (!pending) { + pending = this.checkProtocol(origin).catch((err) => { + // Let a transient failure retry on the next sync. + this.probes.delete(origin); + throw err; + }); + this.probes.set(origin, pending); + } + await pending; + // Establish the mux session now, so a host without forwarding support + // errors the row at sync time and the first connection is one OPEN + // frame on an already-warm pipe instead of a relay dial. + await this.session(target).ready; + } + + private async checkProtocol(origin: string): Promise { + const fetchFn = this.options.fetchFn ?? fetch; + const res = await fetchFn(`${origin}/health`); + if (!res.ok) throw new Error(`Relay health check failed (${res.status})`); + const body = (await res.json()) as { proto?: number }; + if (body.proto !== 2) { + throw new Error( + "Host relay does not support port forwarding (protocol v1)", + ); + } + } + + async openStream(target: ForwardTarget): Promise { + const session = this.session(target); + await session.ready; + return session.openStream(target.remotePort); + } + + private session(target: ForwardTarget): MuxSession { + const key = `${target.hostUrl}|${target.workspaceId}`; + const existing = this.sessions.get(key); + if (existing && !existing.isDead) return existing; + + const token = this.options.getToken(); + if (!token) throw new Error("Not signed in"); + const url = new URL(`${target.hostUrl}/fwd`); + if (url.protocol === "http:") url.protocol = "ws:"; + if (url.protocol === "https:") url.protocol = "wss:"; + url.searchParams.set("workspaceId", target.workspaceId); + url.searchParams.set("token", token); + + const session = new MuxSession(url.toString(), { + onClosed: () => { + if (this.sessions.get(key) === session) this.sessions.delete(key); + }, + }); + this.sessions.set(key, session); + return session; + } +} diff --git a/apps/desktop/src/main/lib/port-forward/types.ts b/apps/desktop/src/main/lib/port-forward/types.ts new file mode 100644 index 00000000000..c8f67d53b13 --- /dev/null +++ b/apps/desktop/src/main/lib/port-forward/types.ts @@ -0,0 +1,20 @@ +import type { Duplex } from "node:stream"; +import type { ForwardTarget, ForwardTransportKind } from "shared/types"; + +/** + * How bytes get from the local machine to 127.0.0.1: on the host. + * The relay implementation ships first; a direct (Tailscale, VPN, LAN) or SSH + * implementation plugs in here without touching the manager or the UI. + */ +export interface ForwardTransport { + readonly kind: ForwardTransportKind; + /** + * Resolves when the transport can serve this target; rejects with a + * reason. Also where a transport warms anything it wants ready before the + * first connection (the relay transport establishes its mux session here, + * so an unsupported host errors the row at sync time, not on first click). + */ + probe(target: ForwardTarget): Promise; + /** One bidirectional byte stream to 127.0.0.1: on the host. */ + openStream(target: ForwardTarget): Promise; +} diff --git a/apps/desktop/src/renderer/lib/auth-client.ts b/apps/desktop/src/renderer/lib/auth-client.ts index 24845787bb5..81badbedd44 100644 --- a/apps/desktop/src/renderer/lib/auth-client.ts +++ b/apps/desktop/src/renderer/lib/auth-client.ts @@ -10,6 +10,7 @@ import { createAuthClient } from "better-auth/react"; import { useSyncExternalStore } from "react"; import { env } from "renderer/env.renderer"; import { decodeJwtExpiresAtMs } from "renderer/lib/jwt-expiry"; +import { electronTrpcClient } from "renderer/lib/trpc-client"; let authToken: string | null = null; const authTokenListeners = new Set<() => void>(); @@ -54,6 +55,13 @@ export function setJwt(token: string | null) { jwt = token; jwtGeneration++; jwtExpiresAtMs = token ? decodeJwtExpiresAtMs(token) : null; + // The main process dials the relay for port forwards with this same token + // and has no auth client of its own; keep its copy current. + electronTrpcClient.portForwards.setRelayToken + .mutate({ token }) + .catch((err) => { + console.warn("[auth] failed to hand relay token to main:", err); + }); } function jwtIsFresh(): boolean { diff --git a/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/components/DashboardSidebarWorkspaceItem/components/DashboardSidebarExpandedWorkspaceRow/components/DashboardSidebarWorkspaceChips/components/DashboardSidebarPortsChip/components/DashboardSidebarPortHoverRow/DashboardSidebarPortHoverRow.tsx b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/components/DashboardSidebarWorkspaceItem/components/DashboardSidebarExpandedWorkspaceRow/components/DashboardSidebarWorkspaceChips/components/DashboardSidebarPortsChip/components/DashboardSidebarPortHoverRow/DashboardSidebarPortHoverRow.tsx index 5078c807977..83085954af8 100644 --- a/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/components/DashboardSidebarWorkspaceItem/components/DashboardSidebarExpandedWorkspaceRow/components/DashboardSidebarWorkspaceChips/components/DashboardSidebarPortsChip/components/DashboardSidebarPortHoverRow/DashboardSidebarPortHoverRow.tsx +++ b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/components/DashboardSidebarWorkspaceItem/components/DashboardSidebarExpandedWorkspaceRow/components/DashboardSidebarWorkspaceChips/components/DashboardSidebarPortsChip/components/DashboardSidebarPortHoverRow/DashboardSidebarPortHoverRow.tsx @@ -3,6 +3,9 @@ import { LuX } from "react-icons/lu"; import { useDashboardSidebarPortKill } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/hooks/useDashboardSidebarPortKill"; import type { DashboardSidebarPort } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/hooks/useDashboardSidebarPortsData"; import { usePortOpenActions } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/hooks/usePortOpenActions"; +import { usePortForward } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/providers/PortForwardsProvider"; +import { formatPortRowLabel } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/utils/formatPortRowLabel"; +import { PortForwardBusyActions } from "renderer/routes/_authenticated/_dashboard/components/PortForwardBusyActions"; import { STROKE_WIDTH } from "renderer/screens/main/components/WorkspaceSidebar/constants"; interface DashboardSidebarPortHoverRowProps { @@ -15,6 +18,8 @@ export function DashboardSidebarPortHoverRow({ const { t } = useLingui(); const { isPending, killPort } = useDashboardSidebarPortKill(); const { openPrimary } = usePortOpenActions(port); + const forward = usePortForward(port); + const address = formatPortRowLabel({ port, forward }); return (
@@ -28,11 +33,14 @@ export function DashboardSidebarPortHoverRow({ {port.label && ( {port.label} )} - - {"localhost:"} - {port.port} + + {address.text} + {forward && } {/* Always in layout, shown via visibility so the row never changes size on hover. */} + )} + + + ); +} diff --git a/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/PortForwardBusyActions/index.ts b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/PortForwardBusyActions/index.ts new file mode 100644 index 00000000000..938de52bd12 --- /dev/null +++ b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/PortForwardBusyActions/index.ts @@ -0,0 +1 @@ +export { PortForwardBusyActions } from "./PortForwardBusyActions"; diff --git a/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/RemotePortForwarder/RemotePortForwarder.tsx b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/RemotePortForwarder/RemotePortForwarder.tsx new file mode 100644 index 00000000000..f79dc82a880 --- /dev/null +++ b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/RemotePortForwarder/RemotePortForwarder.tsx @@ -0,0 +1,19 @@ +import { useRouterState } from "@tanstack/react-router"; +import { useRemotePortForwarding } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/hooks/useRemotePortForwarding"; + +const V2_WORKSPACE_PATH = /^\/v2-workspace\/([^/]+)/; + +/** + * Renders nothing; keeps the main process's port forwards in step with the + * workspace on screen. Reads the live pathname itself, so only this leaf + * re-renders on navigation, and forwards stop the moment the user leaves a + * v2 workspace route. + */ +export function RemotePortForwarder() { + const pathname = useRouterState({ + select: (state) => state.location.pathname, + }); + const activeWorkspaceId = V2_WORKSPACE_PATH.exec(pathname)?.[1] ?? null; + useRemotePortForwarding(activeWorkspaceId); + return null; +} diff --git a/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/RemotePortForwarder/index.ts b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/RemotePortForwarder/index.ts new file mode 100644 index 00000000000..0ca98e445c6 --- /dev/null +++ b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/RemotePortForwarder/index.ts @@ -0,0 +1 @@ +export { RemotePortForwarder } from "./RemotePortForwarder"; diff --git a/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/TopBar/components/TopBarPortsDropdown/components/TopBarPortRow/TopBarPortRow.tsx b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/TopBar/components/TopBarPortsDropdown/components/TopBarPortRow/TopBarPortRow.tsx index 12355b469a2..0e62e36b307 100644 --- a/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/TopBar/components/TopBarPortsDropdown/components/TopBarPortRow/TopBarPortRow.tsx +++ b/apps/desktop/src/renderer/routes/_authenticated/_dashboard/components/TopBar/components/TopBarPortsDropdown/components/TopBarPortRow/TopBarPortRow.tsx @@ -6,6 +6,9 @@ import { useCopyToClipboard } from "renderer/hooks/useCopyToClipboard"; import { useDashboardSidebarPortKill } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/hooks/useDashboardSidebarPortKill"; import type { DashboardSidebarPort } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/hooks/useDashboardSidebarPortsData"; import { usePortOpenActions } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/hooks/usePortOpenActions"; +import { usePortForward } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/providers/PortForwardsProvider"; +import { formatPortRowLabel } from "renderer/routes/_authenticated/_dashboard/components/DashboardSidebar/utils/formatPortRowLabel"; +import { PortForwardBusyActions } from "renderer/routes/_authenticated/_dashboard/components/PortForwardBusyActions"; import { STROKE_WIDTH } from "renderer/screens/main/components/WorkspaceSidebar/constants"; interface TopBarPortRowProps { @@ -25,6 +28,8 @@ export function TopBarPortRow({ port, onNavigate }: TopBarPortRowProps) { const { canOpenInBrowser, portUrl, openExternal, openPrimary } = usePortOpenActions(port); const { copyToClipboard } = useCopyToClipboard(); + const forward = usePortForward(port); + const address = formatPortRowLabel({ port, forward }); const description = [port.label, port.processName] .filter(Boolean) @@ -40,8 +45,8 @@ export function TopBarPortRow({ port, onNavigate }: TopBarPortRowProps) { onNavigate(); }; - // Only offered for local-device ports: `portUrl` is always a localhost - // address, which isn't reachable for a remote host's port. + // Only offered when `portUrl` reaches this machine: a local port, or a + // remote one the main process forwards. const handleCopy = async () => { try { await copyToClipboard(portUrl); @@ -82,10 +87,18 @@ export function TopBarPortRow({ port, onNavigate }: TopBarPortRowProps) { :{port.port} - - {description || `localhost:${port.port}`} + + {description ? `${description} · ${address.text}` : address.text} + {forward?.status.state === "busy" && ( +
+ +
+ )}
{canOpenInBrowser && (