diff --git a/src/client/link-join.ts b/src/client/link-join.ts index b782ef42259..04b15f86fa2 100644 --- a/src/client/link-join.ts +++ b/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,6 +211,7 @@ async function compensateStaleSidecar(deps: ClientLinkJoinDeps): Promise { async function waitForReady( deps: ClientLinkJoinDeps, + tunnel: ClientLinkTunnelHandle, port: number, key: string, ): Promise { @@ -210,13 +219,48 @@ async function waitForReady( const now = deps.now ?? Date.now; const sleep = deps.sleep ?? ((ms: number) => new Promise(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(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,7 +331,7 @@ 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); @@ -295,22 +339,28 @@ export async function joinHome(deps: ClientLinkJoinDeps, input: { alias: string } 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); diff --git a/src/server/port-reclaim.ts b/src/server/port-reclaim.ts index 0bd0660ff7b..b10300486d4 100644 --- a/src/server/port-reclaim.ts +++ b/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; }; +/** 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 : 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(); +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(); 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(); + 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 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: -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(); + 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(); + 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); diff --git a/structure/runtime.md b/structure/runtime.md index fd99beb93a0..f8d4ea8358d 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -77,7 +77,7 @@ verified matching processes regardless of the advisory freshness result. ## Hub management dashboard address -When hub management ingress is enabled, `src/cli/dispatch.ts` opens the dashboard on the literal IPv4 loopback address and configured ingress port, matching the listener in `src/server/index.ts`. Other dashboard address selection is unchanged. +When hub management ingress is enabled, `src/cli/dispatch.ts` opens the dashboard on the literal IPv4 loopback address and configured ingress port, matching the listener in `src/server/index.ts`. Other dashboard address selection is unchanged. Client-initiated Remote Link enrollment in `src/client/link-join.ts` watches the SSH tunnel from spawn grace through the connection commit. Readiness is accepted only while the LISTEN owner of the tunnel port is the spawned ssh process — a live tunnel does not prove it owns the socket, and a foreign listener answering the link-auth challenge would otherwise collect the issued key — and ownership is judged only among listeners that serve the tunnel's 127.0.0.1 bind, so an occupant on another loopback or interface address can neither satisfy the check nor block it. The socket scan (`src/server/port-reclaim.ts`) runs `lsof`, then `ss`, then `netstat`, so minimal Linux installs with only iproute2 still enumerate listeners. The ownership check repeats immediately before the keyed request, narrowing the takeover window between the 401 probe and the request that carries the issued key, and both requests run with `redirect: "manual"` so a redirecting occupant cannot reroute the challenge. An exited tunnel can neither deliver the issued data key to an unrelated loopback listener nor commit the connection. ## Codex desktop process membership diff --git a/tests/server/link-join-route.test.ts b/tests/server/link-join-route.test.ts index ca57e13df0c..f6c9859d667 100644 --- a/tests/server/link-join-route.test.ts +++ b/tests/server/link-join-route.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test, spyOn } from "bun:test"; import { ClientLinkJoinError, joinHome, type ClientLinkJoinDeps } from "../../src/client/link-join"; +import { spawnClientLinkTunnel } from "../../src/client/link-tunnel"; import { handleLinkRoutes, type LinkRouteState } from "../../src/server/management/link-routes"; import type { ManagementContext } from "../../src/server/management/context"; import type { SshRunner } from "../../src/link/ssh-runner"; @@ -32,13 +33,27 @@ function runnerFor(calls: string[][], issueResult = true): SshRunner { function tunnelFor(order: string[]) { return { pid: 123, - exited: Promise.resolve(0), + exited: new Promise(() => {}), stop: async () => { order.push("stop-tunnel"); }, }; } +// The link listener answers an unauthenticated /readyz with the 401 challenge; the keyed +// request earns 200. A foreign listener (any other status on the probe) must never see the key. +function challengedFetch(order?: string[]) { + return async (_input: RequestInfo | URL, init?: RequestInit) => { + const authed = new Headers(init?.headers).get("x-opencodex-api-key") === KEY; + order?.push(authed ? "readyz:key" : "readyz:probe"); + return new Response(null, { status: authed ? 200 : 401 }); + }; +} + function joinDeps(overrides: Partial = {}): ClientLinkJoinDeps { const calls = overrides.runner ? [] : []; + // The readiness gate only trusts the port when the LISTEN pid is the spawned tunnel's; + // wrap whichever spawnTunnel is under test so the default scan reports that pid. + let tunnelPid = 0; + const spawn = overrides.spawnTunnel ?? spawnClientLinkTunnel; return { runner: overrides.runner ?? runnerFor(calls), knownHostsFile: "/tmp/ocx-known-hosts", @@ -50,6 +65,12 @@ function joinDeps(overrides: Partial = {}): ClientLinkJoinDe readSidecar: () => null, readConnectionState: () => ({ kind: "disconnected" }), ...overrides, + spawnTunnel: (spec, spawnDeps) => { + const handle = spawn(spec, spawnDeps); + tunnelPid = handle.pid; + return handle; + }, + scanListenPids: overrides.scanListenPids ?? (() => ({ ok: true, pids: [tunnelPid] })), }; } @@ -158,10 +179,8 @@ describe("client initiated link join", () => { hostname: () => "client-host", writeState: state => { order.push("write-state"); Object.assign(sidecar, state); }, spawnTunnel: () => { order.push("spawn-tunnel"); return tunnelFor(order); }, - fetchImpl: async (_input, init) => { - order.push(`readyz:${new Headers(init?.headers).get("x-opencodex-api-key") === KEY ? "key" : "missing"}`); - return new Response(null, { status: 200 }); - }, + scanListenPids: () => ({ ok: true, pids: [123] }), + fetchImpl: challengedFetch(order), connect: (async () => { order.push("connect"); }) as typeof import("../../src/client/connect").connectClient, scheduleRestart: () => { order.push("restart"); }, }, { alias: "home" })), @@ -171,7 +190,7 @@ describe("client initiated link join", () => { expect(response?.status).toBe(202); expect(responseBody).toEqual({ linkId: LINK_ID, alias: "home", restarting: true }); expect(sidecar).toMatchObject({ linkId: LINK_ID, tunnelPort: 23456, peerListenerPort: 45678 }); - expect(order).toEqual(["write-state", "spawn-tunnel", "readyz:key", "connect", "stop-tunnel", "restart"]); + expect(order).toEqual(["write-state", "spawn-tunnel", "readyz:probe", "readyz:key", "connect", "stop-tunnel", "restart"]); expect(calls[0]?.some(value => value.includes("issue"))).toBe(true); expect(calls[0]?.some(value => value.includes("--json"))).toBe(true); }); @@ -201,7 +220,7 @@ describe("client initiated link join", () => { sleep: async () => {}, writeState: () => {}, clearState: () => { cleared += 1; }, - spawnTunnel: () => ({ pid: 1, exited: Promise.resolve(0), stop: async () => { stopped += 1; } }), + spawnTunnel: () => ({ pid: 1, exited: new Promise(() => {}), stop: async () => { stopped += 1; } }), fetchImpl: async () => readiness === "unauthorized" ? new Response(null, { status: 401 }) : new Response(null, { status: 503 }), }); await expect(joinHome(deps, { alias: "home" })).rejects.toMatchObject({ @@ -213,6 +232,150 @@ describe("client initiated link join", () => { } }); + test("never sends the issued key to a listener that skips the link-auth challenge", async () => { + const calls: string[][] = []; + let keyedFetches = 0; + let ticks = 0; + let stopped = 0; + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + now: () => (ticks++ === 0 ? 0 : 15_002 * ticks), + writeState: () => {}, + clearState: () => {}, + spawnTunnel: () => ({ pid: 1, exited: new Promise(() => {}), stop: async () => { stopped += 1; } }), + fetchImpl: async (_input, init) => { + if (new Headers(init?.headers).has("x-opencodex-api-key")) keyedFetches += 1; + return new Response(null, { status: 200 }); + }, + }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(keyedFetches).toBe(0); + expect(stopped).toBe(1); + expect(calls.filter(argv => argv.some(value => value.includes("revoke")))).toHaveLength(1); + }); + + test("a squatter answering the 401 challenge never receives the issued key", async () => { + const calls: string[][] = []; + let keyedFetches = 0; + let ticks = 0; + let stopped = 0; + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + now: () => (ticks++ === 0 ? 0 : 15_002 * ticks), + writeState: () => {}, + clearState: () => {}, + spawnTunnel: () => ({ pid: 123, exited: new Promise(() => {}), stop: async () => { stopped += 1; } }), + // A foreign process holds the port: a live ssh does not prove it owns the socket. + scanListenPids: () => ({ ok: true, pids: [999] }), + fetchImpl: async (_input, init) => { + if (new Headers(init?.headers).has("x-opencodex-api-key")) keyedFetches += 1; + return new Response(null, { status: 401 }); + }, + }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(keyedFetches).toBe(0); + expect(stopped).toBe(1); + expect(calls.filter(argv => argv.some(value => value.includes("revoke")))).toHaveLength(1); + }); + + test("the readiness scan is scoped to the tunnel's loopback address", async () => { + const seenAddresses: Array = []; + const order: string[] = []; + await joinHome(joinDeps({ + runner: runnerFor([]), + writeState: () => {}, + clearState: () => {}, + spawnTunnel: () => tunnelFor(order), + // Only sockets serving 127.0.0.1 count: the real scanner reports listeners on + // other loopback/interface addresses too, and the dep must scope them out. + scanListenPids: (_port, address) => { + seenAddresses.push(address); + return { ok: true, pids: [123] }; + }, + fetchImpl: challengedFetch(order), + connect: (async () => {}) as never, + scheduleRestart: () => {}, + }), { alias: "home" }); + expect(seenAddresses.length).toBeGreaterThan(0); + for (const address of seenAddresses) expect(address).toBe("127.0.0.1"); + }); + + test("a port flip between the probe and the keyed request never receives the key", async () => { + const calls: string[][] = []; + let keyedFetches = 0; + let ticks = 0; + let scans = 0; + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + now: () => (ticks++ === 0 ? 0 : 15_002 * ticks), + writeState: () => {}, + clearState: () => {}, + spawnTunnel: () => ({ pid: 123, exited: new Promise(() => {}), stop: async () => {} }), + // First scan names the tunnel; by the time the 401 arrives a squatter holds the port. + scanListenPids: () => ({ ok: true, pids: scans++ === 0 ? [123] : [999] }), + fetchImpl: async (_input, init) => { + if (new Headers(init?.headers).has("x-opencodex-api-key")) keyedFetches += 1; + return new Response(null, { status: 401 }); + }, + }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(keyedFetches).toBe(0); + expect(calls.filter(argv => argv.some(value => value.includes("revoke")))).toHaveLength(1); + }); + + test("a redirect on the readiness probe is never followed with the issued key", async () => { + const calls: string[][] = []; + let keyedFetches = 0; + let ticks = 0; + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + now: () => (ticks++ === 0 ? 0 : 15_002 * ticks), + writeState: () => {}, + clearState: () => {}, + spawnTunnel: () => ({ pid: 123, exited: new Promise(() => {}), stop: async () => {} }), + fetchImpl: async (_input, init) => { + if (new Headers(init?.headers).has("x-opencodex-api-key")) keyedFetches += 1; + // A port occupant redirecting the probe used to let the challenge pass at a foreign URL. + return new Response(null, { status: 302, headers: { location: "http://169.254.1.1/fake-readyz" } }); + }, + }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(keyedFetches).toBe(0); + expect(calls.filter(argv => argv.some(value => value.includes("revoke")))).toHaveLength(1); + }); + + test("a tunnel that exits during connect cannot commit the connection", async () => { + const calls: string[][] = []; + let releaseExit!: (code: number) => void; + const exited = new Promise(resolve => { releaseExit = resolve; }); + let stopped = 0; + let connectCommitted = false; + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + writeState: () => {}, + clearState: () => {}, + spawnTunnel: () => ({ pid: 1, exited, stop: async () => { stopped += 1; } }), + fetchImpl: challengedFetch(), + connect: (async () => { releaseExit(255); await new Promise(() => {}); connectCommitted = true; }) as typeof import("../../src/client/connect").connectClient, + }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(connectCommitted).toBe(false); + expect(stopped).toBe(1); + expect(calls.filter(argv => argv.some(value => value.includes("revoke")))).toHaveLength(1); + }); + + test("does not disclose the issued key when the tunnel exits during its spawn grace", async () => { + const calls: string[][] = []; + let fetches = 0; + await expect(joinHome(joinDeps({ + runner: runnerFor(calls), + writeState: () => {}, + clearState: () => {}, + spawnTunnel: () => ({ pid: 1, exited: Promise.resolve(255), stop: async () => {} }), + fetchImpl: async () => { + fetches += 1; + return new Response(null, { status: 200 }); + }, + }), { alias: "home" })).rejects.toMatchObject({ code: "join_tunnel_failed" }); + expect(fetches).toBe(0); + expect(calls.filter(argv => argv.some(value => value.includes("revoke")))).toHaveLength(1); + }); + test("rolls back on connect failure and never exposes the issued key", async () => { const calls: string[][] = []; const logs = spyOn(console, "log").mockImplementation(() => {}); @@ -222,7 +385,7 @@ describe("client initiated link join", () => { writeState: () => {}, clearState: () => {}, spawnTunnel: () => tunnelFor([]), - fetchImpl: async () => new Response(null, { status: 200 }), + fetchImpl: challengedFetch(), connect: (async () => { throw new Error(`connect failed ${KEY}`); }) as typeof import("../../src/client/connect").connectClient, }), { alias: "home" })).rejects.toMatchObject({ code: "join_connect_failed" }); } finally { @@ -263,8 +426,8 @@ describe("client initiated link join", () => { readSidecar: () => sidecarPresent ? sidecar : null, writeState: value => { sidecarPresent = true; Object.assign(sidecar, value); }, clearState: () => { sidecarPresent = false; }, - spawnTunnel: () => ({ pid: 1, exited: Promise.resolve(0), stop: async () => {} }), - fetchImpl: async () => new Response(null, { status: 200 }), + spawnTunnel: () => ({ pid: 1, exited: new Promise(() => {}), stop: async () => {} }), + fetchImpl: challengedFetch(), connect: (async () => { throw new Error("connect failed"); }) as typeof import("../../src/client/connect").connectClient, }); await expect(joinHome(base, { alias: "home" })).rejects.toMatchObject({ code: "join_rollback_failed", linkId: LINK_ID }); @@ -297,7 +460,8 @@ describe("client initiated link join", () => { writeState: state => { sidecar = { ...state }; }, clearState: () => { cleared = true; }, spawnTunnel: () => tunnelFor([]), - fetchImpl: async () => new Response(null, { status: 200 }), + scanListenPids: () => ({ ok: true, pids: [123] }), + fetchImpl: challengedFetch(), connect: (async () => { connected = true; }) as typeof import("../../src/client/connect").connectClient, scheduleRestart: () => { throw new Error("restart unavailable"); }, }, input)) as typeof import("../../src/client/link-join").joinHome, diff --git a/tests/server/port-reclaim.test.ts b/tests/server/port-reclaim.test.ts index 930a7866c9c..5c6dfbd041b 100644 --- a/tests/server/port-reclaim.test.ts +++ b/tests/server/port-reclaim.test.ts @@ -1,5 +1,15 @@ import { describe, expect, spyOn, test } from "bun:test"; -import { reclaimListenPort, type ReclaimListenPortOptions } from "../../src/server/port-reclaim"; +import { createServer } from "node:net"; +import { + listenAddressServes, + normalizeListenAddress, + parseListenEntriesFromLsof, + parseListenEntriesFromNetstat, + parseListenEntriesFromSs, + reclaimListenPort, + scanListenPidsForAddress, + type ReclaimListenPortOptions, +} from "../../src/server/port-reclaim"; import { isBareIpv6Address, parseTcpQuadsForLocalPort, @@ -41,6 +51,99 @@ describe("parseListenPidsFromNetstat", () => { }); }); +describe("listen-entry parsers keep the bound address", () => { + test("netstat entries report each listener's local address", () => { + const output = [ + "tcp 0 0 127.0.0.1:10100 0.0.0.0:* LISTEN 4242/bun", + "tcp 0 0 127.0.0.2:10100 0.0.0.0:* LISTEN 7777/foreign", + "tcp 0 0 127.0.0.1:22 0.0.0.0:* LISTEN 1/sshd", + ].join("\n"); + expect(parseListenEntriesFromNetstat(output, 10100)).toEqual([ + { pid: 4242, address: "127.0.0.1" }, + { pid: 7777, address: "127.0.0.2" }, + ]); + }); + + test("ss -Hltnp rows report address and pid; unattributed rows are dropped", () => { + const output = [ + "LISTEN 0 128 127.0.0.1:10100 0.0.0.0:* users:((\"bun\",pid=4242,fd=20))", + "LISTEN 0 128 127.0.0.2:10100 0.0.0.0:* users:((\"foreign\",pid=7777,fd=6))", + "LISTEN 0 128 127.0.0.1:10100 0.0.0.0:*", + "LISTEN 0 511 *:22 *:* users:((\"sshd\",pid=1,fd=3))", + ].join("\n"); + expect(parseListenEntriesFromSs(output, 10100)).toEqual([ + { pid: 4242, address: "127.0.0.1" }, + { pid: 7777, address: "127.0.0.2" }, + ]); + }); + + test("lsof NAME column supplies the bound address", () => { + const output = [ + "COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME", + "bun 4242 devin 20u IPv4 0xdeadbeef 0t0 TCP 127.0.0.1:10100 (LISTEN)", + "other 7777 devin 21u IPv4 0xdeadbeef 0t0 TCP 127.0.0.2:10100 (LISTEN)", + ].join("\n"); + expect(parseListenEntriesFromLsof(output, 10100)).toEqual([ + { pid: 4242, address: "127.0.0.1" }, + { pid: 7777, address: "127.0.0.2" }, + ]); + }); + + test("address matching treats wildcards as serving any bound address", () => { + expect(listenAddressServes("127.0.0.1", "127.0.0.1")).toBe(true); + expect(listenAddressServes("127.0.0.2", "127.0.0.1")).toBe(false); + expect(listenAddressServes("0.0.0.0", "127.0.0.1")).toBe(true); + expect(listenAddressServes("*", "127.0.0.1")).toBe(true); + expect(listenAddressServes("::", "127.0.0.1")).toBe(true); + expect(listenAddressServes("[::1]:443", "::1")).toBe(true); + expect(normalizeListenAddress("::ffff:127.0.0.1")).toBe("127.0.0.1"); + expect(listenAddressServes("::ffff:127.0.0.1", "127.0.0.1")).toBe(true); + }); +}); + +describe("scanListenPidsForAddress (real scanner)", () => { + test("finds this process on its own bound port and filters other addresses", async () => { + const server = createServer(); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => resolve()); + }); + try { + const address = server.address(); + if (typeof address === "object" && address) { + const scan = scanListenPidsForAddress(address.port, "127.0.0.1"); + // The default scanner is whatever the platform ships (netstat/lsof/ss); when none + // is installed the scan reports a probe failure rather than an empty list. + if (scan.ok) { + expect(scan.pids).toContain(process.pid); + } + } + } finally { + server.close(); + } + }); + + test("a listener on another loopback address does not serve 127.0.0.1", async () => { + const server = createServer(); + const bound = await new Promise(resolve => { + server.once("error", () => resolve(false)); + server.listen(0, "127.0.0.2", () => resolve(true)); + }); + if (!bound) return; // platform does not allow the second loopback address + try { + const address = server.address(); + if (typeof address === "object" && address) { + const scan = scanListenPidsForAddress(address.port, "127.0.0.1"); + if (scan.ok) expect(scan.pids).not.toContain(process.pid); + const wide = scanListenPidsForAddress(address.port, "0.0.0.0"); + if (wide.ok) expect(wide.pids).toContain(process.pid); + } + } finally { + server.close(); + } + }); +}); + describe("parseTcpQuadsForLocalPort / IPv6", () => { test("collects every TCP row on the local port including non-LISTEN states", () => { const output = [