diff --git a/backend/src/workspaces/diagnostics.ts b/backend/src/workspaces/diagnostics.ts index b37cfb5b..981faffb 100644 --- a/backend/src/workspaces/diagnostics.ts +++ b/backend/src/workspaces/diagnostics.ts @@ -160,9 +160,9 @@ export interface ConcreteDiagnosticAdapterDependencies { directProtocol?: DirectProtocolFactory; sshProcess?: SshProcessFactory; databaseClient?: DatabaseDiagnosticClientFactory; - sshSpawn?: (args: readonly string[]) => { kill(signal?: NodeJS.Signals): boolean; once?(event: "error" | "exit", listener: (...args: any[]) => void): unknown }; + sshSpawn?: (args: readonly string[]) => { kill(signal?: NodeJS.Signals): boolean; once?(event: "error" | "exit", listener: (...args: any[]) => void): unknown; stderr?: { on(event: "data", listener: (data: Buffer | string) => void): unknown; off?(event: "data", listener: (data: Buffer | string) => void): unknown } }; reserveLoopbackPort?: () => Promise; - waitForSshReady?: (tunnel: LoopbackTunnel, signal: AbortSignal) => Promise; + sshForwardConfirmed?: (tunnel: LoopbackTunnel, signal: AbortSignal) => Promise; } /** @@ -232,18 +232,6 @@ async function reserveLoopbackPort(): Promise { } } -async function waitForLoopbackTunnel(tunnel: LoopbackTunnel, signal: AbortSignal): Promise { - while (!signal.aborted) { - try { - await connectTcp(tunnel.host, tunnel.port, signal); - return; - } catch { - await new Promise((resolve) => setTimeout(resolve, 25)); - } - } - throw new Error("SSH tunnel readiness failed"); -} - /** * Concrete production adapters deliberately retain only probe metadata. Protocol failures and * response bodies are discarded at this boundary; callers receive fixed diagnostics instead. @@ -251,9 +239,8 @@ async function waitForLoopbackTunnel(tunnel: LoopbackTunnel, signal: AbortSignal export function createConcreteDiagnosticAdapters( dependencies: ConcreteDiagnosticAdapterDependencies = {}, ): DiagnosticAdapters { - const spawnSsh = dependencies.sshSpawn ?? ((args: readonly string[]) => spawn("ssh", [...args], { stdio: "ignore" })); + const spawnSsh = dependencies.sshSpawn ?? ((args: readonly string[]) => spawn("ssh", [...args], { stdio: ["ignore", "ignore", "pipe"] })); const reserveSshPort = dependencies.reserveLoopbackPort ?? reserveLoopbackPort; - const waitForSshReady = dependencies.waitForSshReady ?? waitForLoopbackTunnel; const sshProcess = dependencies.sshProcess ?? { async start(request: SshTunnelRequest, args: readonly string[]) { const port = await reserveSshPort(); @@ -266,7 +253,19 @@ export function createConcreteDiagnosticAdapters( try { await withTimeout(request.timeoutMs, async (signal) => { await Promise.race([ - waitForSshReady(tunnel, signal), + dependencies.sshForwardConfirmed + ? dependencies.sshForwardConfirmed(tunnel, signal) + : new Promise((resolve, reject) => { + const confirm = (data: Buffer | string) => { + if (new RegExp(`Local forwarding listening on 127\\.0\\.0\\.1 port ${port}\\.?`).test(data.toString())) { + child.stderr?.off?.("data", confirm); + resolve(); + } + }; + if (!child.stderr) return reject(new Error("SSH tunnel readiness failed")); + child.stderr.on("data", confirm); + signal.addEventListener("abort", () => reject(new Error("SSH tunnel readiness failed")), { once: true }); + }), new Promise((_resolve, reject) => { child.once?.("error", () => reject(new Error("SSH tunnel readiness failed"))); child.once?.("exit", () => reject(new Error("SSH tunnel readiness failed"))); @@ -282,8 +281,11 @@ export function createConcreteDiagnosticAdapters( tunnel, async close() { request.signal.removeEventListener("abort", abort); + const exited = child.once + ? new Promise((resolve) => child.once?.("exit", resolve)) + : Promise.resolve(); child.kill("SIGTERM"); - if (child.once) await withTimeout(request.timeoutMs, () => new Promise((resolve) => child.once?.("exit", resolve))).catch(() => undefined); + await withTimeout(request.timeoutMs, () => exited).catch(() => undefined); }, }; }, @@ -366,7 +368,7 @@ export function createConcreteDiagnosticAdapters( throw new Error("SSH probe failed"); } const args = [ - "-N", "-o", "BatchMode=yes", "-o", "ExitOnForwardFailure=yes", "-o", "StrictHostKeyChecking=yes", + "-N", "-v", "-o", "BatchMode=yes", "-o", "ExitOnForwardFailure=yes", "-o", "StrictHostKeyChecking=yes", "-o", `UserKnownHostsFile=${request.knownHostsFile}`, "-i", request.privateKeyFile, "-p", String(request.sshPort), "-L", `127.0.0.1:0:${request.targetHost}:${request.targetPort}`, `${request.sshUser}@${request.sshHost}`, diff --git a/backend/test/workspaces-diagnostics.test.ts b/backend/test/workspaces-diagnostics.test.ts index 11673102..f19c28a5 100644 --- a/backend/test/workspaces-diagnostics.test.ts +++ b/backend/test/workspaces-diagnostics.test.ts @@ -1,4 +1,5 @@ import { expect, test, vi } from "vitest"; +import { EventEmitter } from "node:events"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -333,7 +334,7 @@ test("provides a default bounded SSH factory through injected spawn and loopback const kill = vi.fn(() => true); const sshSpawn = vi.fn(() => ({ kill })); try { - const adapter = createConcreteDiagnosticAdapters({ sshSpawn, reserveLoopbackPort: async () => 45432, waitForSshReady: async () => undefined } as any); + const adapter = createConcreteDiagnosticAdapters({ sshSpawn, reserveLoopbackPort: async () => 45432, sshForwardConfirmed: async () => undefined } as any); await adapter.withSshTunnel({ sshHost: "bastion.example.test", sshPort: 22, sshUser: "tunnel", privateKeyFile, knownHostsFile: "/run/secrets/known-hosts", targetHost: "dwh.internal", targetPort: 5432, localHost: "127.0.0.1", localPort: 0, timeoutMs: 5000, signal: new AbortController().signal }, async () => undefined); expect(sshSpawn).toHaveBeenCalledWith(expect.arrayContaining([ "StrictHostKeyChecking=yes", "UserKnownHostsFile=/run/secrets/known-hosts", "-L", "127.0.0.1:45432:dwh.internal:5432", @@ -353,7 +354,7 @@ test("waits for SSH readiness before probing and includes ExitOnForwardFailure", const probe = vi.fn(async () => undefined); const sshSpawn = vi.fn(() => ({ kill: vi.fn(() => true) })); try { - const adapter = createConcreteDiagnosticAdapters({ sshSpawn, reserveLoopbackPort: async () => 45432, waitForSshReady: async () => await ready } as any); + const adapter = createConcreteDiagnosticAdapters({ sshSpawn, reserveLoopbackPort: async () => 45432, sshForwardConfirmed: async () => await ready } as any); const running = adapter.withSshTunnel({ sshHost: "bastion.example.test", sshPort: 22, sshUser: "tunnel", privateKeyFile, knownHostsFile: "/run/secrets/known-hosts", targetHost: "dwh.internal", targetPort: 5432, localHost: "127.0.0.1", localPort: 0, timeoutMs: 5000, signal: new AbortController().signal }, probe); await Promise.resolve(); expect(probe).not.toHaveBeenCalled(); @@ -366,6 +367,45 @@ test("waits for SSH readiness before probing and includes ExitOnForwardFailure", } }); +test("rejects unrelated listener readiness until the SSH child confirms its own forward", async () => { + const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-")); + const privateKeyFile = join(directory, "ssh-key"); + await writeFile(privateKeyFile, "test-key\n", { mode: 0o600 }); + const child = Object.assign(new EventEmitter(), { kill: vi.fn(() => true) }); + const probe = vi.fn(async () => undefined); + try { + const adapter = createConcreteDiagnosticAdapters({ sshSpawn: vi.fn(() => child), reserveLoopbackPort: async () => 45432, waitForSshReady: async () => undefined } as any); + await expect(adapter.withSshTunnel({ sshHost: "bastion.example.test", sshPort: 22, sshUser: "tunnel", privateKeyFile, knownHostsFile: "/run/secrets/known-hosts", targetHost: "dwh.internal", targetPort: 5432, localHost: "127.0.0.1", localPort: 0, timeoutMs: 20, signal: new AbortController().signal }, probe)).rejects.toThrow("SSH tunnel readiness failed"); + expect(probe).not.toHaveBeenCalled(); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + +test("permits the probe only after this SSH child confirms its forwarded port", async () => { + const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-")); + const privateKeyFile = join(directory, "ssh-key"); + await writeFile(privateKeyFile, "test-key\n", { mode: 0o600 }); + const stderr = new EventEmitter(); + const child = Object.assign(new EventEmitter(), { kill: vi.fn(() => true), stderr }); + child.kill.mockImplementation(() => { child.emit("exit", 0); return true; }); + const probe = vi.fn(async () => undefined); + try { + const adapter = createConcreteDiagnosticAdapters({ sshSpawn: vi.fn(() => child), reserveLoopbackPort: async () => 45432 } as any); + const running = adapter.withSshTunnel({ sshHost: "bastion.example.test", sshPort: 22, sshUser: "tunnel", privateKeyFile, knownHostsFile: "/run/secrets/known-hosts", targetHost: "dwh.internal", targetPort: 5432, localHost: "127.0.0.1", localPort: 0, timeoutMs: 500, signal: new AbortController().signal }, probe); + for (let attempt = 0; attempt < 20 && stderr.listenerCount("data") === 0; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 1)); + } + expect(stderr.listenerCount("data")).toBeGreaterThan(0); + expect(probe).not.toHaveBeenCalled(); + stderr.emit("data", "debug1: Local forwarding listening on 127.0.0.1 port 45432.\n"); + await running; + expect(probe).toHaveBeenCalledOnce(); + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + test("requires a matching embedding model vector and removes its unique write probe", async () => { const adapters = successfulAdapters();