fix: confirm SSH forward ownership

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<number>;
-  waitForSshReady?: (tunnel: LoopbackTunnel, signal: AbortSignal) => Promise<void>;
+  sshForwardConfirmed?: (tunnel: LoopbackTunnel, signal: AbortSignal) => Promise<void>;
 }
 
 /**
@@ -232,18 +232,6 @@ async function reserveLoopbackPort(): Promise<number> {
   }
 }
 
-async function waitForLoopbackTunnel(tunnel: LoopbackTunnel, signal: AbortSignal): Promise<void> {
-  while (!signal.aborted) {
-    try {
-      await connectTcp(tunnel.host, tunnel.port, signal);
-      return;
-    } catch {
-      await new Promise<void>((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<void>((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<never>((_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<void>((resolve) => child.once?.("exit", resolve))
+            : Promise.resolve();
           child.kill("SIGTERM");
-          if (child.once) await withTimeout(request.timeoutMs, () => new Promise<void>((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}`,

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