fix(link): gate join credential on tunnel survival

src/client/link-join.ts

@@ -1,6 +1,7 @@
 import { randomBytes } from "node:crypto";
 import { hostname } from "node:os";
 import { findAvailablePort } from "../server/ports";
+import { scanListenPidsForAddress, type ListenPidScan } from "../server/port-reclaim";
 import { isLinkPort } from "../link/ports";
 import { buildExecArgv } from "../link/ssh-argv";
 import type { SshRunner } from "../link/ssh-runner";
@@ -22,6 +23,7 @@ import type { OcxConnectedClientId } from "../types";
 
 const JOIN_TUNNEL_READY_TIMEOUT_MS = 15_000;
 const JOIN_TUNNEL_POLL_MS = 100;
+const JOIN_TUNNEL_SPAWN_GRACE_MS = 100;
 const JOIN_REVOKE_TIMEOUT_MS = 30_000;
 const JOIN_CONFIRM_TTL_MS = 5 * 60_000;
 const LINK_ID = /^lnk_[0-9a-f]{16}$/;
@@ -71,6 +73,12 @@ export interface ClientLinkJoinDeps {
   hostname?: () => string;
   randomBytes?: (size: number) => Uint8Array;
   fetchImpl?: typeof fetch;
+  /**
+   * LISTEN-owner probe for the tunnel port; defaults to the netstat/lsof/ss scan.
+   * Receives the loopback address the tunnel binds so listeners on unrelated
+   * addresses do not confuse the readiness check.
+   */
+  scanListenPids?: (port: number, address?: string) => ListenPidScan;
   spawnTunnel?: (spec: {
     linkId: string;
     alias: string;
@@ -203,20 +211,56 @@ async function compensateStaleSidecar(deps: ClientLinkJoinDeps): Promise<void> {
 
 async function waitForReady(
   deps: ClientLinkJoinDeps,
+  tunnel: ClientLinkTunnelHandle,
   port: number,
   key: string,
 ): Promise<void> {
   const fetchImpl = deps.fetchImpl ?? fetch;
   const now = deps.now ?? Date.now;
   const sleep = deps.sleep ?? ((ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms)));
   const deadline = now() + JOIN_TUNNEL_READY_TIMEOUT_MS;
+  const tunnelExited = tunnel.exited.then(() => { throw new ClientLinkJoinError("join_tunnel_failed"); });
+  await Promise.race([
+    tunnelExited,
+    new Promise<void>(resolve => setTimeout(resolve, JOIN_TUNNEL_SPAWN_GRACE_MS)),
+  ]);
+  const listenPids = deps.scanListenPids ?? scanListenPidsForAddress;
+  // The tunnel binds 127.0.0.1; a listener on a different loopback or interface address
+  // never receives our requests, so ownership is only judged among sockets that serve it.
+  const tunnelAddress = "127.0.0.1";
   for (;;) {
     try {
-      const response = await fetchImpl(`http://127.0.0.1:${port}/readyz`, {
-        headers: { "x-opencodex-api-key": key },
-      });
-      if (response.status === 200) return;
-      if (response.status === 401) throw new ClientLinkJoinError("admission_failed");
+      // A squatter answering the 401 challenge would otherwise collect the issued key:
+      // the only listener allowed a keyed request is the ssh process we spawned — it owns
+      // the port only after a successful bind, and ExitOnForwardFailure makes it exit when
+      // it cannot take the port. An unverifiable scan stays "not ready", never a pass.
+      const ownership = listenPids(port, tunnelAddress);
+      if (ownership.ok && ownership.pids.length === 1 && ownership.pids[0] === tunnel.pid) {
+        // Never follow redirects: a port occupant must not reroute the challenge, and a
+        // redirected keyed request would carry the issued key to an unrelated listener.
+        const probe = await Promise.race([
+          tunnelExited,
+          fetchImpl(`http://127.0.0.1:${port}/readyz`, { redirect: "manual" }),
+        ]);
+        if (probe.status === 401) {
+          // Ownership can flip between the probe and the keyed request (a squatter
+          // takes the port after the tunnel dies). Re-scan in the same iteration and
+          // skip the keyed request if the port is no longer solely the tunnel's.
+          const recheck = listenPids(port, tunnelAddress);
+          if (!recheck.ok || recheck.pids.length !== 1 || recheck.pids[0] !== tunnel.pid) {
+            continue;
+          }
+          const response = await Promise.race([
+            tunnelExited,
+            fetchImpl(`http://127.0.0.1:${port}/readyz`, {
+              headers: { "x-opencodex-api-key": key },
+              redirect: "manual",
+            }),
+          ]);
+          if (response.status === 200) return;
+          if (response.status === 401) throw new ClientLinkJoinError("admission_failed");
+        }
+      }
     } catch (error) {
       if (error instanceof ClientLinkJoinError) throw error;
     }
@@ -287,30 +331,36 @@ export async function joinHome(deps: ClientLinkJoinDeps, input: { alias: string
       configDir: deps.configDir,
       knownHostsFile: deps.knownHostsFile,
     });
-    await waitForReady(deps, tunnelPort, issued.key);
+    await waitForReady(deps, tunnel, tunnelPort, issued.key);
   } catch (error) {
     const code = error instanceof ClientLinkJoinError ? error.code : "join_tunnel_failed";
     await rollback(deps, issued.linkId, tunnel);
     throw new ClientLinkJoinError(code);
   }
 
   try {
+    if (!tunnel) throw new ClientLinkJoinError("join_tunnel_failed");
     const connect = deps.connect ?? connectClient;
-    await connect({
-      serverUrl: `http://127.0.0.1:${tunnelPort}`,
-      managementUrl: `http://127.0.0.1:${tunnelPort}`,
-      credential: { kind: "link", apiKeyId: issued.apiKeyId, key: issued.key },
-      transport: "link",
-      link: { tunnelPort, linkId: issued.linkId },
-      selectedClients: deps.selectedClients ?? ["codex", "claude"],
-      managementTransport: "direct",
-    }, {
-      fetchImpl: deps.fetchImpl,
-      ...deps.connectDeps,
-    });
-  } catch {
+    // Keep watching the tunnel until the connection commits: an exited tunnel
+    // must not let the issued key ride out to whatever next holds the port.
+    await Promise.race([
+      tunnel.exited.then(() => { throw new ClientLinkJoinError("join_tunnel_failed"); }),
+      connect({
+        serverUrl: `http://127.0.0.1:${tunnelPort}`,
+        managementUrl: `http://127.0.0.1:${tunnelPort}`,
+        credential: { kind: "link", apiKeyId: issued.apiKeyId, key: issued.key },
+        transport: "link",
+        link: { tunnelPort, linkId: issued.linkId },
+        selectedClients: deps.selectedClients ?? ["codex", "claude"],
+        managementTransport: "direct",
+      }, {
+        fetchImpl: deps.fetchImpl,
+        ...deps.connectDeps,
+      }),
+    ]);
+  } catch (error) {
     await rollback(deps, issued.linkId, tunnel);
-    throw new ClientLinkJoinError("join_connect_failed");
+    throw new ClientLinkJoinError(error instanceof ClientLinkJoinError ? error.code : "join_connect_failed");
   }
 
   await stopTunnel(tunnel);

src/server/port-reclaim.ts

@@ -16,6 +16,16 @@ export type ListenPidScan =
   | { ok: true; pids: number[] }
   | { ok: false; error?: string };
 
+/** One listening socket with its bound local address (host part only). */
+export interface ListenEntry {
+  pid: number;
+  address: string;
+}
+
+export type ListenEntryScan =
+  | { ok: true; listeners: ListenEntry[] }
+  | { ok: false; error?: string };
+
 export type ReclaimListenPortOptions = WaitForPortOptions & {
   /**
    * When true AND `onlyKillPids` is a non-empty allowlist, those PIDs may be
@@ -58,12 +68,49 @@ export type ReclaimListenPortOptions = WaitForPortOptions & {
   sleepMs?: (ms: number) => Promise<void>;
 };
 
+/** Split `host:port`/`[v6]:port` on a numeric port boundary; returns the host part. */
+function listenHost(token: string): string {
+  const bracketed = /^(\[[0-9a-fA-F:.]+\]):/.exec(token);
+  if (bracketed) return bracketed[1].slice(1, -1).toLowerCase();
+  // Only a trailing :<digits> is a port; a bare "::" or hostname wildcard has none.
+  const withPort = /^(.*):(\d+)$/.exec(token);
+  return (withPort ? withPort[1] : token).toLowerCase();
+}
+
+/** Normalize a listen-address host: strips brackets and the IPv4-mapped prefix. */
+export function normalizeListenAddress(token: string): string {
+  let host = listenHost(token);
+  if (host.startsWith("::ffff:")) host = host.slice(7);
+  return host;
+}
+
+/** Normalize a bare bind address (no port): drops brackets, keeps bare IPv6 whole. */
+function bareListenAddress(address: string): string {
+  let host = address.replace(/^\[|\]$/g, "").toLowerCase();
+  if (host.startsWith("::ffff:")) host = host.slice(7);
+  return host;
+}
+
+const WILDCARD_LISTEN_HOSTS = new Set(["", "*", "0.0.0.0", "::"]);
+
 /**
- * Parse `netstat -ano` (Windows) / `netstat -anlp` listen lines for a port.
- * Exported for unit tests.
+ * Whether a socket bound to `listenerAddress` also serves connections to `bound` —
+ * exact match, or a wildcard listener, or a wildcard `bound` (the caller listens on
+ * every address). IPv4-mapped IPv6 forms of the same address are equalized first.
  */
-export function parseListenPidsFromNetstat(output: string, port: number): number[] {
-  const pids = new Set<number>();
+export function listenAddressServes(listenerAddress: string, bound: string): boolean {
+  const listener = normalizeListenAddress(listenerAddress);
+  const want = bareListenAddress(bound);
+  return WILDCARD_LISTEN_HOSTS.has(listener) || WILDCARD_LISTEN_HOSTS.has(want)
+    || listener === want;
+}
+
+/**
+ * Parse `netstat -ano` (Windows) / `netstat -anlp` listen lines for a port, keeping
+ * each listener's bound local address. Exported for unit tests.
+ */
+export function parseListenEntriesFromNetstat(output: string, port: number): ListenEntry[] {
+  const entries = new Map<number, ListenEntry>();
   const portSuffix = `:${port}`;
   for (const rawLine of output.split(/\r?\n/)) {
     const line = rawLine.trim();
@@ -86,9 +133,66 @@ export function parseListenPidsFromNetstat(output: string, port: number): number
       : unixPid
         ? Number(unixPid[1])
         : NaN;
-    if (Number.isSafeInteger(pid) && pid > 0) pids.add(pid);
+    if (Number.isSafeInteger(pid) && pid > 0) {
+      entries.set(pid, { pid, address: normalizeListenAddress(parts[localIdx]) });
+    }
+  }
+  return [...entries.values()];
+}
+
+/**
+ * Parse `netstat -ano` (Windows) / `netstat -anlp` listen lines for a port.
+ * Exported for unit tests.
+ */
+export function parseListenPidsFromNetstat(output: string, port: number): number[] {
+  return parseListenEntriesFromNetstat(output, port).map(entry => entry.pid);
+}
+
+/**
+ * Parse `ss -Hltnp` rows for a port, keeping the bound local address. A row without
+ * a `pid=` attribution (another user's socket) is dropped rather than reported
+ * unverifiable. Exported for unit tests.
+ */
+export function parseListenEntriesFromSs(output: string, port: number): ListenEntry[] {
+  const entries = new Map<number, ListenEntry>();
+  const portSuffix = `:${port}`;
+  for (const rawLine of output.split(/\r?\n/)) {
+    const line = rawLine.trim();
+    if (!/^LISTEN\b/i.test(line)) continue;
+    const parts = line.split(/\s+/);
+    // LISTEN <recv-q> <send-q> <local-addr:port> <peer-addr:port> users:(...)
+    const localIdx = parts.findIndex(part => part.endsWith(portSuffix) || part.endsWith(`]:${port}`));
+    if (localIdx < 0) continue;
+    const pidMatch = /pid=(\d+)/.exec(line);
+    const pid = pidMatch ? Number(pidMatch[1]) : NaN;
+    if (Number.isSafeInteger(pid) && pid > 0) {
+      entries.set(pid, { pid, address: normalizeListenAddress(parts[localIdx]) });
+    }
   }
-  return [...pids];
+  return [...entries.values()];
+}
+
+/**
+ * Parse `lsof -nP -iTCP:<port> -sTCP:LISTEN` output (without -t). The NAME column is
+ * the last address token, optionally followed by the `(LISTEN)` state; skip the
+ * header and any line whose pid is not numeric. Exported for unit tests.
+ */
+export function parseListenEntriesFromLsof(output: string, port: number): ListenEntry[] {
+  const entries = new Map<number, ListenEntry>();
+  const portSuffix = `:${port}`;
+  for (const rawLine of output.split(/\r?\n/)) {
+    const line = rawLine.trim();
+    if (!line || /^COMMAND\b/.test(line)) continue;
+    const parts = line.split(/\s+/);
+    const pid = /^\d+$/.test(parts[1] ?? "") ? Number(parts[1]) : NaN;
+    if (!Number.isSafeInteger(pid) || pid <= 0) continue;
+    let addressIdx = parts.length - 1;
+    if (/^\(.*\)$/.test(parts[addressIdx] ?? "")) addressIdx -= 1;
+    const address = parts[addressIdx] ?? "";
+    if (!address.endsWith(portSuffix) && !address.endsWith(`]:${port}`)) continue;
+    entries.set(pid, { pid, address: normalizeListenAddress(address) });
+  }
+  return [...entries.values()];
 }
 
 function normalizeListenPidScan(result: ListenPidScan | number[]): ListenPidScan {
@@ -119,50 +223,84 @@ function readWindowsNetstatAno(): string {
 }
 
 /**
- * Scan for PIDs currently LISTENing on `port`.
- * Distinguishes probe failure (`ok: false`) from a successful empty result.
+ * Scan for the sockets currently LISTENing on `port`, with each listener's bound
+ * local address. Distinguishes probe failure (`ok: false`) from a successful empty
+ * result. POSIX backends are tried in order — `lsof`, `ss` (iproute2, the only
+ * scanner on minimal Linux installs), then `netstat` — and a missing scanner falls
+ * through to the next instead of failing the scan.
  */
-export function scanListenPids(port: number): ListenPidScan {
+export function scanListenEntries(port: number): ListenEntryScan {
   if (!Number.isFinite(port) || port <= 0 || port > 65535) {
     return { ok: false, error: "invalid port" };
   }
+  const scanned = Math.trunc(port);
   try {
     if (process.platform === "win32") {
-      return { ok: true, pids: parseListenPidsFromNetstat(readWindowsNetstatAno(), port) };
+      return { ok: true, listeners: parseListenEntriesFromNetstat(readWindowsNetstatAno(), scanned) };
     }
+    const errors: string[] = [];
     try {
-      const output = execFileSync("lsof", ["-nP", `-iTCP:${port}`, "-sTCP:LISTEN", "-t"], {
+      const output = execFileSync("lsof", ["-nP", `-iTCP:${scanned}`, "-sTCP:LISTEN"], {
         encoding: "utf-8",
         stdio: ["ignore", "pipe", "ignore"],
         timeout: 3000,
       });
-      return {
-        ok: true,
-        pids: output
-          .split(/\r?\n/)
-          .map(line => Number(line.trim()))
-          .filter(pid => Number.isSafeInteger(pid) && pid > 0),
-      };
-    } catch (lsofErr) {
-      try {
-        const output = execFileSync("netstat", ["-anlp"], {
-          encoding: "utf-8",
-          stdio: ["ignore", "pipe", "ignore"],
-          timeout: 3000,
-        });
-        return { ok: true, pids: parseListenPidsFromNetstat(output, Math.trunc(port)) };
-      } catch (netstatErr) {
-        return {
-          ok: false,
-          error: `lsof/netstat unavailable: ${String(lsofErr)} / ${String(netstatErr)}`,
-        };
-      }
+      return { ok: true, listeners: parseListenEntriesFromLsof(output, scanned) };
+    } catch (error) {
+      errors.push(`lsof: ${String(error)}`);
+    }
+    try {
+      const output = execFileSync("ss", ["-Hltnp"], {
+        encoding: "utf-8",
+        stdio: ["ignore", "pipe", "ignore"],
+        timeout: 3000,
+      });
+      return { ok: true, listeners: parseListenEntriesFromSs(output, scanned) };
+    } catch (error) {
+      errors.push(`ss: ${String(error)}`);
     }
+    try {
+      const output = execFileSync("netstat", ["-anlp"], {
+        encoding: "utf-8",
+        stdio: ["ignore", "pipe", "ignore"],
+        timeout: 3000,
+      });
+      return { ok: true, listeners: parseListenEntriesFromNetstat(output, scanned) };
+    } catch (error) {
+      errors.push(`netstat: ${String(error)}`);
+    }
+    return { ok: false, error: `no listener scanner available (${errors.join(" / ")})` };
   } catch (error) {
     return { ok: false, error: String(error) };
   }
 }
 
+/**
+ * Scan for PIDs currently LISTENing on `port`.
+ * Distinguishes probe failure (`ok: false`) from a successful empty result.
+ */
+export function scanListenPids(port: number): ListenPidScan {
+  const scan = scanListenEntries(port);
+  if (!scan.ok) return { ok: false, error: scan.error };
+  return { ok: true, pids: [...new Set(scan.listeners.map(entry => entry.pid))] };
+}
+
+/**
+ * PIDs LISTENing on `port` that actually serve `address`: listeners bound to that
+ * exact address plus wildcards (0.0.0.0/::). A listener on a different loopback or
+ * interface address (e.g. 127.0.0.2 while the tunnel binds 127.0.0.1) never receives
+ * the connection and must not block or qualify a readiness check.
+ */
+export function scanListenPidsForAddress(port: number, address: string): ListenPidScan {
+  const scan = scanListenEntries(port);
+  if (!scan.ok) return { ok: false, error: scan.error };
+  const pids = new Set<number>();
+  for (const entry of scan.listeners) {
+    if (listenAddressServes(entry.address, address)) pids.add(entry.pid);
+  }
+  return { ok: true, pids: [...pids] };
+}
+
 /** Best-effort PIDs currently LISTENing on `port`. Empty on probe failure. */
 export function listListenPids(port: number): number[] {
   const scan = scanListenPids(port);