From 053e8fee45c3e33ad2a9f6205a8391b1a25bacce Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Tue, 14 Jul 2026 21:23:54 -0400 Subject: [PATCH 1/4] fix(sync): validate loopback ADE handshake before publishing a port Host-side fix for the 2026-07-14 loopback port-collision incident: a foreign 127.0.0.1: listener shadowed ADE's 0.0.0.0 bind with no EADDRINUSE, routing Tailscale + relay to the wrong process. A 426 loopback probe now validates each port before it is accepted/published; shadowed ports drift to a validated-free port with atomic route republish; relay validates before forwarding; ade sync status + doctor report per-route health. 343 tests + a real macOS collision repro green. Co-Authored-By: Claude Opus 4.8 --- apps/ade-cli/src/bootstrap.ts | 1 + apps/ade-cli/src/cli.test.ts | 75 ++++++ apps/ade-cli/src/cli.ts | 75 ++++++ .../src/services/sync/sharedSyncListener.ts | 140 +++++++--- .../src/services/sync/syncHostService.test.ts | 30 +++ .../src/services/sync/syncHostService.ts | 87 +++++- .../sync/syncLoopbackCollision.test.ts | 253 ++++++++++++++++++ .../src/services/sync/syncLoopbackProbe.ts | 105 ++++++++ apps/ade-cli/src/services/sync/syncService.ts | 137 +++++++++- .../sync/syncTunnelClientService.test.ts | 57 +++- .../services/sync/syncTunnelClientService.ts | 57 +++- apps/desktop/src/shared/types/sync.ts | 34 +++ apps/ios/ADE/Services/SyncService.swift | 131 +++++---- apps/ios/ADETests/ADETests.swift | 73 +++++ 14 files changed, 1145 insertions(+), 110 deletions(-) create mode 100644 apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts create mode 100644 apps/ade-cli/src/services/sync/syncLoopbackProbe.ts diff --git a/apps/ade-cli/src/bootstrap.ts b/apps/ade-cli/src/bootstrap.ts index 8053af5e4..758568087 100644 --- a/apps/ade-cli/src/bootstrap.ts +++ b/apps/ade-cli/src/bootstrap.ts @@ -1588,6 +1588,7 @@ export async function createAdeRuntime(args: { remoteCommandExecutor: resolvedArgs.syncRuntime.remoteCommandExecutor, getModelPickerStore: () => getSharedModelPickerStore(db), cloudRelayStore, + syncTunnelClientService, onCloudRelayEnabledChanged: (enabled) => { // Same gate as startup: only the sync-hosting runtime may register // the relay tunnel (see canHostRelayTunnel above). diff --git a/apps/ade-cli/src/cli.test.ts b/apps/ade-cli/src/cli.test.ts index 4bbeb495e..367dd852f 100644 --- a/apps/ade-cli/src/cli.test.ts +++ b/apps/ade-cli/src/cli.test.ts @@ -3315,6 +3315,81 @@ describe("ADE CLI", () => { expect(output).toContain("Git repository detected"); }); + it("adds sync route health to doctor and names a loopback listener mismatch", () => { + const projectRoot = fs.mkdtempSync(path.join(os.tmpdir(), "ade-cli-doctor-sync-")); + fs.mkdirSync(path.join(projectRoot, ".ade"), { recursive: true }); + try { + const plan = expectExecutePlan(buildCliPlan(["doctor"])); + expect(plan.steps).toContainEqual({ + key: "syncStatus", + method: "sync.getStatus", + params: { includeTransferReadiness: false }, + optional: true, + }); + const summary = summarizeExecution({ + plan, + connection: { + mode: "runtime-socket", + projectRoot, + workspaceRoot: projectRoot, + socketPath: path.join(projectRoot, ".ade", "ade.sock"), + }, + values: { + rpcActions: { actions: [{}] }, + actions: { actions: [{}] }, + syncStatus: { + pairingConnectInfo: { port: 8787 }, + routeHealth: { + listener: { + listenerBound: true, + loopbackAdeValidated: false, + reason: "Expected ADE 426 Upgrade Required; received 404 Not Found.", + }, + tailscale: { + enabled: true, + tailscaleReachable: false, + reason: "Tailscale route points at the listener mismatch.", + }, + relay: { + enabled: true, + relayControlConnected: true, + relayBridgeValidated: false, + reason: "Relay bridge refused the listener mismatch.", + }, + }, + }, + }, + } as any) as Record; + + expect(summary.sync).toMatchObject({ + enabled: true, + usable: false, + status: "warning", + }); + expect(summary.sync.failingRoutes).toEqual([ + expect.stringContaining("listener"), + expect.stringContaining("tailscale"), + expect.stringContaining("relay"), + ]); + expect(summary.sync.message).toContain("404 Not Found"); + const output = formatOutput(summary, { + projectRoot, + workspaceRoot: projectRoot, + role: "agent", + headless: false, + requireSocket: false, + socketPath: null, + pretty: true, + text: true, + timeoutMs: 1000, + }, "doctor"); + expect(output).toContain("Sync route failure"); + expect(output).toContain("listener"); + } finally { + fs.rmSync(projectRoot, { recursive: true, force: true }); + } + }); + it("detects project-local Linear credentials in doctor readiness", () => { const previousAdeLinearApi = process.env.ADE_LINEAR_API; const previousLinearApiKey = process.env.LINEAR_API_KEY; diff --git a/apps/ade-cli/src/cli.ts b/apps/ade-cli/src/cli.ts index 6239223db..1fe9b2e31 100644 --- a/apps/ade-cli/src/cli.ts +++ b/apps/ade-cli/src/cli.ts @@ -11510,6 +11510,12 @@ function buildCliPlan( ...actionStep("projectConfig", "project_config", "get"), optional: true, }, + { + key: "syncStatus", + method: "sync.getStatus", + params: { includeTransferReadiness: false }, + optional: true, + }, ], }; } @@ -12126,6 +12132,70 @@ function checkStorageReadiness(projectRoot: string): ReadinessCheck { } } +function checkSyncReadiness(value: unknown): ReadinessCheck & { + enabled: boolean; + usable: boolean; + failingRoutes: string[]; +} { + const snapshot = isRecord(value) ? value : null; + const routeHealth = snapshot && isRecord(snapshot.routeHealth) ? snapshot.routeHealth : null; + const listener = routeHealth && isRecord(routeHealth.listener) ? routeHealth.listener : null; + const tailscale = routeHealth && isRecord(routeHealth.tailscale) ? routeHealth.tailscale : null; + const relay = routeHealth && isRecord(routeHealth.relay) ? routeHealth.relay : null; + const enabled = Boolean(snapshot?.pairingConnectInfo) || relay?.enabled === true; + if (!snapshot || !routeHealth) { + return { + ready: false, + enabled: false, + usable: false, + status: "unavailable", + message: "Sync route health is unavailable.", + nextAction: "Run 'ade sync status --text' against the live ADE runtime.", + failingRoutes: [], + }; + } + if (!enabled) { + return { + ready: true, + enabled: false, + usable: false, + status: "unavailable", + message: "Phone sync hosting is not enabled in this runtime.", + failingRoutes: [], + details: { routeHealth }, + }; + } + + const failures: string[] = []; + if (listener?.listenerBound !== true || listener?.loopbackAdeValidated !== true) { + failures.push(`listener: ${asString(listener?.reason) ?? "loopback listener mismatch"}`); + } + if (tailscale?.enabled === true && tailscale?.tailscaleReachable !== true) { + failures.push(`tailscale: ${asString(tailscale.reason) ?? "published route is not reachable"}`); + } + if ( + relay?.enabled === true + && (relay?.relayControlConnected !== true || asString(relay?.reason) != null) + ) { + failures.push(`relay: ${asString(relay.reason) ?? "control channel is not connected"}`); + } + const usable = failures.length === 0; + return { + ready: usable, + enabled: true, + usable, + status: usable ? "ready" : "warning", + message: usable + ? "Enabled sync routes are usable." + : `Sync route failure: ${failures.join("; ")}`, + nextAction: usable + ? undefined + : "Run 'ade sync status --text' and resolve the named listener or route failure.", + failingRoutes: failures, + details: { routeHealth }, + }; +} + function requireAdeLayout(): { resolveAdeLayout: (projectRoot: string) => { secretsDir: string }; } { @@ -12183,6 +12253,7 @@ function buildReadinessSnapshot(args: { computerUse: checkComputerUseReadiness(), path: checkPathReadiness(), storage: checkStorageReadiness(connection.projectRoot), + sync: checkSyncReadiness(values.syncStatus), }; const recommendations = Object.entries(checks) .filter(([, check]) => check.nextAction) @@ -12258,6 +12329,7 @@ function buildReadinessSnapshot(args: { computerUse: checks.computerUse, path: checks.path, storage: checks.storage, + sync: checks.sync, auth: { localProjectAccess: projectInitialized && actions.length > 0, providerSecretsExposed: false, @@ -17064,6 +17136,8 @@ function formatTextOutput( isRecord(value) && isRecord(value.path) ? value.path : {}; const storage = isRecord(value) && isRecord(value.storage) ? value.storage : {}; + const sync = + isRecord(value) && isRecord(value.sync) ? value.sync : {}; const recommendations = isRecord(value) && Array.isArray(value.recommendations) ? value.recommendations @@ -17087,6 +17161,7 @@ function formatTextOutput( ["computer use", computerUse.message], ["path", pathStatus.message], ["storage", storage.message], + ["sync", sync.message], ["recommendation", isRecord(value) ? value.recommendation : null], ]), ...(recommendations.length diff --git a/apps/ade-cli/src/services/sync/sharedSyncListener.ts b/apps/ade-cli/src/services/sync/sharedSyncListener.ts index f46b751b9..f5e7b6b43 100644 --- a/apps/ade-cli/src/services/sync/sharedSyncListener.ts +++ b/apps/ade-cli/src/services/sync/sharedSyncListener.ts @@ -1,6 +1,13 @@ import { WebSocketServer, WebSocket, type RawData } from "ws"; import type { SyncPeerMetadata } from "../../../../desktop/src/shared/types"; import { DEFAULT_SYNC_HOST_PORT } from "./syncProtocol"; +import { + assertAdeLoopbackListener, + isLoopbackShadowedError, + probeAdeLoopbackListener, + type SyncLoopbackProbeResult, + type SyncLoopbackValidationStatus, +} from "./syncLoopbackProbe"; // Bind the sync host on all interfaces by default so phones on the same // wifi/LAN can reach it without Tailscale. 0.0.0.0 is a superset of loopback, @@ -103,6 +110,7 @@ export type SharedSyncListener = { ensureListening(portCandidates: number[]): Promise; getPort(): number | null; isListening(): boolean; + getLoopbackValidationStatus(): SyncLoopbackValidationStatus; /** * Install the connection handler for NEW sockets. Returns a detach function * that only clears the handler if it has not been superseded by a newer @@ -133,10 +141,28 @@ type ParkedEntry = { }; function isRetryableListenerBindError(error: unknown): boolean { + if (isLoopbackShadowedError(error)) return true; const code = (error as NodeJS.ErrnoException | null | undefined)?.code ?? ""; return code === "EADDRINUSE" || code === "EACCES"; } +async function closeCandidateServer(candidateServer: WebSocketServer): Promise { + for (const client of candidateServer.clients) { + try { + client.terminate(); + } catch { + // ignore cleanup failures on a rejected candidate + } + } + await new Promise((resolve) => { + try { + candidateServer.close(() => resolve()); + } catch { + resolve(); + } + }); +} + function rawDataBytes(data: RawData): number { if (typeof data === "string") return Buffer.byteLength(data, "utf8"); if (Buffer.isBuffer(data)) return data.length; @@ -152,11 +178,13 @@ export function createSharedSyncListener(options: { bindHost?: string; maxPayloadBytes?: number; parkedPeerGraceMs?: number; + loopbackProbe?: (port: number) => Promise; } = {}): SharedSyncListener { const logger = options.logger ?? {}; const bindHost = options.bindHost ?? SYNC_HOST_BIND_HOST; const maxPayloadBytes = options.maxPayloadBytes ?? SYNC_HOST_MAX_PAYLOAD_BYTES; const parkedPeerGraceMs = Math.max(50, Math.floor(options.parkedPeerGraceMs ?? DEFAULT_PARKED_PEER_GRACE_MS)); + const loopbackProbe = options.loopbackProbe ?? probeAdeLoopbackListener; let server: WebSocketServer | null = null; let listeningPromise: Promise | null = null; @@ -164,6 +192,13 @@ export function createSharedSyncListener(options: { let fallbackHandler: SharedSyncListenerConnectionHandler | null = null; let fallbackSuppressedUntilMs = 0; let closed = false; + let loopbackValidationStatus: SyncLoopbackValidationStatus = { + port: null, + loopbackAdeValidated: false, + lastFailureAt: null, + reason: "The shared sync listener has not been validated yet.", + lastSuccessAt: null, + }; const parked = new Map(); const unpark = (entry: ParkedEntry): void => { @@ -276,8 +311,10 @@ export function createSharedSyncListener(options: { ); let lastError: unknown = null; let previousAttemptedPort: number | null = null; + const shadowedPorts = new Set(); for (const attemptedPort of attemptPlan) { if (closed) throw new Error("The shared sync listener has been closed."); + if (shadowedPorts.has(attemptedPort)) continue; if (previousAttemptedPort === attemptedPort) { await new Promise((resolve) => setTimeout(resolve, PREFERRED_PORT_BIND_RETRY_DELAY_MS)); } @@ -287,22 +324,72 @@ export function createSharedSyncListener(options: { port: attemptedPort, maxPayload: maxPayloadBytes, }); + // Install the handler before the validation RTT so a LAN peer that + // arrives in that narrow window is parked/owned instead of orphaned. + candidateServer.on("connection", (ws, request) => { + const connection: SharedSyncListenerConnection = { + ws, + remoteAddress: request.socket.remoteAddress ?? null, + remotePort: request.socket.remotePort ?? null, + }; + const fallbackSuppressed = fallbackSuppressedUntilMs > Date.now(); + const activeHandler = handler ?? (fallbackSuppressed ? null : fallbackHandler); + if (activeHandler) { + activeHandler(connection); + return; + } + park({ + ws, + remoteAddress: connection.remoteAddress, + remotePort: connection.remotePort, + metadata: null, + authKind: null, + pairedDeviceId: null, + connectedAt: new Date().toISOString(), + }); + }); try { const resolvedPort = await new Promise((resolve, reject) => { - const onListening = () => { - cleanup(); + const onListening = async () => { const address = candidateServer.address(); - resolve(typeof address === "object" && address ? address.port : attemptedPort); + const port = typeof address === "object" && address ? address.port : attemptedPort; + try { + const result = await assertAdeLoopbackListener(port, loopbackProbe); + cleanup(); + loopbackValidationStatus = { + port, + loopbackAdeValidated: true, + lastFailureAt: loopbackValidationStatus.lastFailureAt, + reason: null, + lastSuccessAt: result.checkedAt, + }; + resolve(port); + } catch (error) { + cleanup(); + if (isLoopbackShadowedError(error)) { + loopbackValidationStatus = { + port, + loopbackAdeValidated: false, + lastFailureAt: error.failedAt, + reason: error.message, + lastSuccessAt: loopbackValidationStatus.lastSuccessAt, + }; + } + reject(error instanceof Error ? error : new Error(String(error))); + } }; const onError = (error: unknown) => { cleanup(); reject(error instanceof Error ? error : new Error(String(error))); }; + const handleListening = () => { + void onListening(); + }; const cleanup = () => { - candidateServer.off("listening", onListening); + candidateServer.off("listening", handleListening); candidateServer.off("error", onError); }; - candidateServer.on("listening", onListening); + candidateServer.on("listening", handleListening); candidateServer.on("error", onError); }); server = candidateServer; @@ -313,42 +400,17 @@ export function createSharedSyncListener(options: { port: resolvedPort, }); }); - server.on("connection", (ws, request) => { - const connection: SharedSyncListenerConnection = { - ws, - remoteAddress: request.socket.remoteAddress ?? null, - remotePort: request.socket.remotePort ?? null, - }; - const fallbackSuppressed = fallbackSuppressedUntilMs > Date.now(); - const activeHandler = handler ?? (fallbackSuppressed ? null : fallbackHandler); - if (activeHandler) { - activeHandler(connection); - return; - } - // No host service owns the listener right now (mid project switch - // or before the first host starts). Park the socket and buffer its - // frames; the next host adopts it via takePeers(). - park({ - ws, - remoteAddress: connection.remoteAddress, - remotePort: connection.remotePort, - metadata: null, - authKind: null, - pairedDeviceId: null, - connectedAt: new Date().toISOString(), - }); - }); return resolvedPort; } catch (error) { lastError = error; - try { - candidateServer.close(); - } catch { - // ignore cleanup failures - } - const retryable = isRetryableListenerBindError(error) && attemptedPort !== 0; + await closeCandidateServer(candidateServer); + if (isLoopbackShadowedError(error)) shadowedPorts.add(attemptedPort); + const retryable = isRetryableListenerBindError(error) + && (attemptedPort !== 0 || isLoopbackShadowedError(error)); logger.warn?.( - retryable ? "sync_listener.bind_port_conflict" : "sync_listener.bind_failed", + isLoopbackShadowedError(error) + ? "sync_listener.loopback_shadowed" + : retryable ? "sync_listener.bind_port_conflict" : "sync_listener.bind_failed", { attemptedPort, error: error instanceof Error ? error.message : String(error), @@ -385,6 +447,10 @@ export function createSharedSyncListener(options: { return server?.address() != null; }, + getLoopbackValidationStatus(): SyncLoopbackValidationStatus { + return { ...loopbackValidationStatus }; + }, + setConnectionHandler(nextHandler: SharedSyncListenerConnectionHandler): () => void { handler = nextHandler; fallbackSuppressedUntilMs = 0; diff --git a/apps/ade-cli/src/services/sync/syncHostService.test.ts b/apps/ade-cli/src/services/sync/syncHostService.test.ts index 64820daf2..8e1f50f5e 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.test.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.test.ts @@ -1768,6 +1768,36 @@ describe("createSyncHostService LAN discovery", () => { vi.restoreAllMocks(); }); + it("rejects a self-owned listener before discovery when loopback is not ADE", async () => { + const { projectRoot, cleanup } = createTempProjectRoot(); + const host = createSyncHostService({ + ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), + port: 0, + loopbackProbe: async (port: number) => ({ + ok: false, + port, + statusCode: 404, + statusMessage: "Not Found", + checkedAt: new Date().toISOString(), + reason: "foreign loopback listener", + }), + } as unknown as Parameters[0]); + + try { + await expect(host.waitUntilListening()).rejects.toThrow("foreign loopback listener"); + expect(host.getLoopbackValidationStatus()).toMatchObject({ + loopbackAdeValidated: false, + reason: "foreign loopback listener", + }); + expect(host.getTailnetDiscoveryStatus().updatedAt).toBeNull(); + expect(publishMock).not.toHaveBeenCalled(); + expect(spawnMock).not.toHaveBeenCalled(); + } finally { + await host.dispose(); + cleanup(); + } + }); + it("closes inbound sockets that never authenticate", async () => { const { projectRoot, cleanup } = createTempProjectRoot(); const host = createSyncHostService({ diff --git a/apps/ade-cli/src/services/sync/syncHostService.ts b/apps/ade-cli/src/services/sync/syncHostService.ts index 15b58640b..eaaa052b4 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.ts @@ -144,6 +144,13 @@ import { type SharedSyncListener, type SyncPeerHandoffSnapshot, } from "./sharedSyncListener"; +import { + assertAdeLoopbackListener, + isLoopbackShadowedError, + probeAdeLoopbackListener, + type SyncLoopbackProbeResult, + type SyncLoopbackValidationStatus, +} from "./syncLoopbackProbe"; export { selectChangesetBatchChunk } from "./changesetPump"; const execFileAsync = promisify(execFile); // db_version window per pump poll. Large enough to cross sparse version @@ -710,6 +717,8 @@ type SyncHostServiceArgs = { * re-scanning a QR. */ getCloudRelayWssUrl?: () => string | null; + /** Test seam; production always uses the HTTP 426 loopback probe. */ + loopbackProbe?: (port: number) => Promise; }; function sanitizeRemoteAddress(remoteAddress: string | null | undefined): string | null { @@ -1967,6 +1976,16 @@ export function createSyncHostService(args: SyncHostServiceArgs) { let disposed = false; let startupError: Error | null = null; + const loopbackProbe = args.loopbackProbe ?? probeAdeLoopbackListener; + let loopbackValidationStatus: SyncLoopbackValidationStatus = sharedListener + ? sharedListener.getLoopbackValidationStatus() + : { + port: null, + loopbackAdeValidated: false, + lastFailureAt: null, + reason: "The sync host listener has not been validated yet.", + lastSuccessAt: null, + }; let bonjourInstance: Bonjour | null = null; let bonjourAnnouncement: BonjourService | null = null; let nativeBonjourProcess: ChildProcess | null = null; @@ -2841,6 +2860,47 @@ export function createSyncHostService(args: SyncHostServiceArgs) { } }; + const validateListeningPort = async (port: number): Promise => { + if ( + loopbackValidationStatus.port === port + && loopbackValidationStatus.loopbackAdeValidated + ) return; + try { + const result = await assertAdeLoopbackListener(port, loopbackProbe); + loopbackValidationStatus = { + port, + loopbackAdeValidated: true, + lastFailureAt: loopbackValidationStatus.lastFailureAt, + reason: null, + lastSuccessAt: result.checkedAt, + }; + } catch (error) { + if (isLoopbackShadowedError(error)) { + loopbackValidationStatus = { + port, + loopbackAdeValidated: false, + lastFailureAt: error.failedAt, + reason: error.message, + lastSuccessAt: loopbackValidationStatus.lastSuccessAt, + }; + } + throw error; + } + }; + + const publishValidatedDiscovery = async ( + port: number, + options?: { forceLan?: boolean; forceTailnet?: boolean }, + ): Promise => { + const lanPortChanged = bonjourPort != null && bonjourPort !== port; + const tailnetPortChanged = tailnetServePort != null && tailnetServePort !== port; + if (lanPortChanged) unpublishLanDiscovery(); + if (tailnetPortChanged) await unpublishTailnetDiscovery(); + if (disposed) return; + publishLanDiscovery(port, { force: options?.forceLan }); + publishTailnetDiscovery(port, { force: options?.forceTailnet }); + }; + function peerForSocket(ws: WebSocket): PeerState | null { for (const peer of peers) { if (peer.ws === ws) return peer; @@ -5312,8 +5372,11 @@ export function createSyncHostService(args: SyncHostServiceArgs) { // ensureListening is idempotent and returns the existing port. const port = sharedListener!.getPort() ?? await sharedListener!.ensureListening([args.port ?? DEFAULT_SYNC_HOST_PORT]); - publishLanDiscovery(port); - publishTailnetDiscovery(port); + loopbackValidationStatus = sharedListener!.getLoopbackValidationStatus(); + if (!loopbackValidationStatus.loopbackAdeValidated || loopbackValidationStatus.port !== port) { + throw new Error(`The shared sync listener on 127.0.0.1:${port} was not ADE-validated.`); + } + await publishValidatedDiscovery(port); return port; } if (startupError) { @@ -5322,8 +5385,8 @@ export function createSyncHostService(args: SyncHostServiceArgs) { if (server.address()) { const address = server.address(); const port = typeof address === "object" && address ? address.port : DEFAULT_SYNC_HOST_PORT; - publishLanDiscovery(port); - publishTailnetDiscovery(port); + await validateListeningPort(port); + await publishValidatedDiscovery(port); return port; } await new Promise((resolve, reject) => { @@ -5355,8 +5418,8 @@ export function createSyncHostService(args: SyncHostServiceArgs) { }); const address = server.address(); const port = typeof address === "object" && address ? address.port : DEFAULT_SYNC_HOST_PORT; - publishLanDiscovery(port); - publishTailnetDiscovery(port); + await validateListeningPort(port); + await publishValidatedDiscovery(port); return port; }, @@ -5375,8 +5438,12 @@ export function createSyncHostService(args: SyncHostServiceArgs) { refreshLanDiscovery(options?: { forceLan?: boolean; forceTailnet?: boolean }): void { const port = getListeningPort(); if (port != null) { - publishLanDiscovery(port, { force: options?.forceLan }); - publishTailnetDiscovery(port, { force: options?.forceTailnet }); + void publishValidatedDiscovery(port, options).catch((error) => { + args.logger.warn("sync_host.discovery_refresh_failed", { + port, + error: error instanceof Error ? error.message : String(error), + }); + }); } }, @@ -5453,6 +5520,10 @@ export function createSyncHostService(args: SyncHostServiceArgs) { return { ...tailnetDiscoveryStatus }; }, + getLoopbackValidationStatus(): SyncLoopbackValidationStatus { + return { ...loopbackValidationStatus }; + }, + getLanePresenceSnapshot(): Array<{ laneId: string; devicesOpen: DeviceMarker[] }> { return getLanePresenceSnapshot(); }, diff --git a/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts new file mode 100644 index 000000000..30fb8602e --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts @@ -0,0 +1,253 @@ +import fs from "node:fs"; +import http from "node:http"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { openKvDb, type AdeDb } from "../../../../desktop/src/main/services/state/kvDb"; +import { createSharedSyncListener } from "./sharedSyncListener"; +import { createSyncService, type SyncService } from "./syncService"; +import { probeAdeLoopbackListener, type SyncLoopbackProbeResult } from "./syncLoopbackProbe"; + +const ORIGINAL_BIND_HOST = vi.hoisted(() => process.env.ADE_SYNC_BIND_HOST); +vi.hoisted(() => { + process.env.ADE_SYNC_BIND_HOST = "0.0.0.0"; +}); + +const publishMock = vi.hoisted(() => vi.fn()); +const bonjourDestroyMock = vi.hoisted(() => vi.fn()); +const bonjourConstructorMock = vi.hoisted(() => vi.fn()); + +vi.mock("bonjour-service", () => ({ + Bonjour: bonjourConstructorMock, +})); + +afterAll(() => { + if (ORIGINAL_BIND_HOST === undefined) delete process.env.ADE_SYNC_BIND_HOST; + else process.env.ADE_SYNC_BIND_HOST = ORIGINAL_BIND_HOST; +}); + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createService( + db: AdeDb, + projectRoot: string, + overrides: Partial[0]> = {}, +): SyncService { + return createSyncService({ + db, + logger: createLogger() as any, + projectRoot, + hostStartupEnabled: true, + hostDiscoveryEnabled: true, + forceHostRole: true, + localDeviceIdPath: path.join(projectRoot, ".ade", "secrets", "sync-device-id"), + phonePairingStateDir: path.join(projectRoot, ".ade", "secrets", "sync"), + fileService: {} as any, + laneService: { list: vi.fn(async () => []) } as any, + prService: {} as any, + sessionService: { + list: vi.fn(() => []), + get: vi.fn(() => null), + readTranscriptTail: vi.fn(async () => ""), + } as any, + ptyService: { + readTranscriptTail: vi.fn(async () => ""), + enrichSessions: vi.fn((rows: unknown[]) => rows), + } as any, + computerUseArtifactBrokerService: { listArtifacts: vi.fn(() => []) } as any, + agentChatService: { + listSessions: vi.fn(async () => []), + subscribeToEvents: vi.fn(() => () => {}), + } as any, + processService: { listRuntime: vi.fn(() => []) } as any, + ...overrides, + }); +} + +async function listen(server: http.Server, port: number, host: string): Promise { + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(port, host, () => { + server.off("error", reject); + resolve(); + }); + }); +} + +async function close(server: http.Server): Promise { + await new Promise((resolve) => server.close(() => resolve())); +} + +async function bindForeignLegacyListener(): Promise<{ server: http.Server; port: number }> { + for (let port = 8787; port <= 8800; port += 1) { + const server = http.createServer((_request, response) => { + response.writeHead(404, "Not Found"); + response.end("foreign"); + }); + try { + await listen(server, port, "127.0.0.1"); + return { server, port }; + } catch { + try { + server.close(); + } catch {} + } + } + throw new Error("No free legacy sync port was available for the collision test."); +} + +async function findFreeLegacyPort(): Promise { + for (let port = 8787; port <= 8800; port += 1) { + const server = http.createServer(); + try { + await listen(server, port, "127.0.0.1"); + await close(server); + return port; + } catch { + try { + server.close(); + } catch {} + } + } + throw new Error("No free legacy sync port was available for the startup-order test."); +} + +function publishedPorts(): number[] { + return publishMock.mock.calls + .map(([options]) => (options as { port?: unknown }).port) + .filter((port): port is number => typeof port === "number"); +} + +describe("sync loopback collision recovery", () => { + beforeEach(() => { + publishMock.mockReset(); + bonjourDestroyMock.mockReset(); + bonjourConstructorMock.mockReset(); + publishMock.mockImplementation(() => ({ on: vi.fn(), stop: vi.fn() })); + bonjourConstructorMock.mockImplementation(() => ({ + publish: publishMock, + destroy: bonjourDestroyMock, + })); + }); + + it.runIf(process.platform === "darwin")( + "scans past a foreign 127.0.0.1 listener and publishes only the ADE-validated port", + async () => { + const projectRoot = fs.mkdtempSync(path.join(os.tmpdir(), "ade-sync-loopback-shadow-")); + const lockPath = path.join(projectRoot, "sync-host.lock"); + const previousLockPath = process.env.ADE_SYNC_HOST_LOCK_PATH; + process.env.ADE_SYNC_HOST_LOCK_PATH = lockPath; + const foreign = await bindForeignLegacyListener(); + const listener = createSharedSyncListener({ bindHost: "0.0.0.0" }); + const db = await openKvDb(path.join(projectRoot, ".ade", "kv.sqlite"), createLogger() as any); + (db.sync as { isAvailable?: () => boolean }).isAvailable = () => true; + const service = createService(db, projectRoot, { sharedSyncListener: listener }); + service.getDeviceRegistryService().touchLocalDevice({ lastPort: foreign.port }); + + try { + await service.initialize(); + const status = await service.getStatus({ includeTransferReadiness: false }); + const resolvedPort = status.routeHealth.listener.port; + + expect(resolvedPort).not.toBe(foreign.port); + expect(status.routeHealth.listener).toMatchObject({ + listenerBound: true, + loopbackAdeValidated: true, + }); + expect(status.routeHealth.listener.lastFailureAt).not.toBeNull(); + expect(status.localDevice.lastPort).toBe(resolvedPort); + expect(status.pairingConnectInfo?.port).toBe(resolvedPort); + expect(status.tailnetDiscovery).toMatchObject({ + servicePort: resolvedPort, + updatedAt: expect.any(String), + }); + expect(publishedPorts()).not.toContain(foreign.port); + expect(new Set(publishedPorts())).toEqual(new Set([resolvedPort!])); + await expect(probeAdeLoopbackListener(foreign.port)).resolves.toMatchObject({ + ok: false, + statusCode: 404, + }); + await expect(probeAdeLoopbackListener(resolvedPort!)).resolves.toMatchObject({ + ok: true, + statusCode: 426, + }); + } finally { + await service.dispose(); + await listener.close(); + db.close(); + await close(foreign.server); + fs.rmSync(projectRoot, { recursive: true, force: true }); + if (previousLockPath === undefined) delete process.env.ADE_SYNC_HOST_LOCK_PATH; + else process.env.ADE_SYNC_HOST_LOCK_PATH = previousLockPath; + } + }, + ); + + it("does not publish or persist a candidate until its loopback check passes", async () => { + const projectRoot = fs.mkdtempSync(path.join(os.tmpdir(), "ade-sync-loopback-order-")); + const lockPath = path.join(projectRoot, "sync-host.lock"); + const previousLockPath = process.env.ADE_SYNC_HOST_LOCK_PATH; + process.env.ADE_SYNC_HOST_LOCK_PATH = lockPath; + const db = await openKvDb(path.join(projectRoot, ".ade", "kv.sqlite"), createLogger() as any); + (db.sync as { isAvailable?: () => boolean }).isAvailable = () => true; + const preferredPort = await findFreeLegacyPort(); + let releaseFirstProbe: ((result: SyncLoopbackProbeResult) => void) | null = null; + let firstProbePort: number | null = null; + const firstProbe = new Promise((resolve) => { + releaseFirstProbe = resolve; + }); + const loopbackProbe = vi.fn(async (port: number) => { + if (firstProbePort == null) { + firstProbePort = port; + return await firstProbe; + } + return await probeAdeLoopbackListener(port); + }); + const listener = createSharedSyncListener({ bindHost: "127.0.0.1", loopbackProbe }); + const service = createService(db, projectRoot, { sharedSyncListener: listener }); + service.getDeviceRegistryService().touchLocalDevice({ lastPort: preferredPort }); + + try { + const initializing = service.initialize(); + await vi.waitFor(() => expect(firstProbePort).not.toBeNull()); + expect(publishMock).not.toHaveBeenCalled(); + expect(service.getDeviceRegistryService().ensureLocalDevice().lastPort).toBe(preferredPort); + + releaseFirstProbe!({ + ok: false, + port: firstProbePort!, + statusCode: 404, + statusMessage: "Not Found", + checkedAt: new Date().toISOString(), + reason: "foreign loopback listener", + }); + await initializing; + + const status = await service.getStatus({ includeTransferReadiness: false }); + const resolvedPort = status.routeHealth.listener.port; + expect(resolvedPort).not.toBe(firstProbePort); + expect(status.localDevice.lastPort).toBe(resolvedPort); + expect(status.routeHealth.listener.lastFailureAt).not.toBeNull(); + expect(status.tailnetDiscovery).toMatchObject({ + servicePort: resolvedPort, + updatedAt: expect.any(String), + }); + expect(publishedPorts()).not.toContain(firstProbePort); + expect(new Set(publishedPorts())).toEqual(new Set([resolvedPort!])); + } finally { + await service.dispose(); + await listener.close(); + db.close(); + fs.rmSync(projectRoot, { recursive: true, force: true }); + if (previousLockPath === undefined) delete process.env.ADE_SYNC_HOST_LOCK_PATH; + else process.env.ADE_SYNC_HOST_LOCK_PATH = previousLockPath; + } + }); +}); diff --git a/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts b/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts new file mode 100644 index 000000000..4df5c8add --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts @@ -0,0 +1,105 @@ +import http from "node:http"; + +export const SYNC_LOOPBACK_PROBE_TIMEOUT_MS = 1_500; + +export type SyncLoopbackProbeResult = { + ok: boolean; + port: number; + statusCode: number | null; + statusMessage: string | null; + checkedAt: string; + reason: string | null; +}; + +export type SyncLoopbackValidationStatus = { + port: number | null; + loopbackAdeValidated: boolean; + lastFailureAt: string | null; + reason: string | null; + lastSuccessAt: string | null; +}; + +export class LoopbackShadowedError extends Error { + readonly code = "ELOOPBACKSHADOWED"; + readonly port: number; + readonly failedAt: string; + + constructor(port: number, reason: string, failedAt = new Date().toISOString()) { + super(reason); + this.name = "LoopbackShadowedError"; + this.port = port; + this.failedAt = failedAt; + } +} + +export function isLoopbackShadowedError(error: unknown): error is LoopbackShadowedError { + return error instanceof LoopbackShadowedError + || (error as { code?: unknown } | null | undefined)?.code === "ELOOPBACKSHADOWED"; +} + +/** + * ADE's WebSocketServer answers a plain HTTP request with 426 Upgrade Required. + * Probe the exact loopback route used by Tailscale Serve and the cloud relay so + * a more-specific foreign 127.0.0.1 listener cannot hide behind a successful + * wildcard bind. + */ +export async function probeAdeLoopbackListener( + port: number, + timeoutMs = SYNC_LOOPBACK_PROBE_TIMEOUT_MS, +): Promise { + const checkedAt = new Date().toISOString(); + return await new Promise((resolve) => { + let settled = false; + const finish = (result: Omit): void => { + if (settled) return; + settled = true; + resolve({ ...result, port, checkedAt }); + }; + const request = http.get({ + host: "127.0.0.1", + port, + path: "/", + headers: { connection: "close" }, + }, (response) => { + response.resume(); + const statusCode = response.statusCode ?? null; + const statusMessage = response.statusMessage ?? null; + const ok = statusCode === 426 + && (statusMessage == null || statusMessage.toLowerCase() === "upgrade required"); + finish({ + ok, + statusCode, + statusMessage, + reason: ok + ? null + : `Expected ADE 426 Upgrade Required on 127.0.0.1:${port}, received ${statusCode ?? "no status"}${statusMessage ? ` ${statusMessage}` : ""}.`, + }); + }); + request.setTimeout(timeoutMs, () => { + request.destroy(new Error(`Loopback ADE probe timed out after ${timeoutMs}ms.`)); + }); + request.once("error", (error) => { + finish({ + ok: false, + statusCode: null, + statusMessage: null, + reason: `ADE loopback probe failed on 127.0.0.1:${port}: ${error.message}`, + }); + }); + }); +} + +export async function assertAdeLoopbackListener( + port: number, + probe: (port: number) => Promise = probeAdeLoopbackListener, +): Promise { + const result = await probe(port); + if (!result.ok) { + throw new LoopbackShadowedError( + port, + result.reason ?? `The listener on 127.0.0.1:${port} is not ADE.`, + result.checkedAt, + ); + } + return result; +} diff --git a/apps/ade-cli/src/services/sync/syncService.ts b/apps/ade-cli/src/services/sync/syncService.ts index 9e4ae2c40..383b45d84 100644 --- a/apps/ade-cli/src/services/sync/syncService.ts +++ b/apps/ade-cli/src/services/sync/syncService.ts @@ -9,6 +9,7 @@ import type { SyncGetStatusArgs, SyncPairingConnectInfo, PersonalChatScopeContract, + SyncRouteHealth, SyncRoleSnapshot, SyncTailnetDiscoveryStatus, SyncTransferBlocker, @@ -81,6 +82,12 @@ import { getSharedAccountAuthService, type AccountAttestationConfig, } from "../account/sharedAccountAuthService"; +import type { SyncTunnelClientService } from "./syncTunnelClientService"; +import { + isLoopbackShadowedError, + type SyncLoopbackProbeResult, + type SyncLoopbackValidationStatus, +} from "./syncLoopbackProbe"; type SyncServiceArgs = { db: AdeDb; @@ -159,6 +166,7 @@ type SyncServiceArgs = { * absent a store is created under the pairing state dir. */ cloudRelayStore?: SyncCloudRelayStore; + syncTunnelClientService?: Pick | null; /** Fired when the ADE relay kill-switch flips (start/stop tunnel). */ onCloudRelayEnabledChanged?: (enabled: boolean) => void; projectCatalogProvider?: SyncProjectCatalogProvider; @@ -179,6 +187,8 @@ type SyncServiceArgs = { * `deeplinks.open` sync command reports unavailable. */ dispatchDeeplinkUrl?: (url: string) => Promise<{ ok: boolean; message?: string }>; + /** Test seam for self-owned host startup; production uses the HTTP 426 probe. */ + loopbackProbe?: (port: number) => Promise; }; const DRAFT_FILE = "sync-peer-draft.json"; @@ -364,6 +374,7 @@ function isViewerDraftTransportError(error: unknown): boolean { } function isRetryableHostBindError(error: unknown): boolean { + if (isLoopbackShadowedError(error)) return true; const code = (error as NodeJS.ErrnoException | null | undefined)?.code ?? ""; return code === "EADDRINUSE" || code === "EACCES"; } @@ -470,6 +481,10 @@ export function createSyncService(args: SyncServiceArgs) { let hostService: SyncHostService | null = null; let hostSingletonLease: SyncHostSingletonLease | null = null; + let listenerValidationHistory: Pick = { + lastFailureAt: null, + lastSuccessAt: null, + }; let refreshRunning = false; let refreshQueued = false; let disposed = false; @@ -743,11 +758,17 @@ export function createSyncService(args: SyncServiceArgs) { requireDpop: () => securityStore.getRequireDpop(), getCloudRelayWssUrl: () => cloudRelayStore.isEnabled() ? cloudRelayStore.getRelayWssUrl() : null, + loopbackProbe: args.loopbackProbe, onStateChanged: () => { void refreshRoleState(); }, }); const finishHostStartup = (started: SyncHostService, resolvedPort: number): void => { + const validation = started.getLoopbackValidationStatus(); + listenerValidationHistory = { + lastFailureAt: validation.lastFailureAt ?? listenerValidationHistory.lastFailureAt, + lastSuccessAt: validation.lastSuccessAt ?? listenerValidationHistory.lastSuccessAt, + }; hostService = started; hostSingletonLease?.updatePort(resolvedPort); hostService.setLocalActiveLanePresence?.(activeLocalLanePresenceIds); @@ -789,7 +810,9 @@ export function createSyncService(args: SyncServiceArgs) { : [candidatePort], ); let previousAttemptedPort: number | null = null; + const shadowedPorts = new Set(); for (const attemptedPort of attemptPlan) { + if (shadowedPorts.has(attemptedPort)) continue; if (previousAttemptedPort === attemptedPort) { await new Promise((resolve) => setTimeout(resolve, PREFERRED_PORT_BIND_RETRY_DELAY_MS)); } @@ -801,10 +824,20 @@ export function createSyncService(args: SyncServiceArgs) { return; } catch (error) { lastError = error; + if (isLoopbackShadowedError(error)) { + shadowedPorts.add(attemptedPort); + listenerValidationHistory = { + ...listenerValidationHistory, + lastFailureAt: error.failedAt, + }; + } await candidateHostService.dispose().catch(() => {}); - const retryable = isRetryableHostBindError(error) && attemptedPort !== 0; + const retryable = isRetryableHostBindError(error) + && (attemptedPort !== 0 || isLoopbackShadowedError(error)); args.logger.warn( - retryable ? "sync.host_start_port_conflict" : "sync.host_start_failed", + isLoopbackShadowedError(error) + ? "sync.host_start_loopback_shadowed" + : retryable ? "sync.host_start_port_conflict" : "sync.host_start_failed", { preferredPort, attemptedPort, @@ -1166,6 +1199,90 @@ export function createSyncService(args: SyncServiceArgs) { ...peer, isHost: Boolean(peer.isHost ?? peer.isBrain), })); + const tailnetDiscovery = canHostPhonePairing && hostService + ? hostService.getTailnetDiscoveryStatus() + : createInactiveTailnetDiscoveryStatus( + canHostPhonePairing + ? "Tailnet discovery is waiting for the ADE runtime to start." + : "Tailnet discovery is only published by the host ADE runtime.", + ); + const listenerPort = hostService?.getPort() ?? args.sharedSyncListener?.getPort() ?? null; + const rawListenerValidation = hostService?.getLoopbackValidationStatus() + ?? args.sharedSyncListener?.getLoopbackValidationStatus() + ?? { + port: null, + loopbackAdeValidated: false, + lastFailureAt: listenerValidationHistory.lastFailureAt, + reason: "The ADE sync listener has not started.", + lastSuccessAt: listenerValidationHistory.lastSuccessAt, + }; + const listenerBound = listenerPort != null; + const loopbackAdeValidated = listenerBound + && rawListenerValidation.port === listenerPort + && rawListenerValidation.loopbackAdeValidated; + const listenerReason = !listenerBound + ? "The ADE sync listener is not bound." + : loopbackAdeValidated + ? null + : rawListenerValidation.reason + ?? `127.0.0.1:${listenerPort} did not answer as ADE.`; + const tunnelStatus = args.syncTunnelClientService?.getStatus() ?? null; + const tailscalePublished = tailnetDiscovery.state === "published"; + const tailscaleEnabled = canHostPhonePairing + && tailnetDiscovery.state !== "disabled" + && tailnetDiscovery.state !== "unavailable"; + const tailscaleReachable = tailscalePublished && loopbackAdeValidated; + const tailscaleReason = !tailscaleEnabled + ? tailnetDiscovery.error + : !loopbackAdeValidated + ? `Tailscale route is unusable because ${listenerReason ?? "the loopback ADE check failed"}` + : tailscalePublished + ? null + : tailnetDiscovery.error ?? `Tailscale Serve is ${tailnetDiscovery.state}.`; + const relayEnabled = canHostPhonePairing && cloudRelayStore.isEnabled(); + const relayControlConnected = tunnelStatus?.connected === true; + const relayBridgeValidated = tunnelStatus?.relayBridgeValidated === true; + const relayReason = !relayEnabled + ? null + : !loopbackAdeValidated + ? `Relay route is unusable because ${listenerReason ?? "the loopback ADE check failed"}` + : !tunnelStatus + ? "Relay tunnel status is unavailable in this ADE process." + : !relayControlConnected + ? tunnelStatus.lastError ?? "Relay control is not connected." + : tunnelStatus.lastError; + const routeHealth: SyncRouteHealth = { + listener: { + listenerBound, + loopbackAdeValidated, + port: listenerPort, + lastFailureAt: rawListenerValidation.lastFailureAt ?? listenerValidationHistory.lastFailureAt, + reason: listenerReason, + lastSuccessAt: rawListenerValidation.lastSuccessAt ?? listenerValidationHistory.lastSuccessAt, + }, + tailscale: { + enabled: tailscaleEnabled, + tailscalePublished, + tailscaleReachable, + lastFailureAt: tailscaleEnabled && !tailscaleReachable + ? (tailnetDiscovery.state === "failed" + ? tailnetDiscovery.updatedAt + : rawListenerValidation.lastFailureAt) + : null, + reason: tailscaleReason, + lastSuccessAt: tailscaleReachable ? tailnetDiscovery.updatedAt : null, + }, + relay: { + enabled: relayEnabled, + relayControlConnected, + relayBridgeValidated, + lastFailureAt: relayEnabled && relayReason + ? (tunnelStatus?.lastFailureAt ?? rawListenerValidation.lastFailureAt) + : null, + reason: relayReason, + lastSuccessAt: relayReason == null ? (tunnelStatus?.lastSuccessAt ?? null) : null, + }, + }; return { mode, role, @@ -1188,13 +1305,8 @@ export function createSyncService(args: SyncServiceArgs) { }) : null, connectedPeers, - tailnetDiscovery: canHostPhonePairing && hostService - ? hostService.getTailnetDiscoveryStatus() - : createInactiveTailnetDiscoveryStatus( - canHostPhonePairing - ? "Tailnet discovery is waiting for the ADE runtime to start." - : "Tailnet discovery is only published by the host ADE runtime.", - ), + tailnetDiscovery, + routeHealth, client, transferReadiness: options?.includeTransferReadiness === false ? (transferReadinessCache?.value ?? buildSkippedTransferReadiness()) @@ -1350,11 +1462,18 @@ export function createSyncService(args: SyncServiceArgs) { getCloudRelayStatus(): SyncCloudRelayStatus { const config = cloudRelayStore.getConfig(); + const tunnelStatus = args.syncTunnelClientService?.getStatus() ?? null; return { enabled: config.enabled, relayWssUrl: cloudRelayStore.getRelayWssUrl(), machineKey: config.machineKey, relayUrl: cloudRelayStore.getRelayUrl(), + connected: tunnelStatus?.connected ?? false, + activeTunnels: tunnelStatus?.activeTunnels ?? 0, + relayBridgeValidated: tunnelStatus?.relayBridgeValidated ?? false, + lastFailureAt: tunnelStatus?.lastFailureAt ?? null, + lastSuccessAt: tunnelStatus?.lastSuccessAt ?? null, + lastError: tunnelStatus?.lastError ?? null, }; }, diff --git a/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts b/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts index 5e0dd8d98..7ad9270dd 100644 --- a/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts +++ b/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts @@ -1,4 +1,5 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; +import { WebSocketServer } from "ws"; import { computeBackoffMs, createSyncTunnelClientService, @@ -9,14 +10,14 @@ import type { SyncCloudRelayStore } from "./syncCloudRelayStore"; // syncCloudRelayStore itself (enablement default/migration, identity mint, url // derivation, signature builders) is covered in syncCloudRelayStore.test.ts. -function fakeStore(enabled: boolean): SyncCloudRelayStore { +function fakeStore(enabled: boolean, relayUrl = "https://relay.example.com"): SyncCloudRelayStore { const identity = { machineKey: "a".repeat(32), secret: "b".repeat(48) }; return { getConfig: () => ({ enabled, ...identity }), isEnabled: () => enabled, setEnabled: () => ({ enabled, ...identity }), getMachineIdentity: () => identity, - getRelayUrl: () => "https://relay.example.com", + getRelayUrl: () => relayUrl, setRelayUrl: () => ({ enabled, ...identity }), getRelayWssUrl: () => `wss://relay.example.com/connect/${identity.machineKey}`, } as unknown as SyncCloudRelayStore; @@ -72,4 +73,54 @@ describe("createSyncTunnelClientService", () => { expect(status.relayUrl).toBe("https://relay.example.com"); await service.dispose(); }); + + it("refuses a relay pipe before forwarding when loopback is not ADE", async () => { + const relay = new WebSocketServer({ host: "127.0.0.1", port: 0 }); + await new Promise((resolve, reject) => { + relay.once("listening", resolve); + relay.once("error", reject); + }); + const address = relay.address(); + const relayPort = typeof address === "object" && address ? address.port : 0; + const connections: string[] = []; + relay.on("connection", (socket, request) => { + connections.push(request.url ?? ""); + if (connections.length === 1) { + socket.send(JSON.stringify({ t: "open", id: "abcdef01" })); + } + }); + const originalFetch = globalThis.fetch; + globalThis.fetch = async () => new Response(null, { status: 204 }); + const service = createSyncTunnelClientService({ + getSyncPort: () => 8787, + configStore: fakeStore(true, `http://127.0.0.1:${relayPort}`), + loopbackProbe: async (port) => ({ + ok: false, + port, + statusCode: 404, + statusMessage: "Not Found", + checkedAt: new Date().toISOString(), + reason: "foreign listener returned 404", + }), + }); + + try { + await service.start(); + await vi.waitFor(() => { + expect(service.getStatus().lastError).toContain("Relay bridge refused"); + }); + expect(connections).toHaveLength(1); + expect(connections[0]).toContain(`/host/${"a".repeat(32)}`); + expect(service.getStatus()).toMatchObject({ + connected: true, + activeTunnels: 0, + relayBridgeValidated: false, + }); + expect(service.getStatus().lastFailureAt).not.toBeNull(); + } finally { + await service.dispose(); + globalThis.fetch = originalFetch; + await new Promise((resolve) => relay.close(() => resolve())); + } + }); }); diff --git a/apps/ade-cli/src/services/sync/syncTunnelClientService.ts b/apps/ade-cli/src/services/sync/syncTunnelClientService.ts index b579b98a3..4d9056203 100644 --- a/apps/ade-cli/src/services/sync/syncTunnelClientService.ts +++ b/apps/ade-cli/src/services/sync/syncTunnelClientService.ts @@ -7,6 +7,11 @@ import { signRelayHmacHex, type SyncCloudRelayStore, } from "./syncCloudRelayStore"; +import { + assertAdeLoopbackListener, + probeAdeLoopbackListener, + type SyncLoopbackProbeResult, +} from "./syncLoopbackProbe"; type Logger = { info?: (event: string, data?: Record) => void; @@ -20,6 +25,10 @@ export type SyncTunnelClientStatus = { connected: boolean; activeTunnels: number; lastError: string | null; + relayBridgeValidated: boolean; + validatedPort: number | null; + lastFailureAt: string | null; + lastSuccessAt: string | null; relayUrl: string; machineKey: string; }; @@ -40,6 +49,8 @@ type SyncTunnelClientArgs = { configStore: SyncCloudRelayStore; /** Overrides the identity from configStore (e.g. a shared machine store). */ machineIdentity?: () => MachineIdentity | null; + /** Test seam; production always uses the HTTP 426 loopback probe. */ + loopbackProbe?: (port: number) => Promise; }; const BACKOFF_BASE_MS = 1_000; @@ -95,8 +106,17 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT let stopped = false; let connected = false; let lastError: string | null = null; + let validatedPort: number | null = null; + let lastFailureAt: string | null = null; + let lastSuccessAt: string | null = null; let claimed = false; const tunnels = new Set(); + const loopbackProbe = args.loopbackProbe ?? probeAdeLoopbackListener; + + const recordFailure = (reason: string): void => { + lastError = reason; + lastFailureAt = new Date().toISOString(); + }; const identity = (): MachineIdentity => { const override = args.machineIdentity?.(); @@ -145,7 +165,7 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT try { await claimOnce(id); } catch (error) { - lastError = error instanceof Error ? error.message : String(error); + recordFailure(error instanceof Error ? error.message : String(error)); log.warn?.("sync_tunnel.claim_failed", { error: lastError }); scheduleReconnect(); return; @@ -158,13 +178,14 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT const socket = new WebSocket(url); control = socket; armOpenDeadline(socket, () => { - lastError = "relay control socket connect timed out"; + recordFailure("relay control socket connect timed out"); }); socket.on("open", () => { attempt = 0; connected = true; lastError = null; + lastSuccessAt = new Date().toISOString(); log.info?.("sync_tunnel.control_open", { machineKey: id.machineKey }); }); socket.on("message", (raw: RawData) => { @@ -172,7 +193,7 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT if (message?.t === "open") void openTunnel(id, message.id); }); socket.on("error", (error: Error) => { - lastError = error.message; + recordFailure(error.message); log.warn?.("sync_tunnel.control_error", { error: error.message }); }); socket.on("close", () => { @@ -185,9 +206,26 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT const openTunnel = async (id: MachineIdentity, connectionId: string): Promise => { const port = args.getSyncPort(); if (port == null) { + recordFailure("Relay bridge refused because the ADE sync listener is not bound."); log.warn?.("sync_tunnel.no_sync_port", { connectionId }); return; } + try { + const result = await assertAdeLoopbackListener(port, loopbackProbe); + validatedPort = port; + lastError = null; + lastSuccessAt = result.checkedAt; + } catch (error) { + validatedPort = null; + const reason = `Relay bridge refused because 127.0.0.1:${port} is not the ADE sync listener: ${error instanceof Error ? error.message : String(error)}`; + recordFailure(reason); + log.warn?.("sync_tunnel.loopback_validation_failed", { + connectionId, + port, + error: reason, + }); + return; + } const ts = nowSeconds(); const sig = signRelayHmacHex(id.secret, buildPipeSignatureBase(id.machineKey, connectionId, ts)); const pipeUrl = `${httpToWsUrl(relayHttpUrl())}/host/${id.machineKey}/pipe/${connectionId}?ts=${ts}&sig=${sig}`; @@ -195,10 +233,10 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT const pipe = new WebSocket(pipeUrl); const local = new WebSocket(`ws://127.0.0.1:${String(port)}`); armOpenDeadline(pipe, () => { - lastError = "relay pipe connect timed out"; + recordFailure("relay pipe connect timed out"); }); armOpenDeadline(local, () => { - lastError = "local sync socket connect timed out"; + recordFailure("local sync socket connect timed out"); }); const tunnel: Tunnel = { pipe, local, connectionId }; tunnels.add(tunnel); @@ -230,11 +268,11 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT pipe.on("close", closeBoth); local.on("close", closeBoth); pipe.on("error", (error: Error) => { - lastError = error.message; + recordFailure(error.message); closeBoth(); }); local.on("error", (error: Error) => { - lastError = error.message; + recordFailure(error.message); closeBoth(); }); log.debug?.("sync_tunnel.open", { connectionId }); @@ -282,11 +320,16 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT getStatus(): SyncTunnelClientStatus { const { machineKey } = identity(); + const currentPort = args.getSyncPort(); return { enabled: args.configStore.isEnabled(), connected, activeTunnels: tunnels.size, lastError, + relayBridgeValidated: currentPort != null && validatedPort === currentPort, + validatedPort, + lastFailureAt, + lastSuccessAt, relayUrl: relayHttpUrl(), machineKey, }; diff --git a/apps/desktop/src/shared/types/sync.ts b/apps/desktop/src/shared/types/sync.ts index 247dccb15..bdb2a4513 100644 --- a/apps/desktop/src/shared/types/sync.ts +++ b/apps/desktop/src/shared/types/sync.ts @@ -219,6 +219,33 @@ export type SyncTailnetDiscoveryStatus = { stderr: string | null; }; +export type SyncRouteHealth = { + listener: { + listenerBound: boolean; + loopbackAdeValidated: boolean; + port: number | null; + lastFailureAt: string | null; + reason: string | null; + lastSuccessAt: string | null; + }; + tailscale: { + enabled: boolean; + tailscalePublished: boolean; + tailscaleReachable: boolean; + lastFailureAt: string | null; + reason: string | null; + lastSuccessAt: string | null; + }; + relay: { + enabled: boolean; + relayControlConnected: boolean; + relayBridgeValidated: boolean; + lastFailureAt: string | null; + reason: string | null; + lastSuccessAt: string | null; + }; +}; + export type SyncRoleSnapshot = { mode: SyncMode; role: SyncRole; @@ -238,6 +265,7 @@ export type SyncRoleSnapshot = { pairingConnectInfo: SyncPairingConnectInfo | null; connectedPeers: SyncPeerConnectionState[]; tailnetDiscovery: SyncTailnetDiscoveryStatus; + routeHealth: SyncRouteHealth; client: SyncClientStatus; transferReadiness: SyncTransferReadiness; survivableStateText: string; @@ -594,6 +622,12 @@ export type SyncCloudRelayStatus = { machineKey: string; /** http(s) base URL of the relay worker. */ relayUrl: string; + connected: boolean; + activeTunnels: number; + relayBridgeValidated: boolean; + lastFailureAt: string | null; + lastSuccessAt: string | null; + lastError: string | null; }; /** diff --git a/apps/ios/ADE/Services/SyncService.swift b/apps/ios/ADE/Services/SyncService.swift index 1c521c708..34d158aa2 100644 --- a/apps/ios/ADE/Services/SyncService.swift +++ b/apps/ios/ADE/Services/SyncService.swift @@ -469,6 +469,31 @@ struct SyncConnectionEndpointAttempt: Equatable, Hashable { var port: Int } +@MainActor +func syncFirstSuccessfulConnectionEndpoint( + _ attempts: [SyncConnectionEndpointAttempt], + attempt: (SyncConnectionEndpointAttempt) async -> Result, + shouldContinueAfterFailure: (Error) -> Bool = { _ in true } +) async throws -> SyncConnectionEndpointAttempt { + var lastFailure: Error? + for endpoint in attempts { + switch await attempt(endpoint) { + case .success: + return endpoint + case .failure(let error): + lastFailure = error + if !shouldContinueAfterFailure(error) { + throw error + } + } + } + throw lastFailure ?? NSError( + domain: "ADE", + code: 19, + userInfo: [NSLocalizedDescriptionKey: "Unable to reach the saved ADE machine."] + ) +} + private struct SyncRankedEndpointAttempt { var attempt: SyncConnectionEndpointAttempt var routeKey: String @@ -8792,7 +8817,6 @@ final class SyncService: ObservableObject { preferLiveCandidatesOnly: Bool, publishConnecting: Bool ) async throws -> (host: String, port: Int) { - var lastFailure: Error? let matchingDiscovery = discoveredHosts.filter { host in matchesDiscoveredHost(host, profile: profile) } @@ -8851,54 +8875,57 @@ final class SyncService: ObservableObject { ) markConnectAttemptStarted(connectAttemptGeneration) - for attempt in orderedEndpointAttempts { - guard isCurrentConnectAttempt(connectAttemptGeneration) else { - throw CancellationError() - } - let kind = addressCandidateKind(attempt.address, profile: profile, explicitTailscaleAddress: nil) - syncConnectLog.info("ADE_SYNC_TRACE reconnect attempt host=\(attempt.address, privacy: .public) port=\(attempt.port) kind=\(kind, privacy: .public)") - do { - try await openSocket( - host: attempt.address, - port: attempt.port, - connectAttemptGeneration: connectAttemptGeneration, - publishConnecting: publishConnecting - ) - try await hello( - host: attempt.address, - port: attempt.port, - token: token, - authKind: profile.authKind, - pairedDeviceId: profile.pairedDeviceId, - expectedHostIdentity: profile.hostIdentity, - connectAttemptGeneration: connectAttemptGeneration - ) - guard isCurrentConnectAttempt(connectAttemptGeneration) else { - throw CancellationError() + let connectedEndpoint = try await syncFirstSuccessfulConnectionEndpoint( + orderedEndpointAttempts, + attempt: { attempt in + guard self.isCurrentConnectAttempt(connectAttemptGeneration) else { + return .failure(CancellationError()) } - syncConnectLog.info("ADE_SYNC_TRACE reconnect success host=\(attempt.address, privacy: .public) port=\(attempt.port)") - return (host: attempt.address, port: attempt.port) - } catch { - let reconnectError = errorByMarkingAmbiguousRouteAuthFailure( - error, - attemptedAddress: attempt.address, - expectedHostIdentity: profile.hostIdentity - ) - syncConnectLog.info("ADE_SYNC_TRACE reconnect failure host=\(attempt.address, privacy: .public) port=\(attempt.port) error=\(syncLogErrorSummary(reconnectError), privacy: .public)") - lastFailure = reconnectError - if shouldInvalidateSavedPairing(for: reconnectError) { - forgetHost() - throw reconnectError + let kind = self.addressCandidateKind(attempt.address, profile: profile, explicitTailscaleAddress: nil) + syncConnectLog.info("ADE_SYNC_TRACE reconnect attempt host=\(attempt.address, privacy: .public) port=\(attempt.port) kind=\(kind, privacy: .public)") + do { + try await self.openSocket( + host: attempt.address, + port: attempt.port, + connectAttemptGeneration: connectAttemptGeneration, + publishConnecting: publishConnecting + ) + try await self.hello( + host: attempt.address, + port: attempt.port, + token: token, + authKind: profile.authKind, + pairedDeviceId: profile.pairedDeviceId, + expectedHostIdentity: profile.hostIdentity, + connectAttemptGeneration: connectAttemptGeneration + ) + guard self.isCurrentConnectAttempt(connectAttemptGeneration) else { + throw CancellationError() + } + syncConnectLog.info("ADE_SYNC_TRACE reconnect success host=\(attempt.address, privacy: .public) port=\(attempt.port)") + return .success(()) + } catch { + let reconnectError = self.errorByMarkingAmbiguousRouteAuthFailure( + error, + attemptedAddress: attempt.address, + expectedHostIdentity: profile.hostIdentity + ) + syncConnectLog.info("ADE_SYNC_TRACE reconnect failure host=\(attempt.address, privacy: .public) port=\(attempt.port) error=\(syncLogErrorSummary(reconnectError), privacy: .public)") + if self.shouldInvalidateSavedPairing(for: reconnectError) { + self.forgetHost() + } else { + // A TCP/WebSocket open is not a successful candidate until hello + // completes. Tear it down and keep walking after timeout/error. + self.teardownSocket() + } + return .failure(reconnectError) } - // Tear down this attempt's socket and keep iterating through the - // remaining ports and addresses. Only surface an error if every - // candidate fails. - teardownSocket() - continue + }, + shouldContinueAfterFailure: { error in + !self.shouldInvalidateSavedPairing(for: error) } - } - - throw lastFailure ?? NSError(domain: "ADE", code: 19, userInfo: [NSLocalizedDescriptionKey: "Unable to reach the saved ADE machine."]) + ) + return (host: connectedEndpoint.address, port: connectedEndpoint.port) } private func handleReconnectFailure( @@ -9479,6 +9506,18 @@ final class SyncService: ObservableObject { publishSocketConnecting(to: host) } + func simulateHelloErrorForTesting(message: String = "Authentication failed.") -> NSError { + connectionState = .error + return NSError( + domain: "ADE", + code: 5, + userInfo: [ + NSLocalizedDescriptionKey: message, + "ADEErrorCode": "auth_failed", + ] + ) + } + func automaticReconnectAddressesForTesting(_ profile: HostConnectionProfile) -> [String] { automaticReconnectAddresses(for: profile) } diff --git a/apps/ios/ADETests/ADETests.swift b/apps/ios/ADETests/ADETests.swift index 4d31839ae..2530b6a6f 100644 --- a/apps/ios/ADETests/ADETests.swift +++ b/apps/ios/ADETests/ADETests.swift @@ -4116,6 +4116,79 @@ final class ADETests: XCTestCase { XCTAssertNotEqual(service.connectionState, .connected) } + @MainActor + func testSyncReconnectHelloTimeoutAdvancesToNextCandidateWithoutConnectingFailedSocket() async throws { + let service = SyncService(database: makeDatabase(baseURL: makeTemporaryDirectory())) + let attempts = [ + SyncConnectionEndpointAttempt(address: "192.168.1.10", port: 8787), + SyncConnectionEndpointAttempt(address: "192.168.1.11", port: 8787), + ] + var attemptedAddresses: [String] = [] + var failedCandidateStates: [RemoteConnectionState] = [] + + let winner = try await syncFirstSuccessfulConnectionEndpoint(attempts) { attempt in + attemptedAddresses.append(attempt.address) + service.simulateSocketOpenWithoutHelloForTesting(host: attempt.address) + if attempt == attempts[0] { + failedCandidateStates.append(service.connectionState) + return .failure(NSError( + domain: "ADE", + code: 2, + userInfo: [NSLocalizedDescriptionKey: "Timed out waiting for the machine."] + )) + } + do { + try service.applyHelloPayloadForTesting([ + "brain": ["deviceId": "host-1", "deviceName": "Mac Studio"], + "features": [:], + ]) + return .success(()) + } catch { + return .failure(error) + } + } + + XCTAssertEqual(attemptedAddresses, ["192.168.1.10", "192.168.1.11"]) + XCTAssertEqual(winner, attempts[1]) + XCTAssertEqual(failedCandidateStates, [.connecting]) + XCTAssertEqual(service.connectionState, .connected) + } + + @MainActor + func testSyncReconnectHelloErrorAdvancesToNextCandidateWithoutConnectingRejectedSocket() async throws { + let service = SyncService(database: makeDatabase(baseURL: makeTemporaryDirectory())) + let attempts = [ + SyncConnectionEndpointAttempt(address: "100.64.0.10", port: 8787), + SyncConnectionEndpointAttempt(address: "100.64.0.11", port: 8787), + ] + var attemptedAddresses: [String] = [] + var failedCandidateStates: [RemoteConnectionState] = [] + + let winner = try await syncFirstSuccessfulConnectionEndpoint(attempts) { attempt in + attemptedAddresses.append(attempt.address) + service.simulateSocketOpenWithoutHelloForTesting(host: attempt.address) + if attempt == attempts[0] { + let error = service.simulateHelloErrorForTesting() + failedCandidateStates.append(service.connectionState) + return .failure(error) + } + do { + try service.applyHelloPayloadForTesting([ + "brain": ["deviceId": "host-1", "deviceName": "Mac Studio"], + "features": [:], + ]) + return .success(()) + } catch { + return .failure(error) + } + } + + XCTAssertEqual(attemptedAddresses, ["100.64.0.10", "100.64.0.11"]) + XCTAssertEqual(winner, attempts[1]) + XCTAssertEqual(failedCandidateStates, [.error]) + XCTAssertEqual(service.connectionState, .connected) + } + @MainActor func testSyncServiceHelloStampsWinningRouteSuccessState() throws { let profileKey = "ade.sync.hostProfile" From 302714933a75c6ee9724ab127991178ea978d12b Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Tue, 14 Jul 2026 21:57:21 -0400 Subject: [PATCH 2/4] fix(sync): report relay route unhealthy until bridge is revalidated Gate relayReason on relayBridgeValidated so ade doctor / sync status surface a stale relay bridge (control-connected but not revalidated against the current sync port) instead of reporting the route healthy. Mirrors the tailscaleReason gating so the diagnostics this PR adds don't hide a stale-but-connected bridge. Adds a producer-level route-health regression test (relay bridge not validated => non-null relay.reason) and coverage for the shared-listener waitUntilListening ADE-validation throw path. Addresses CodeRabbit review on #816. Co-Authored-By: Claude Opus 4.8 --- .../src/services/sync/syncHostService.test.ts | 73 +++++++++++++++++++ .../sync/syncLoopbackCollision.test.ts | 67 +++++++++++++++++ apps/ade-cli/src/services/sync/syncService.ts | 4 +- 3 files changed, 143 insertions(+), 1 deletion(-) diff --git a/apps/ade-cli/src/services/sync/syncHostService.test.ts b/apps/ade-cli/src/services/sync/syncHostService.test.ts index 8e1f50f5e..42423862a 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.test.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.test.ts @@ -3093,6 +3093,79 @@ describe("sync host handoff over a shared listener", () => { }); }); +describe("shared listener waitUntilListening ADE-validation gate", () => { + beforeEach(() => { + publishMock.mockReset(); + spawnMock.mockReset(); + bonjourDestroyMock.mockReset(); + bonjourConstructorMock.mockReset(); + spawnMock.mockImplementation(() => ({ kill: vi.fn(), once: vi.fn(), unref: vi.fn() })); + bonjourConstructorMock.mockImplementation(() => ({ + publish: publishMock, + destroy: bonjourDestroyMock, + })); + publishMock.mockImplementation(() => ({ on: vi.fn(), stop: vi.fn() })); + }); + + it("throws before discovery when the shared listener loopback is not ADE-validated", async () => { + const { projectRoot, cleanup } = createTempProjectRoot(); + const listener = createSharedSyncListener({ bindHost: "127.0.0.1" }); + const boundPort = await listener.ensureListening([0]); + // The brain-level listener bound, but its loopback probe never confirmed ADE. + vi.spyOn(listener, "getLoopbackValidationStatus").mockReturnValue({ + port: boundPort, + loopbackAdeValidated: false, + lastFailureAt: new Date().toISOString(), + reason: `127.0.0.1:${boundPort} did not answer as ADE.`, + lastSuccessAt: null, + }); + const host = createSyncHostService({ + ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), + sharedListener: listener, + } as unknown as Parameters[0]); + + try { + await expect(host.waitUntilListening()).rejects.toThrow(/was not ADE-validated/); + // The host must refuse to advertise an unvalidated shared listener. + expect(publishMock).not.toHaveBeenCalled(); + expect(spawnMock).not.toHaveBeenCalled(); + } finally { + await host.dispose(); + await listener.close(); + cleanup(); + } + }); + + it("throws before discovery when the shared listener validated a different port", async () => { + const { projectRoot, cleanup } = createTempProjectRoot(); + const listener = createSharedSyncListener({ bindHost: "127.0.0.1" }); + const boundPort = await listener.ensureListening([0]); + // Loopback was validated, but for a stale port that no longer matches the + // listener's live bind — the host must reject rather than publish it. + vi.spyOn(listener, "getLoopbackValidationStatus").mockReturnValue({ + port: boundPort + 1, + loopbackAdeValidated: true, + lastFailureAt: null, + reason: null, + lastSuccessAt: new Date().toISOString(), + }); + const host = createSyncHostService({ + ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), + sharedListener: listener, + } as unknown as Parameters[0]); + + try { + await expect(host.waitUntilListening()).rejects.toThrow(/was not ADE-validated/); + expect(publishMock).not.toHaveBeenCalled(); + expect(spawnMock).not.toHaveBeenCalled(); + } finally { + await host.dispose(); + await listener.close(); + cleanup(); + } + }); +}); + describe("sync host reliability guards", () => { beforeEach(() => { publishMock.mockReset(); diff --git a/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts index 30fb8602e..460674c97 100644 --- a/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts +++ b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts @@ -5,8 +5,10 @@ import path from "node:path"; import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; import { openKvDb, type AdeDb } from "../../../../desktop/src/main/services/state/kvDb"; import { createSharedSyncListener } from "./sharedSyncListener"; +import { createSyncCloudRelayStore } from "./syncCloudRelayStore"; import { createSyncService, type SyncService } from "./syncService"; import { probeAdeLoopbackListener, type SyncLoopbackProbeResult } from "./syncLoopbackProbe"; +import type { SyncTunnelClientStatus } from "./syncTunnelClientService"; const ORIGINAL_BIND_HOST = vi.hoisted(() => process.env.ADE_SYNC_BIND_HOST); vi.hoisted(() => { @@ -250,4 +252,69 @@ describe("sync loopback collision recovery", () => { else process.env.ADE_SYNC_HOST_LOCK_PATH = previousLockPath; } }); + + // Regression for PR #816: when relay is enabled and the loopback listener is + // genuinely ADE-validated, but the relay bridge has NOT been validated against + // the current sync port, the relay route must report a non-null reason (an + // unhealthy/failing route) rather than swallowing it as a null lastError. + it.runIf(process.platform === "darwin")( + "flags the relay route unhealthy when the relay bridge is not validated against the current sync port", + async () => { + const projectRoot = fs.mkdtempSync(path.join(os.tmpdir(), "ade-sync-relay-bridge-")); + const lockPath = path.join(projectRoot, "sync-host.lock"); + const previousLockPath = process.env.ADE_SYNC_HOST_LOCK_PATH; + process.env.ADE_SYNC_HOST_LOCK_PATH = lockPath; + const listener = createSharedSyncListener({ bindHost: "127.0.0.1" }); + const db = await openKvDb(path.join(projectRoot, ".ade", "kv.sqlite"), createLogger() as any); + (db.sync as { isAvailable?: () => boolean }).isAvailable = () => true; + const cloudRelayStore = createSyncCloudRelayStore({ + filePath: path.join(projectRoot, ".ade", "secrets", "sync", "cloud-relay.json"), + }); + cloudRelayStore.setEnabled(true); + // Relay control is up, but the bridge has not been validated against the + // live sync port — exactly the state that must surface as a failing route. + const tunnelStatus: SyncTunnelClientStatus = { + enabled: true, + connected: true, + activeTunnels: 0, + lastError: null, + relayBridgeValidated: false, + validatedPort: null, + lastFailureAt: null, + lastSuccessAt: null, + relayUrl: "https://relay.test.ade", + machineKey: "a".repeat(32), + }; + const service = createService(db, projectRoot, { + sharedSyncListener: listener, + cloudRelayStore, + syncTunnelClientService: { getStatus: () => tunnelStatus }, + }); + const preferredPort = await findFreeLegacyPort(); + service.getDeviceRegistryService().touchLocalDevice({ lastPort: preferredPort }); + + try { + await service.initialize(); + const status = await service.getStatus({ includeTransferReadiness: false }); + + // Sanity: the loopback listener really is ADE-validated, so the relay + // reason below cannot be attributed to a bad loopback branch. + expect(status.routeHealth.listener.loopbackAdeValidated).toBe(true); + expect(status.routeHealth.relay.enabled).toBe(true); + expect(status.routeHealth.relay.relayControlConnected).toBe(true); + + // The regression: an unvalidated relay bridge must yield a non-null reason. + expect(status.routeHealth.relay.relayBridgeValidated).toBe(false); + expect(typeof status.routeHealth.relay.reason).toBe("string"); + expect(status.routeHealth.relay.reason).toMatch(/not been validated/); + } finally { + await service.dispose(); + await listener.close(); + db.close(); + fs.rmSync(projectRoot, { recursive: true, force: true }); + if (previousLockPath === undefined) delete process.env.ADE_SYNC_HOST_LOCK_PATH; + else process.env.ADE_SYNC_HOST_LOCK_PATH = previousLockPath; + } + }, + ); }); diff --git a/apps/ade-cli/src/services/sync/syncService.ts b/apps/ade-cli/src/services/sync/syncService.ts index 383b45d84..6b5aa5a56 100644 --- a/apps/ade-cli/src/services/sync/syncService.ts +++ b/apps/ade-cli/src/services/sync/syncService.ts @@ -1250,7 +1250,9 @@ export function createSyncService(args: SyncServiceArgs) { ? "Relay tunnel status is unavailable in this ADE process." : !relayControlConnected ? tunnelStatus.lastError ?? "Relay control is not connected." - : tunnelStatus.lastError; + : !relayBridgeValidated + ? tunnelStatus.lastError ?? `Relay bridge to 127.0.0.1:${listenerPort} has not been validated against the current sync port.` + : tunnelStatus.lastError; const routeHealth: SyncRouteHealth = { listener: { listenerBound, From d6ba14c669f151fc8136e09b9e8b94b150f75bd6 Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Tue, 14 Jul 2026 23:06:42 -0400 Subject: [PATCH 3/4] fix(sync): make loopback ADE probe forge-resistant + close revalidation gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses three P1 review findings (Codex x2, Greptile) on the loopback port-collision hardening. All security/reliability in the sync host path. 1) Probe was not ADE-specific (Codex P1). probeAdeLoopbackListener treated any HTTP 426 as ADE, but a bare 'ws' server (a stale/foreign WebSocket process, incl. a second ADE instance) also answers plain GETs with 426 — so the 'ADE handshake' could be satisfied by any WebSocket listener. Both ADE sync server sites (syncHostService self-owned host; sharedSyncListener bindOnce) now front their WebSocketServer with an explicit http.Server whose non-upgrade 426 response carries a marker header 'x-ade-sync-loopback: 1'; the probe requires status 426 AND that marker. The WS upgrade path is unchanged (WebSocketServer constructed with { server }). NOTE: the 426 deliberately omits Upgrade/Connection headers — sending them on a non-upgrade response corrupts Node's keep-alive state machine (subsequent requests 400). 2) Refresh/re-enable skipped revalidation (Codex P1). refreshLanDiscovery and setDiscoveryEnabled(true) republished Bonjour/Tailscale without re-running the loopback check, so a shadow that appeared AFTER startup could re-advertise a stale port. Both now validateListeningPort(port,{force:true}) before (re)publishing and fail closed (skip publish, mark route unvalidated) on a shadow. 3) Ephemeral (port 0) shadow could not retry (Greptile P1). A loopback-shadowed bind(0) exhausted the single-entry attempt plan and failed instead of asking the OS for a fresh port. Port 0 now gets bounded PREFERRED_PORT_BIND_ATTEMPTS re-binds (each a fresh OS port), tracks the RESOLVED shadowed port (not literal 0), and skips the fixed-port retry delay; loopback-shadow errors are now retryable. Mirrored in the syncService self-owned startup loop. Incidental: fronting ws with an external http.Server changes close() semantics (ws.close() drains clients and does not close the http server), so listener close / candidate cleanup / host dispose now terminate clients and close the http server (closeAllConnections) to free the port deterministically. Tests: new syncLoopbackProbe.test.ts (marker required); collision tests for a markerless-426 foreign listener rejection and ephemeral-shadow retry/termination; host tests for refresh/re-enable revalidation. tsc clean; sync suites green. Co-Authored-By: Claude Opus 4.8 --- .../src/services/sync/sharedSyncListener.ts | 81 ++++++++++++-- .../src/services/sync/syncHostService.test.ts | 79 ++++++++++++++ .../src/services/sync/syncHostService.ts | 66 +++++++++-- .../sync/syncLoopbackCollision.test.ts | 103 ++++++++++++++++++ .../services/sync/syncLoopbackProbe.test.ts | 100 +++++++++++++++++ .../src/services/sync/syncLoopbackProbe.ts | 52 ++++++++- apps/ade-cli/src/services/sync/syncService.ts | 16 ++- 7 files changed, 470 insertions(+), 27 deletions(-) create mode 100644 apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts diff --git a/apps/ade-cli/src/services/sync/sharedSyncListener.ts b/apps/ade-cli/src/services/sync/sharedSyncListener.ts index f5e7b6b43..2ea02471a 100644 --- a/apps/ade-cli/src/services/sync/sharedSyncListener.ts +++ b/apps/ade-cli/src/services/sync/sharedSyncListener.ts @@ -1,3 +1,4 @@ +import http from "node:http"; import { WebSocketServer, WebSocket, type RawData } from "ws"; import type { SyncPeerMetadata } from "../../../../desktop/src/shared/types"; import { DEFAULT_SYNC_HOST_PORT } from "./syncProtocol"; @@ -5,6 +6,7 @@ import { assertAdeLoopbackListener, isLoopbackShadowedError, probeAdeLoopbackListener, + writeAdeLoopbackUpgradeResponse, type SyncLoopbackProbeResult, type SyncLoopbackValidationStatus, } from "./syncLoopbackProbe"; @@ -146,7 +148,10 @@ function isRetryableListenerBindError(error: unknown): boolean { return code === "EADDRINUSE" || code === "EACCES"; } -async function closeCandidateServer(candidateServer: WebSocketServer): Promise { +async function closeCandidateServer( + candidateServer: WebSocketServer, + candidateHttpServer: http.Server, +): Promise { for (const client of candidateServer.clients) { try { client.terminate(); @@ -154,9 +159,24 @@ async function closeCandidateServer(candidateServer: WebSocketServer): Promise((resolve) => { + if (!candidateHttpServer.listening) { + resolve(); + return; + } try { - candidateServer.close(() => resolve()); + candidateHttpServer.close(() => resolve()); + candidateHttpServer.closeAllConnections?.(); } catch { resolve(); } @@ -187,6 +207,9 @@ export function createSharedSyncListener(options: { const loopbackProbe = options.loopbackProbe ?? probeAdeLoopbackListener; let server: WebSocketServer | null = null; + // The http.Server fronting `server`. `ws` does not own/close an + // externally-supplied server, so we track it to free the port on close. + let httpServer: http.Server | null = null; let listeningPromise: Promise | null = null; let handler: SharedSyncListenerConnectionHandler | null = null; let fallbackHandler: SharedSyncListenerConnectionHandler | null = null; @@ -304,26 +327,37 @@ export function createSharedSyncListener(options: { const bindOnce = async (portCandidates: number[]): Promise => { const candidates = portCandidates.length > 0 ? portCandidates : [DEFAULT_SYNC_HOST_PORT]; + // A fixed preferred port is re-attempted so a dying listener can free it. + // An ephemeral port (0) is ALSO re-attempted, but for a different reason: + // each bind(0) yields a fresh OS-assigned port, so a loopback shadow on the + // first resolved port is escaped simply by re-binding. Both are bounded by + // PREFERRED_PORT_BIND_ATTEMPTS so a persistent shadow still terminates. const attemptPlan = candidates.flatMap((candidatePort, candidateIndex) => - candidateIndex === 0 && candidatePort !== 0 + (candidateIndex === 0 && candidatePort !== 0) || candidatePort === 0 ? Array.from({ length: PREFERRED_PORT_BIND_ATTEMPTS }, () => candidatePort) : [candidatePort], ); let lastError: unknown = null; let previousAttemptedPort: number | null = null; + // Tracks RESOLVED shadowed ports. For port 0 the literal 0 is never added + // (each re-bind resolves a different port), so a shadow on one ephemeral + // port does not short-circuit the remaining fresh-port attempts. const shadowedPorts = new Set(); for (const attemptedPort of attemptPlan) { if (closed) throw new Error("The shared sync listener has been closed."); - if (shadowedPorts.has(attemptedPort)) continue; - if (previousAttemptedPort === attemptedPort) { + if (attemptedPort !== 0 && shadowedPorts.has(attemptedPort)) continue; + // The retry delay lets a dying listener free a FIXED port; an ephemeral + // re-bind gets a fresh port immediately, so skip the delay for port 0. + if (previousAttemptedPort === attemptedPort && attemptedPort !== 0) { await new Promise((resolve) => setTimeout(resolve, PREFERRED_PORT_BIND_RETRY_DELAY_MS)); } previousAttemptedPort = attemptedPort; + const candidateHttpServer = http.createServer(writeAdeLoopbackUpgradeResponse); const candidateServer = new WebSocketServer({ - host: bindHost, - port: attemptedPort, + server: candidateHttpServer, maxPayload: maxPayloadBytes, }); + candidateHttpServer.listen(attemptedPort, bindHost); // Install the handler before the validation RTT so a LAN peer that // arrives in that narrow window is parked/owned instead of orphaned. candidateServer.on("connection", (ws, request) => { @@ -393,6 +427,7 @@ export function createSharedSyncListener(options: { candidateServer.on("error", onError); }); server = candidateServer; + httpServer = candidateHttpServer; server.on("error", (error: unknown) => { logger.warn?.("sync_listener.server_error", { error: error instanceof Error ? error.message : String(error), @@ -403,8 +438,12 @@ export function createSharedSyncListener(options: { return resolvedPort; } catch (error) { lastError = error; - await closeCandidateServer(candidateServer); - if (isLoopbackShadowedError(error)) shadowedPorts.add(attemptedPort); + await closeCandidateServer(candidateServer, candidateHttpServer); + // Record the RESOLVED port (not the literal 0) so an ephemeral shadow + // does not poison the remaining fresh-port re-binds. + if (isLoopbackShadowedError(error)) { + shadowedPorts.add(attemptedPort === 0 ? error.port : attemptedPort); + } const retryable = isRetryableListenerBindError(error) && (attemptedPort !== 0 || isLoopbackShadowedError(error)); logger.warn?.( @@ -507,24 +546,42 @@ export function createSharedSyncListener(options: { for (const entry of [...parked.values()]) { unpark(entry); try { - entry.snapshot.ws.close(); + entry.snapshot.ws.terminate(); } catch { // ignore close failures } } const current = server; + const currentHttp = httpServer; server = null; + httpServer = null; if (!current) return; + // Force-terminate every client. ws's close() for an EXTERNALLY-supplied + // http server resolves only after every client socket drains, so a single + // wedged socket would hang final shutdown; terminate side-steps that. for (const ws of current.clients) { try { - ws.close(); + ws.terminate(); } catch { // ignore close failures } } + try { + // Detach ws's listeners from the http server (external server: this does + // NOT close the http server, so we close it ourselves below). + current.close(); + } catch { + // ignore + } await new Promise((resolve) => { + if (!currentHttp || !currentHttp.listening) { + resolve(); + return; + } try { - current.close(() => resolve()); + currentHttp.close(() => resolve()); + // Force any lingering upgraded/keep-alive sockets so close() cannot hang. + currentHttp.closeAllConnections?.(); } catch { resolve(); } diff --git a/apps/ade-cli/src/services/sync/syncHostService.test.ts b/apps/ade-cli/src/services/sync/syncHostService.test.ts index 42423862a..0697f8c06 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.test.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.test.ts @@ -40,6 +40,7 @@ import { import { createBrainProjectActionsSyncHandler } from "./brainProjectActionsSyncHandler"; import { buildChangesetBatchPayload } from "./changesetPump"; import { createSharedSyncListener } from "./sharedSyncListener"; +import type { SyncLoopbackProbeResult } from "./syncLoopbackProbe"; import { createSyncPairingStore, type SyncPairingRecord } from "./syncPairingStore"; import { createSyncPinStore } from "./syncPinStore"; import { buildSyncDpopChallenge, sha256Hex } from "./syncDpop"; @@ -1798,6 +1799,84 @@ describe("createSyncHostService LAN discovery", () => { } }); + it("re-validates the loopback listener before refreshLanDiscovery and skips (re)publish on a post-startup shadow", async () => { + const { projectRoot, cleanup } = createTempProjectRoot(); + let probeOk = true; + const loopbackProbe = vi.fn(async (port: number): Promise => + probeOk + ? { ok: true, port, statusCode: 426, statusMessage: "Upgrade Required", checkedAt: new Date().toISOString(), reason: null } + : { ok: false, port, statusCode: 404, statusMessage: "Not Found", checkedAt: new Date().toISOString(), reason: "shadow appeared after startup" }); + const host = createSyncHostService({ + ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), + port: 0, + loopbackProbe, + } as unknown as Parameters[0]); + + try { + await host.waitUntilListening(); + expect(host.getLoopbackValidationStatus().loopbackAdeValidated).toBe(true); + const probeCallsAfterStartup = loopbackProbe.mock.calls.length; + const publishCallsAfterStartup = publishMock.mock.calls.length; + const spawnCallsAfterStartup = spawnMock.mock.calls.length; + const tailnetUpdatedAfterStartup = host.getTailnetDiscoveryStatus().updatedAt; + + // A foreign listener shadows the loopback route AFTER startup. + probeOk = false; + host.refreshLanDiscovery({ forceLan: true, forceTailnet: true }); + + // The refresh forces a re-probe (bypassing the validated-port short-circuit) + // and, seeing the shadow, marks the route unvalidated and skips publishing. + await vi.waitFor(() => + expect(host.getLoopbackValidationStatus().loopbackAdeValidated).toBe(false)); + expect(loopbackProbe.mock.calls.length).toBeGreaterThan(probeCallsAfterStartup); + expect(host.getLoopbackValidationStatus().reason).toMatch(/shadow appeared/); + // No new bonjour/tailnet advertisement for the stale port. + expect(publishMock.mock.calls.length).toBe(publishCallsAfterStartup); + expect(spawnMock.mock.calls.length).toBe(spawnCallsAfterStartup); + expect(host.getTailnetDiscoveryStatus().updatedAt).toBe(tailnetUpdatedAfterStartup); + } finally { + await host.dispose(); + cleanup(); + } + }); + + it("re-validates the loopback listener before setDiscoveryEnabled(true) and skips publish on a post-startup shadow", async () => { + const { projectRoot, cleanup } = createTempProjectRoot(); + let probeOk = true; + const loopbackProbe = vi.fn(async (port: number): Promise => + probeOk + ? { ok: true, port, statusCode: 426, statusMessage: "Upgrade Required", checkedAt: new Date().toISOString(), reason: null } + : { ok: false, port, statusCode: 404, statusMessage: "Not Found", checkedAt: new Date().toISOString(), reason: "shadow appeared while disabled" }); + const host = createSyncHostService({ + ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), + port: 0, + loopbackProbe, + } as unknown as Parameters[0]); + + try { + await host.waitUntilListening(); + expect(host.getLoopbackValidationStatus().loopbackAdeValidated).toBe(true); + // Turn discovery off, then let a shadow take over the loopback route. + host.setDiscoveryEnabled(false); + const probeCallsBeforeReenable = loopbackProbe.mock.calls.length; + const publishCallsBeforeReenable = publishMock.mock.calls.length; + const spawnCallsBeforeReenable = spawnMock.mock.calls.length; + + probeOk = false; + host.setDiscoveryEnabled(true); + + // Re-enabling forces a fresh loopback check; the shadow blocks (re)publish. + await vi.waitFor(() => + expect(host.getLoopbackValidationStatus().loopbackAdeValidated).toBe(false)); + expect(loopbackProbe.mock.calls.length).toBeGreaterThan(probeCallsBeforeReenable); + expect(publishMock.mock.calls.length).toBe(publishCallsBeforeReenable); + expect(spawnMock.mock.calls.length).toBe(spawnCallsBeforeReenable); + } finally { + await host.dispose(); + cleanup(); + } + }); + it("closes inbound sockets that never authenticate", async () => { const { projectRoot, cleanup } = createTempProjectRoot(); const host = createSyncHostService({ diff --git a/apps/ade-cli/src/services/sync/syncHostService.ts b/apps/ade-cli/src/services/sync/syncHostService.ts index eaaa052b4..8b9cb3144 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.ts @@ -1,4 +1,5 @@ import fs from "node:fs"; +import http from "node:http"; import { execFile, spawn, type ChildProcess } from "node:child_process"; import os from "node:os"; import path from "node:path"; @@ -148,6 +149,7 @@ import { assertAdeLoopbackListener, isLoopbackShadowedError, probeAdeLoopbackListener, + writeAdeLoopbackUpgradeResponse, type SyncLoopbackProbeResult, type SyncLoopbackValidationStatus, } from "./syncLoopbackProbe"; @@ -1966,13 +1968,24 @@ export function createSyncHostService(args: SyncHostServiceArgs) { // Self-owned listener (desktop-embedded / standalone): only created when no // shared listener is injected. The brain injects a shared listener so the // websocket — and every connected phone — survives hosted-project switches. + // + // We front the WebSocketServer with an explicit http.Server so the non-upgrade + // 426 response carries the ADE loopback marker header. `ws`'s built-in `{port}` + // server owns an un-customizable 426 handler, which a bare/foreign `ws` process + // matches exactly — the marker is what lets the loopback probe tell ADE apart. + // Passing `server` (not `port`) means the WS upgrade path still works: `ws` + // re-emits the http server's `listening`/`error` events and delegates + // `address()`, so all existing event wiring below is preserved verbatim. + const httpServer = sharedListener + ? null + : http.createServer(writeAdeLoopbackUpgradeResponse); const server = sharedListener ? null : new WebSocketServer({ - host: SYNC_HOST_BIND_HOST, - port: args.port ?? DEFAULT_SYNC_HOST_PORT, + server: httpServer!, maxPayload: SYNC_HOST_MAX_PAYLOAD_BYTES, }); + httpServer?.listen(args.port ?? DEFAULT_SYNC_HOST_PORT, SYNC_HOST_BIND_HOST); let disposed = false; let startupError: Error | null = null; @@ -2860,9 +2873,16 @@ export function createSyncHostService(args: SyncHostServiceArgs) { } }; - const validateListeningPort = async (port: number): Promise => { + const validateListeningPort = async ( + port: number, + options?: { force?: boolean }, + ): Promise => { + // Re-publish paths pass force:true so a shadow that arose AFTER startup is + // caught before we (re)advertise the port; the startup path keeps the cheap + // short-circuit once a port is validated. if ( - loopbackValidationStatus.port === port + !options?.force + && loopbackValidationStatus.port === port && loopbackValidationStatus.loopbackAdeValidated ) return; try { @@ -5438,7 +5458,13 @@ export function createSyncHostService(args: SyncHostServiceArgs) { refreshLanDiscovery(options?: { forceLan?: boolean; forceTailnet?: boolean }): void { const port = getListeningPort(); if (port != null) { - void publishValidatedDiscovery(port, options).catch((error) => { + // Re-validate the loopback listener before republishing so a post-startup + // shadow cannot re-advertise a stale port. On failure validateListeningPort + // marks the route unvalidated and throws, so we skip the publish. + void (async () => { + await validateListeningPort(port, { force: true }); + await publishValidatedDiscovery(port, options); + })().catch((error) => { args.logger.warn("sync_host.discovery_refresh_failed", { port, error: error instanceof Error ? error.message : String(error), @@ -5466,8 +5492,18 @@ export function createSyncHostService(args: SyncHostServiceArgs) { return; } if (port != null) { - publishLanDiscovery(port, { force: true }); - publishTailnetDiscovery(port, { force: true }); + // Re-enabling discovery must also re-validate the loopback listener so a + // shadow that appeared while discovery was off cannot be published. + void (async () => { + await validateListeningPort(port, { force: true }); + publishLanDiscovery(port, { force: true }); + publishTailnetDiscovery(port, { force: true }); + })().catch((error) => { + args.logger.warn("sync_host.discovery_refresh_failed", { + port, + error: error instanceof Error ? error.message : String(error), + }); + }); } }, @@ -5692,12 +5728,24 @@ export function createSyncHostService(args: SyncHostServiceArgs) { // ignore } } - if (!server.address()) { + // Graceful close frames were sent to peers above. ws's close() for an + // externally-supplied http server resolves only after every client + // socket drains — a wedged socket would hang dispose — and it does NOT + // close the http server (which owns the port). So detach ws's + // listeners, then close the http server directly, forcing any lingering + // sockets so the port frees deterministically. + try { + server.close(); + } catch { + // ignore: we free the port via the http server below + } + if (!httpServer || !httpServer.listening) { finish(); return; } try { - server.close(() => finish()); + httpServer.close(() => finish()); + httpServer.closeAllConnections?.(); } catch { finish(); } diff --git a/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts index 460674c97..9d458f084 100644 --- a/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts +++ b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts @@ -105,6 +105,30 @@ async function bindForeignLegacyListener(): Promise<{ server: http.Server; port: throw new Error("No free legacy sync port was available for the collision test."); } +async function bindForeignBare426Listener(): Promise<{ server: http.Server; port: number }> { + // A bare `ws`-style server answers plain GETs with 426 Upgrade Required but + // WITHOUT the ADE loopback marker header — exactly what the probe must reject. + for (let port = 8787; port <= 8800; port += 1) { + const server = http.createServer((_request, response) => { + const body = "Upgrade Required"; + response.writeHead(426, { + "Content-Type": "text/plain", + "Content-Length": Buffer.byteLength(body), + }); + response.end(body); + }); + try { + await listen(server, port, "127.0.0.1"); + return { server, port }; + } catch { + try { + server.close(); + } catch {} + } + } + throw new Error("No free legacy sync port was available for the bare-426 test."); +} + async function findFreeLegacyPort(): Promise { for (let port = 8787; port <= 8800; port += 1) { const server = http.createServer(); @@ -317,4 +341,83 @@ describe("sync loopback collision recovery", () => { } }, ); + + // Finding #1: a bare `ws`-style 426 (no ADE marker) must be rejected, while the + // real ADE listener — whose 426 carries the marker — passes. Status code alone + // cannot tell ADE apart from any other WebSocket process. + it("rejects a foreign bare-426 listener without the ADE marker but accepts the real ADE listener", async () => { + const foreign = await bindForeignBare426Listener(); + const adeListener = createSharedSyncListener({ bindHost: "127.0.0.1" }); + try { + const adePort = await adeListener.ensureListening([0]); + + const foreignResult = await probeAdeLoopbackListener(foreign.port); + expect(foreignResult).toMatchObject({ ok: false, statusCode: 426 }); + expect(foreignResult.reason).toMatch(/did not present the ADE loopback marker/); + + const adeResult = await probeAdeLoopbackListener(adePort); + expect(adeResult).toMatchObject({ ok: true, statusCode: 426 }); + } finally { + await adeListener.close(); + await close(foreign.server); + } + }); + + // Finding #3: an ephemeral [0] bind whose first resolved port is loopback- + // shadowed must re-bind to a fresh OS-assigned port and succeed. + it("re-binds an ephemeral [0] listener onto a fresh port when the first resolved port is shadowed", async () => { + const shadowedPorts: number[] = []; + let shadowsRemaining = 1; + const loopbackProbe = vi.fn(async (port: number): Promise => { + if (shadowsRemaining > 0) { + shadowsRemaining -= 1; + shadowedPorts.push(port); + return { + ok: false, + port, + statusCode: 404, + statusMessage: "Not Found", + checkedAt: new Date().toISOString(), + reason: "ephemeral loopback shadow", + }; + } + // Once past the injected shadow, run the REAL probe against the real ADE + // listener (which now emits the marker), proving an end-to-end fresh bind. + return await probeAdeLoopbackListener(port); + }); + const listener = createSharedSyncListener({ bindHost: "127.0.0.1", loopbackProbe }); + try { + const port = await listener.ensureListening([0]); + expect(shadowedPorts).toHaveLength(1); + expect(port).toBeGreaterThan(0); + expect(port).not.toBe(shadowedPorts[0]); + expect(listener.isListening()).toBe(true); + expect(listener.getLoopbackValidationStatus().loopbackAdeValidated).toBe(true); + } finally { + await listener.close(); + } + }, 15_000); + + // Finding #3 (bound): a persistently-shadowed ephemeral bind must still + // terminate with a failure rather than spin forever. + it("gives up an ephemeral [0] bind that is persistently loopback-shadowed", async () => { + const loopbackProbe = vi.fn(async (port: number): Promise => ({ + ok: false, + port, + statusCode: 404, + statusMessage: "Not Found", + checkedAt: new Date().toISOString(), + reason: "persistent loopback shadow", + })); + const listener = createSharedSyncListener({ bindHost: "127.0.0.1", loopbackProbe }); + try { + await expect(listener.ensureListening([0])).rejects.toThrow(/persistent loopback shadow/); + // Bounded: it must not probe forever. + expect(loopbackProbe.mock.calls.length).toBeGreaterThan(1); + expect(loopbackProbe.mock.calls.length).toBeLessThanOrEqual(16); + expect(listener.isListening()).toBe(false); + } finally { + await listener.close(); + } + }, 15_000); }); diff --git a/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts b/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts new file mode 100644 index 000000000..ce41fbaf9 --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts @@ -0,0 +1,100 @@ +import http from "node:http"; +import { afterEach, describe, expect, it } from "vitest"; +import { + assertAdeLoopbackListener, + isLoopbackShadowedError, + probeAdeLoopbackListener, + SYNC_LOOPBACK_ADE_MARKER_HEADER, + SYNC_LOOPBACK_ADE_MARKER_VALUE, + writeAdeLoopbackUpgradeResponse, +} from "./syncLoopbackProbe"; + +type RequestHandler = ( + request: http.IncomingMessage, + response: http.ServerResponse, +) => void; + +const servers: http.Server[] = []; + +async function startServer(handler: RequestHandler): Promise { + const server = http.createServer(handler); + servers.push(server); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + server.off("error", reject); + resolve(); + }); + }); + const address = server.address(); + if (typeof address !== "object" || !address) { + throw new Error("Failed to resolve loopback probe test server port."); + } + return address.port; +} + +afterEach(async () => { + while (servers.length > 0) { + const server = servers.pop()!; + await new Promise((resolve) => server.close(() => resolve())); + } +}); + +describe("probeAdeLoopbackListener ADE marker enforcement", () => { + it("accepts a 426 loopback listener that presents the ADE marker header", async () => { + // The production marker-emitting handler is the source of truth. + const port = await startServer(writeAdeLoopbackUpgradeResponse); + const result = await probeAdeLoopbackListener(port); + expect(result).toMatchObject({ + ok: true, + statusCode: 426, + reason: null, + }); + await expect(assertAdeLoopbackListener(port)).resolves.toMatchObject({ ok: true }); + }); + + it("rejects a bare 426 listener that omits the ADE marker (foreign/stale ws)", async () => { + const port = await startServer((_request, response) => { + const body = "Upgrade Required"; + response.writeHead(426, { + "Content-Type": "text/plain", + "Content-Length": Buffer.byteLength(body), + }); + response.end(body); + }); + const result = await probeAdeLoopbackListener(port); + expect(result.ok).toBe(false); + expect(result.statusCode).toBe(426); + expect(result.reason).toMatch(/did not present the ADE loopback marker/); + expect(result.reason).toContain(SYNC_LOOPBACK_ADE_MARKER_HEADER); + await expect(assertAdeLoopbackListener(port)).rejects.toSatisfy(isLoopbackShadowedError); + }); + + it("rejects a 426 listener that sends a mismatched marker value", async () => { + const port = await startServer((_request, response) => { + const body = "Upgrade Required"; + response.writeHead(426, { + "Content-Type": "text/plain", + "Content-Length": Buffer.byteLength(body), + [SYNC_LOOPBACK_ADE_MARKER_HEADER]: "totally-not-ade", + }); + response.end(body); + }); + const result = await probeAdeLoopbackListener(port); + expect(result.ok).toBe(false); + expect(result.reason).toMatch(/did not present the ADE loopback marker/); + }); + + it("rejects a non-426 foreign listener (marker present but wrong status)", async () => { + const port = await startServer((_request, response) => { + response.writeHead(404, "Not Found", { + [SYNC_LOOPBACK_ADE_MARKER_HEADER]: SYNC_LOOPBACK_ADE_MARKER_VALUE, + }); + response.end("nope"); + }); + const result = await probeAdeLoopbackListener(port); + expect(result.ok).toBe(false); + expect(result.statusCode).toBe(404); + expect(result.reason).toMatch(/Expected ADE 426 Upgrade Required/); + }); +}); diff --git a/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts b/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts index 4df5c8add..1f5e4616a 100644 --- a/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts +++ b/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts @@ -2,6 +2,46 @@ import http from "node:http"; export const SYNC_LOOPBACK_PROBE_TIMEOUT_MS = 1_500; +/** + * Marker header ADE's sync WebSocket servers stamp on the 426 Upgrade Required + * response they return for non-upgrade HTTP requests. A bare `ws` server (a + * stale/foreign WebSocket process) also answers plain GETs with 426, so the + * status code alone cannot distinguish ADE from any other WebSocket listener. + * The probe additionally requires this marker before trusting a loopback + * listener as ADE. Both production server construction sites + * (syncHostService self-owned host, sharedSyncListener bindOnce) emit it. + */ +export const SYNC_LOOPBACK_ADE_MARKER_HEADER = "x-ade-sync-loopback"; +export const SYNC_LOOPBACK_ADE_MARKER_VALUE = "1"; + +/** + * Request handler for the http.Server that fronts an ADE sync WebSocketServer. + * Non-upgrade HTTP requests get a 426 Upgrade Required carrying the ADE marker + * header so `probeAdeLoopbackListener` can distinguish an ADE listener from a + * bare/foreign `ws` process (which also answers plain GETs with a 426, but + * without the marker). The WebSocketServer, constructed with `{ server }`, + * intercepts `upgrade` requests before they reach this handler, so real sync + * clients still complete the websocket handshake unchanged. + */ +export function writeAdeLoopbackUpgradeResponse( + _request: http.IncomingMessage, + response: http.ServerResponse, +): void { + // Mirror `ws`'s built-in 426 handler exactly (Content-Type + Content-Length + // only) and add the ADE marker. Do NOT send `Connection: Upgrade` / + // `Upgrade: websocket` on this NON-upgrade response: those confuse Node's http + // server socket state machine and make every other keep-alive request return + // 400, which would intermittently defeat the probe. The probe validates the + // status code and the marker, so no extra headers are required. + const body = "Upgrade Required"; + response.writeHead(426, { + "Content-Type": "text/plain", + "Content-Length": Buffer.byteLength(body), + [SYNC_LOOPBACK_ADE_MARKER_HEADER]: SYNC_LOOPBACK_ADE_MARKER_VALUE, + }); + response.end(body); +} + export type SyncLoopbackProbeResult = { ok: boolean; port: number; @@ -64,15 +104,23 @@ export async function probeAdeLoopbackListener( response.resume(); const statusCode = response.statusCode ?? null; const statusMessage = response.statusMessage ?? null; - const ok = statusCode === 426 + const rawMarker = response.headers[SYNC_LOOPBACK_ADE_MARKER_HEADER]; + const markerValue = Array.isArray(rawMarker) ? rawMarker[0] : rawMarker; + const hasAdeMarker = markerValue === SYNC_LOOPBACK_ADE_MARKER_VALUE; + const statusOk = statusCode === 426 && (statusMessage == null || statusMessage.toLowerCase() === "upgrade required"); + // A foreign/stale bare `ws` server also returns 426, so the status is + // necessary but not sufficient — the ADE marker must be present too. + const ok = statusOk && hasAdeMarker; finish({ ok, statusCode, statusMessage, reason: ok ? null - : `Expected ADE 426 Upgrade Required on 127.0.0.1:${port}, received ${statusCode ?? "no status"}${statusMessage ? ` ${statusMessage}` : ""}.`, + : !statusOk + ? `Expected ADE 426 Upgrade Required on 127.0.0.1:${port}, received ${statusCode ?? "no status"}${statusMessage ? ` ${statusMessage}` : ""}.` + : `The listener on 127.0.0.1:${port} returned 426 but did not present the ADE loopback marker (${SYNC_LOOPBACK_ADE_MARKER_HEADER}: ${SYNC_LOOPBACK_ADE_MARKER_VALUE}); it is not an ADE sync host.`, }); }); request.setTimeout(timeoutMs, () => { diff --git a/apps/ade-cli/src/services/sync/syncService.ts b/apps/ade-cli/src/services/sync/syncService.ts index 6b5aa5a56..00142fa1f 100644 --- a/apps/ade-cli/src/services/sync/syncService.ts +++ b/apps/ade-cli/src/services/sync/syncService.ts @@ -804,16 +804,24 @@ export function createSyncService(args: SyncServiceArgs) { // and a single EADDRINUSE would silently drift the host to port+1 — // stranding paired phones that saved the old port. Re-attempt the // preferred port for a few seconds before falling back to the scan. + // The preferred (fixed) port is re-attempted so a dying listener can free + // it. An ephemeral port (0) is ALSO re-attempted, but because each bind(0) + // resolves a fresh OS-assigned port a loopback shadow on the first port is + // escaped by re-binding. Both are bounded by PREFERRED_PORT_BIND_ATTEMPTS. const attemptPlan = portCandidates.flatMap((candidatePort, candidateIndex) => - candidateIndex === 0 + candidateIndex === 0 || candidatePort === 0 ? Array.from({ length: PREFERRED_PORT_BIND_ATTEMPTS }, () => candidatePort) : [candidatePort], ); let previousAttemptedPort: number | null = null; + // Tracks RESOLVED shadowed ports; the literal 0 is never added so an + // ephemeral shadow does not short-circuit the remaining fresh-port binds. const shadowedPorts = new Set(); for (const attemptedPort of attemptPlan) { - if (shadowedPorts.has(attemptedPort)) continue; - if (previousAttemptedPort === attemptedPort) { + if (attemptedPort !== 0 && shadowedPorts.has(attemptedPort)) continue; + // The retry delay lets a dying listener free a FIXED port; an ephemeral + // re-bind gets a fresh port immediately, so skip the delay for port 0. + if (previousAttemptedPort === attemptedPort && attemptedPort !== 0) { await new Promise((resolve) => setTimeout(resolve, PREFERRED_PORT_BIND_RETRY_DELAY_MS)); } previousAttemptedPort = attemptedPort; @@ -825,7 +833,7 @@ export function createSyncService(args: SyncServiceArgs) { } catch (error) { lastError = error; if (isLoopbackShadowedError(error)) { - shadowedPorts.add(attemptedPort); + shadowedPorts.add(attemptedPort === 0 ? error.port : attemptedPort); listenerValidationHistory = { ...listenerValidationHistory, lastFailureAt: error.failedAt, From a1c307aa3389a4f4928f3e5230e3f2d0b8705ea8 Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Wed, 15 Jul 2026 00:02:28 -0400 Subject: [PATCH 4/4] fix(sync): per-instance loopback identity nonce (unforgeable) + handoff re-probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Codex P1 (require a per-listener loopback proof) and P2 (re-probe the shared listener before publishing) on PR #816. P1: the static public marker 'x-ade-sync-loopback: 1' was forgeable — a stale or second ADE process on 127.0.0.1: emits the same constant, so the probe could not tell OUR listener from another ADE instance shadowing the port. Now each listener generates a fresh 128-bit nonce at construction (generateLoopbackNonce), stamps it as the 426 identity header, and every IN-PROCESS validator compares the probed value against the exact nonce it holds. A shadow is a different process and cannot present this instance's nonce. Threaded through: sharedSyncListener (one nonce/instance, getExpectedLoopbackNonce), self-owned host server (reuses the shared listener's nonce when injected, else its own), the syncService startup loop (via host construction), and the relay tunnel client (nonce provider wired in bootstrap.ts). The relay path fails closed when the identity is unavailable, and relayBridgeValidated now also requires the nonce to still match the current listener (a restarted listener with a new nonce invalidates a stale bridge). P2: SharedSyncListener.revalidateLoopback() force re-probes the current bound port with the instance nonce; the shared-listener branch of waitUntilListening now calls it before republishing LAN/Tailscale discovery, so a shadow that appears AFTER the original bind is caught on a project-host handoff (throws before publish; fails closed). Tests: syncLoopbackProbe.test.ts proves a different-instance ADE listener (and the legacy static '1', and a markerless 426) are rejected while the own-identity passes; collision/host tests cover the handoff re-probe and thread the nonce through the existing shadow/retry/refresh fixtures. tsc clean; sync suites green. Co-Authored-By: Claude Opus 4.8 --- apps/ade-cli/src/bootstrap.ts | 2 + .../src/services/sync/sharedSyncListener.ts | 72 +++++++++++----- .../src/services/sync/syncHostService.test.ts | 82 +++++++++++++++++-- .../src/services/sync/syncHostService.ts | 24 +++++- .../sync/syncLoopbackCollision.test.ts | 31 +++++-- .../services/sync/syncLoopbackProbe.test.ts | 66 ++++++++++----- .../src/services/sync/syncLoopbackProbe.ts | 52 +++++++----- apps/ade-cli/src/services/sync/syncService.ts | 2 +- .../sync/syncTunnelClientService.test.ts | 21 +++-- .../services/sync/syncTunnelClientService.ts | 27 +++++- 10 files changed, 290 insertions(+), 89 deletions(-) diff --git a/apps/ade-cli/src/bootstrap.ts b/apps/ade-cli/src/bootstrap.ts index 758568087..ea234b71a 100644 --- a/apps/ade-cli/src/bootstrap.ts +++ b/apps/ade-cli/src/bootstrap.ts @@ -1515,6 +1515,8 @@ export async function createAdeRuntime(args: { logger, configStore: cloudRelayStore, getSyncPort: () => resolvedArgs.syncRuntime?.sharedSyncListener?.getPort() ?? null, + getExpectedLoopbackNonce: () => + resolvedArgs.syncRuntime?.sharedSyncListener?.getExpectedLoopbackNonce() ?? null, })); // Only the runtime that actually hosts phone sync (owns the brain-level // shared listener) may register the relay tunnel. The relay DO keeps ONE diff --git a/apps/ade-cli/src/services/sync/sharedSyncListener.ts b/apps/ade-cli/src/services/sync/sharedSyncListener.ts index 2ea02471a..6f7d29177 100644 --- a/apps/ade-cli/src/services/sync/sharedSyncListener.ts +++ b/apps/ade-cli/src/services/sync/sharedSyncListener.ts @@ -4,6 +4,7 @@ import type { SyncPeerMetadata } from "../../../../desktop/src/shared/types"; import { DEFAULT_SYNC_HOST_PORT } from "./syncProtocol"; import { assertAdeLoopbackListener, + generateLoopbackNonce, isLoopbackShadowedError, probeAdeLoopbackListener, writeAdeLoopbackUpgradeResponse, @@ -112,7 +113,10 @@ export type SharedSyncListener = { ensureListening(portCandidates: number[]): Promise; getPort(): number | null; isListening(): boolean; + getExpectedLoopbackNonce(): string; getLoopbackValidationStatus(): SyncLoopbackValidationStatus; + /** Force-check that loopback still reaches this exact listener instance. */ + revalidateLoopback(): Promise; /** * Install the connection handler for NEW sockets. Returns a detach function * that only clears the handler if it has not been superseded by a newer @@ -198,13 +202,17 @@ export function createSharedSyncListener(options: { bindHost?: string; maxPayloadBytes?: number; parkedPeerGraceMs?: number; - loopbackProbe?: (port: number) => Promise; + loopbackProbe?: (port: number, expectedNonce: string) => Promise; } = {}): SharedSyncListener { const logger = options.logger ?? {}; const bindHost = options.bindHost ?? SYNC_HOST_BIND_HOST; const maxPayloadBytes = options.maxPayloadBytes ?? SYNC_HOST_MAX_PAYLOAD_BYTES; const parkedPeerGraceMs = Math.max(50, Math.floor(options.parkedPeerGraceMs ?? DEFAULT_PARKED_PEER_GRACE_MS)); const loopbackProbe = options.loopbackProbe ?? probeAdeLoopbackListener; + // One identity per listener instance. Every candidate bind for this + // instance emits the same nonce, and only in-process validators receive the + // expected value they must compare against. + const expectedLoopbackNonce = generateLoopbackNonce(); let server: WebSocketServer | null = null; // The http.Server fronting `server`. `ws` does not own/close an @@ -224,6 +232,34 @@ export function createSharedSyncListener(options: { }; const parked = new Map(); + const validateLoopback = async (port: number): Promise => { + try { + const result = await assertAdeLoopbackListener( + port, + expectedLoopbackNonce, + loopbackProbe, + ); + loopbackValidationStatus = { + port, + loopbackAdeValidated: true, + lastFailureAt: loopbackValidationStatus.lastFailureAt, + reason: null, + lastSuccessAt: result.checkedAt, + }; + } catch (error) { + if (isLoopbackShadowedError(error)) { + loopbackValidationStatus = { + port, + loopbackAdeValidated: false, + lastFailureAt: error.failedAt, + reason: error.message, + lastSuccessAt: loopbackValidationStatus.lastSuccessAt, + }; + } + throw error; + } + }; + const unpark = (entry: ParkedEntry): void => { clearTimeout(entry.expireTimer); entry.snapshot.ws.off("message", entry.onMessage); @@ -352,7 +388,9 @@ export function createSharedSyncListener(options: { await new Promise((resolve) => setTimeout(resolve, PREFERRED_PORT_BIND_RETRY_DELAY_MS)); } previousAttemptedPort = attemptedPort; - const candidateHttpServer = http.createServer(writeAdeLoopbackUpgradeResponse); + const candidateHttpServer = http.createServer((request, response) => { + writeAdeLoopbackUpgradeResponse(request, response, expectedLoopbackNonce); + }); const candidateServer = new WebSocketServer({ server: candidateHttpServer, maxPayload: maxPayloadBytes, @@ -388,27 +426,11 @@ export function createSharedSyncListener(options: { const address = candidateServer.address(); const port = typeof address === "object" && address ? address.port : attemptedPort; try { - const result = await assertAdeLoopbackListener(port, loopbackProbe); + await validateLoopback(port); cleanup(); - loopbackValidationStatus = { - port, - loopbackAdeValidated: true, - lastFailureAt: loopbackValidationStatus.lastFailureAt, - reason: null, - lastSuccessAt: result.checkedAt, - }; resolve(port); } catch (error) { cleanup(); - if (isLoopbackShadowedError(error)) { - loopbackValidationStatus = { - port, - loopbackAdeValidated: false, - lastFailureAt: error.failedAt, - reason: error.message, - lastSuccessAt: loopbackValidationStatus.lastSuccessAt, - }; - } reject(error instanceof Error ? error : new Error(String(error))); } }; @@ -486,10 +508,22 @@ export function createSharedSyncListener(options: { return server?.address() != null; }, + getExpectedLoopbackNonce(): string { + return expectedLoopbackNonce; + }, + getLoopbackValidationStatus(): SyncLoopbackValidationStatus { return { ...loopbackValidationStatus }; }, + async revalidateLoopback(): Promise { + const address = server?.address(); + if (typeof address !== "object" || !address) { + throw new Error("The shared sync listener is not listening."); + } + await validateLoopback(address.port); + }, + setConnectionHandler(nextHandler: SharedSyncListenerConnectionHandler): () => void { handler = nextHandler; fallbackSuppressedUntilMs = 0; diff --git a/apps/ade-cli/src/services/sync/syncHostService.test.ts b/apps/ade-cli/src/services/sync/syncHostService.test.ts index 0697f8c06..e448108ee 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.test.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.test.ts @@ -1779,6 +1779,7 @@ describe("createSyncHostService LAN discovery", () => { port, statusCode: 404, statusMessage: "Not Found", + markerValue: null, checkedAt: new Date().toISOString(), reason: "foreign loopback listener", }), @@ -1802,10 +1803,13 @@ describe("createSyncHostService LAN discovery", () => { it("re-validates the loopback listener before refreshLanDiscovery and skips (re)publish on a post-startup shadow", async () => { const { projectRoot, cleanup } = createTempProjectRoot(); let probeOk = true; - const loopbackProbe = vi.fn(async (port: number): Promise => + const loopbackProbe = vi.fn(async ( + port: number, + expectedNonce: string, + ): Promise => probeOk - ? { ok: true, port, statusCode: 426, statusMessage: "Upgrade Required", checkedAt: new Date().toISOString(), reason: null } - : { ok: false, port, statusCode: 404, statusMessage: "Not Found", checkedAt: new Date().toISOString(), reason: "shadow appeared after startup" }); + ? { ok: true, port, statusCode: 426, statusMessage: "Upgrade Required", markerValue: expectedNonce, checkedAt: new Date().toISOString(), reason: null } + : { ok: false, port, statusCode: 404, statusMessage: "Not Found", markerValue: null, checkedAt: new Date().toISOString(), reason: "shadow appeared after startup" }); const host = createSyncHostService({ ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), port: 0, @@ -1843,10 +1847,13 @@ describe("createSyncHostService LAN discovery", () => { it("re-validates the loopback listener before setDiscoveryEnabled(true) and skips publish on a post-startup shadow", async () => { const { projectRoot, cleanup } = createTempProjectRoot(); let probeOk = true; - const loopbackProbe = vi.fn(async (port: number): Promise => + const loopbackProbe = vi.fn(async ( + port: number, + expectedNonce: string, + ): Promise => probeOk - ? { ok: true, port, statusCode: 426, statusMessage: "Upgrade Required", checkedAt: new Date().toISOString(), reason: null } - : { ok: false, port, statusCode: 404, statusMessage: "Not Found", checkedAt: new Date().toISOString(), reason: "shadow appeared while disabled" }); + ? { ok: true, port, statusCode: 426, statusMessage: "Upgrade Required", markerValue: expectedNonce, checkedAt: new Date().toISOString(), reason: null } + : { ok: false, port, statusCode: 404, statusMessage: "Not Found", markerValue: null, checkedAt: new Date().toISOString(), reason: "shadow appeared while disabled" }); const host = createSyncHostService({ ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), port: 0, @@ -3186,6 +3193,69 @@ describe("shared listener waitUntilListening ADE-validation gate", () => { publishMock.mockImplementation(() => ({ on: vi.fn(), stop: vi.fn() })); }); + it("force re-probes a shared listener at handoff and blocks discovery for a post-bind shadow", async () => { + const { projectRoot, cleanup } = createTempProjectRoot(); + let probeOk = true; + const loopbackProbe = vi.fn(async ( + port: number, + expectedNonce: string, + ): Promise => probeOk + ? { + ok: true, + port, + statusCode: 426, + statusMessage: "Upgrade Required", + markerValue: expectedNonce, + checkedAt: new Date().toISOString(), + reason: null, + } + : { + ok: false, + port, + statusCode: 426, + statusMessage: "Upgrade Required", + markerValue: "post-bind-shadow", + checkedAt: new Date().toISOString(), + reason: "post-bind shadow presented a different loopback identity", + }); + const listener = createSharedSyncListener({ + bindHost: "127.0.0.1", + loopbackProbe, + }); + const boundPort = await listener.ensureListening([0]); + const bindProbeCalls = loopbackProbe.mock.calls.length; + probeOk = false; + const host = createSyncHostService({ + ...createHostArgs(projectRoot, [createDiscoveryProject({ id: "project-1" })]), + sharedListener: listener, + } as unknown as Parameters[0]); + + try { + await expect(host.waitUntilListening()).rejects.toThrow(/post-bind shadow/); + expect(loopbackProbe.mock.calls.length).toBeGreaterThan(bindProbeCalls); + expect(loopbackProbe).toHaveBeenLastCalledWith( + boundPort, + listener.getExpectedLoopbackNonce(), + ); + expect(listener.getLoopbackValidationStatus()).toMatchObject({ + port: boundPort, + loopbackAdeValidated: false, + reason: expect.stringMatching(/post-bind shadow/), + }); + expect(host.getLoopbackValidationStatus()).toMatchObject({ + port: boundPort, + loopbackAdeValidated: false, + reason: expect.stringMatching(/post-bind shadow/), + }); + expect(publishMock).not.toHaveBeenCalled(); + expect(spawnMock).not.toHaveBeenCalled(); + } finally { + await host.dispose(); + await listener.close(); + cleanup(); + } + }); + it("throws before discovery when the shared listener loopback is not ADE-validated", async () => { const { projectRoot, cleanup } = createTempProjectRoot(); const listener = createSharedSyncListener({ bindHost: "127.0.0.1" }); diff --git a/apps/ade-cli/src/services/sync/syncHostService.ts b/apps/ade-cli/src/services/sync/syncHostService.ts index 8b9cb3144..89b949043 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.ts @@ -147,6 +147,7 @@ import { } from "./sharedSyncListener"; import { assertAdeLoopbackListener, + generateLoopbackNonce, isLoopbackShadowedError, probeAdeLoopbackListener, writeAdeLoopbackUpgradeResponse, @@ -720,7 +721,7 @@ type SyncHostServiceArgs = { */ getCloudRelayWssUrl?: () => string | null; /** Test seam; production always uses the HTTP 426 loopback probe. */ - loopbackProbe?: (port: number) => Promise; + loopbackProbe?: (port: number, expectedNonce: string) => Promise; }; function sanitizeRemoteAddress(remoteAddress: string | null | undefined): string | null { @@ -1965,6 +1966,8 @@ export function createSyncHostService(args: SyncHostServiceArgs) { } }; const sharedListener = args.sharedListener ?? null; + const expectedLoopbackNonce = sharedListener?.getExpectedLoopbackNonce() + ?? generateLoopbackNonce(); // Self-owned listener (desktop-embedded / standalone): only created when no // shared listener is injected. The brain injects a shared listener so the // websocket — and every connected phone — survives hosted-project switches. @@ -1978,7 +1981,9 @@ export function createSyncHostService(args: SyncHostServiceArgs) { // `address()`, so all existing event wiring below is preserved verbatim. const httpServer = sharedListener ? null - : http.createServer(writeAdeLoopbackUpgradeResponse); + : http.createServer((request, response) => { + writeAdeLoopbackUpgradeResponse(request, response, expectedLoopbackNonce); + }); const server = sharedListener ? null : new WebSocketServer({ @@ -2886,7 +2891,11 @@ export function createSyncHostService(args: SyncHostServiceArgs) { && loopbackValidationStatus.loopbackAdeValidated ) return; try { - const result = await assertAdeLoopbackListener(port, loopbackProbe); + const result = await assertAdeLoopbackListener( + port, + expectedLoopbackNonce, + loopbackProbe, + ); loopbackValidationStatus = { port, loopbackAdeValidated: true, @@ -5392,7 +5401,14 @@ export function createSyncHostService(args: SyncHostServiceArgs) { // ensureListening is idempotent and returns the existing port. const port = sharedListener!.getPort() ?? await sharedListener!.ensureListening([args.port ?? DEFAULT_SYNC_HOST_PORT]); - loopbackValidationStatus = sharedListener!.getLoopbackValidationStatus(); + try { + // A project-host handoff may happen long after the listener's bind + // probe. Force a fresh identity check before this host republishes + // LAN/Tailscale discovery for the shared port. + await sharedListener!.revalidateLoopback(); + } finally { + loopbackValidationStatus = sharedListener!.getLoopbackValidationStatus(); + } if (!loopbackValidationStatus.loopbackAdeValidated || loopbackValidationStatus.port !== port) { throw new Error(`The shared sync listener on 127.0.0.1:${port} was not ADE-validated.`); } diff --git a/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts index 9d458f084..b3ffc982b 100644 --- a/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts +++ b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts @@ -196,11 +196,17 @@ describe("sync loopback collision recovery", () => { }); expect(publishedPorts()).not.toContain(foreign.port); expect(new Set(publishedPorts())).toEqual(new Set([resolvedPort!])); - await expect(probeAdeLoopbackListener(foreign.port)).resolves.toMatchObject({ + await expect(probeAdeLoopbackListener( + foreign.port, + listener.getExpectedLoopbackNonce(), + )).resolves.toMatchObject({ ok: false, statusCode: 404, }); - await expect(probeAdeLoopbackListener(resolvedPort!)).resolves.toMatchObject({ + await expect(probeAdeLoopbackListener( + resolvedPort!, + listener.getExpectedLoopbackNonce(), + )).resolves.toMatchObject({ ok: true, statusCode: 426, }); @@ -229,12 +235,12 @@ describe("sync loopback collision recovery", () => { const firstProbe = new Promise((resolve) => { releaseFirstProbe = resolve; }); - const loopbackProbe = vi.fn(async (port: number) => { + const loopbackProbe = vi.fn(async (port: number, expectedNonce: string) => { if (firstProbePort == null) { firstProbePort = port; return await firstProbe; } - return await probeAdeLoopbackListener(port); + return await probeAdeLoopbackListener(port, expectedNonce); }); const listener = createSharedSyncListener({ bindHost: "127.0.0.1", loopbackProbe }); const service = createService(db, projectRoot, { sharedSyncListener: listener }); @@ -251,6 +257,7 @@ describe("sync loopback collision recovery", () => { port: firstProbePort!, statusCode: 404, statusMessage: "Not Found", + markerValue: null, checkedAt: new Date().toISOString(), reason: "foreign loopback listener", }); @@ -351,11 +358,12 @@ describe("sync loopback collision recovery", () => { try { const adePort = await adeListener.ensureListening([0]); - const foreignResult = await probeAdeLoopbackListener(foreign.port); + const expectedNonce = adeListener.getExpectedLoopbackNonce(); + const foreignResult = await probeAdeLoopbackListener(foreign.port, expectedNonce); expect(foreignResult).toMatchObject({ ok: false, statusCode: 426 }); - expect(foreignResult.reason).toMatch(/did not present the ADE loopback marker/); + expect(foreignResult.reason).toMatch(/did not present a loopback identity/); - const adeResult = await probeAdeLoopbackListener(adePort); + const adeResult = await probeAdeLoopbackListener(adePort, expectedNonce); expect(adeResult).toMatchObject({ ok: true, statusCode: 426 }); } finally { await adeListener.close(); @@ -368,7 +376,10 @@ describe("sync loopback collision recovery", () => { it("re-binds an ephemeral [0] listener onto a fresh port when the first resolved port is shadowed", async () => { const shadowedPorts: number[] = []; let shadowsRemaining = 1; - const loopbackProbe = vi.fn(async (port: number): Promise => { + const loopbackProbe = vi.fn(async ( + port: number, + expectedNonce: string, + ): Promise => { if (shadowsRemaining > 0) { shadowsRemaining -= 1; shadowedPorts.push(port); @@ -377,13 +388,14 @@ describe("sync loopback collision recovery", () => { port, statusCode: 404, statusMessage: "Not Found", + markerValue: null, checkedAt: new Date().toISOString(), reason: "ephemeral loopback shadow", }; } // Once past the injected shadow, run the REAL probe against the real ADE // listener (which now emits the marker), proving an end-to-end fresh bind. - return await probeAdeLoopbackListener(port); + return await probeAdeLoopbackListener(port, expectedNonce); }); const listener = createSharedSyncListener({ bindHost: "127.0.0.1", loopbackProbe }); try { @@ -406,6 +418,7 @@ describe("sync loopback collision recovery", () => { port, statusCode: 404, statusMessage: "Not Found", + markerValue: null, checkedAt: new Date().toISOString(), reason: "persistent loopback shadow", })); diff --git a/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts b/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts index ce41fbaf9..26504e272 100644 --- a/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts +++ b/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts @@ -2,10 +2,10 @@ import http from "node:http"; import { afterEach, describe, expect, it } from "vitest"; import { assertAdeLoopbackListener, + generateLoopbackNonce, isLoopbackShadowedError, probeAdeLoopbackListener, SYNC_LOOPBACK_ADE_MARKER_HEADER, - SYNC_LOOPBACK_ADE_MARKER_VALUE, writeAdeLoopbackUpgradeResponse, } from "./syncLoopbackProbe"; @@ -40,20 +40,41 @@ afterEach(async () => { } }); -describe("probeAdeLoopbackListener ADE marker enforcement", () => { - it("accepts a 426 loopback listener that presents the ADE marker header", async () => { - // The production marker-emitting handler is the source of truth. - const port = await startServer(writeAdeLoopbackUpgradeResponse); - const result = await probeAdeLoopbackListener(port); - expect(result).toMatchObject({ +describe("probeAdeLoopbackListener per-instance identity enforcement", () => { + it("accepts its own listener identity and rejects a different ADE listener instance", async () => { + const nonceA = generateLoopbackNonce(); + const nonceB = generateLoopbackNonce(); + expect(nonceA).toMatch(/^[a-f0-9]{32}$/); + expect(nonceB).not.toBe(nonceA); + const portA = await startServer((request, response) => { + writeAdeLoopbackUpgradeResponse(request, response, nonceA); + }); + const portB = await startServer((request, response) => { + writeAdeLoopbackUpgradeResponse(request, response, nonceB); + }); + + const ownResult = await probeAdeLoopbackListener(portA, nonceA); + expect(ownResult).toMatchObject({ ok: true, statusCode: 426, + markerValue: nonceA, reason: null, }); - await expect(assertAdeLoopbackListener(port)).resolves.toMatchObject({ ok: true }); + await expect(assertAdeLoopbackListener(portA, nonceA)).resolves.toMatchObject({ ok: true }); + + const otherInstanceResult = await probeAdeLoopbackListener(portB, nonceA); + expect(otherInstanceResult).toMatchObject({ + ok: false, + statusCode: 426, + markerValue: nonceB, + }); + expect(otherInstanceResult.reason).toMatch(/different loopback identity/); + await expect(assertAdeLoopbackListener(portB, nonceA)) + .rejects.toSatisfy(isLoopbackShadowedError); }); - it("rejects a bare 426 listener that omits the ADE marker (foreign/stale ws)", async () => { + it("rejects a bare 426 listener that omits the loopback identity", async () => { + const expectedNonce = generateLoopbackNonce(); const port = await startServer((_request, response) => { const body = "Upgrade Required"; response.writeHead(426, { @@ -62,39 +83,46 @@ describe("probeAdeLoopbackListener ADE marker enforcement", () => { }); response.end(body); }); - const result = await probeAdeLoopbackListener(port); + const result = await probeAdeLoopbackListener(port, expectedNonce); expect(result.ok).toBe(false); expect(result.statusCode).toBe(426); - expect(result.reason).toMatch(/did not present the ADE loopback marker/); + expect(result.markerValue).toBeNull(); + expect(result.reason).toMatch(/did not present a loopback identity/); expect(result.reason).toContain(SYNC_LOOPBACK_ADE_MARKER_HEADER); - await expect(assertAdeLoopbackListener(port)).rejects.toSatisfy(isLoopbackShadowedError); + await expect(assertAdeLoopbackListener(port, expectedNonce)) + .rejects.toSatisfy(isLoopbackShadowedError); }); - it("rejects a 426 listener that sends a mismatched marker value", async () => { + it("rejects the forgeable legacy static marker value", async () => { + const expectedNonce = generateLoopbackNonce(); const port = await startServer((_request, response) => { const body = "Upgrade Required"; response.writeHead(426, { "Content-Type": "text/plain", "Content-Length": Buffer.byteLength(body), - [SYNC_LOOPBACK_ADE_MARKER_HEADER]: "totally-not-ade", + [SYNC_LOOPBACK_ADE_MARKER_HEADER]: "1", }); response.end(body); }); - const result = await probeAdeLoopbackListener(port); - expect(result.ok).toBe(false); - expect(result.reason).toMatch(/did not present the ADE loopback marker/); + const result = await probeAdeLoopbackListener(port, expectedNonce); + expect(result).toMatchObject({ ok: false, statusCode: 426, markerValue: "1" }); + expect(result.reason).toMatch(/different loopback identity/); + await expect(assertAdeLoopbackListener(port, expectedNonce)) + .rejects.toSatisfy(isLoopbackShadowedError); }); it("rejects a non-426 foreign listener (marker present but wrong status)", async () => { + const expectedNonce = generateLoopbackNonce(); const port = await startServer((_request, response) => { response.writeHead(404, "Not Found", { - [SYNC_LOOPBACK_ADE_MARKER_HEADER]: SYNC_LOOPBACK_ADE_MARKER_VALUE, + [SYNC_LOOPBACK_ADE_MARKER_HEADER]: expectedNonce, }); response.end("nope"); }); - const result = await probeAdeLoopbackListener(port); + const result = await probeAdeLoopbackListener(port, expectedNonce); expect(result.ok).toBe(false); expect(result.statusCode).toBe(404); + expect(result.markerValue).toBe(expectedNonce); expect(result.reason).toMatch(/Expected ADE 426 Upgrade Required/); }); }); diff --git a/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts b/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts index 1f5e4616a..fb81de4e7 100644 --- a/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts +++ b/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts @@ -1,18 +1,20 @@ +import crypto from "node:crypto"; import http from "node:http"; export const SYNC_LOOPBACK_PROBE_TIMEOUT_MS = 1_500; /** - * Marker header ADE's sync WebSocket servers stamp on the 426 Upgrade Required - * response they return for non-upgrade HTTP requests. A bare `ws` server (a - * stale/foreign WebSocket process) also answers plain GETs with 426, so the - * status code alone cannot distinguish ADE from any other WebSocket listener. - * The probe additionally requires this marker before trusting a loopback - * listener as ADE. Both production server construction sites - * (syncHostService self-owned host, sharedSyncListener bindOnce) emit it. + * Identity header ADE's sync WebSocket servers stamp on the 426 Upgrade + * Required response they return for non-upgrade HTTP requests. Its value is a + * fresh per-listener nonce known to in-process validators. A bare `ws` server + * (or another ADE process) can also answer plain GETs with 426, but cannot + * present this listener instance's expected identity. */ export const SYNC_LOOPBACK_ADE_MARKER_HEADER = "x-ade-sync-loopback"; -export const SYNC_LOOPBACK_ADE_MARKER_VALUE = "1"; + +export function generateLoopbackNonce(): string { + return crypto.randomBytes(16).toString("hex"); +} /** * Request handler for the http.Server that fronts an ADE sync WebSocketServer. @@ -26,6 +28,7 @@ export const SYNC_LOOPBACK_ADE_MARKER_VALUE = "1"; export function writeAdeLoopbackUpgradeResponse( _request: http.IncomingMessage, response: http.ServerResponse, + nonce: string, ): void { // Mirror `ws`'s built-in 426 handler exactly (Content-Type + Content-Length // only) and add the ADE marker. Do NOT send `Connection: Upgrade` / @@ -37,7 +40,7 @@ export function writeAdeLoopbackUpgradeResponse( response.writeHead(426, { "Content-Type": "text/plain", "Content-Length": Buffer.byteLength(body), - [SYNC_LOOPBACK_ADE_MARKER_HEADER]: SYNC_LOOPBACK_ADE_MARKER_VALUE, + [SYNC_LOOPBACK_ADE_MARKER_HEADER]: nonce, }); response.end(body); } @@ -47,6 +50,7 @@ export type SyncLoopbackProbeResult = { port: number; statusCode: number | null; statusMessage: string | null; + markerValue: string | null; checkedAt: string; reason: string | null; }; @@ -85,6 +89,7 @@ export function isLoopbackShadowedError(error: unknown): error is LoopbackShadow */ export async function probeAdeLoopbackListener( port: number, + expectedNonce: string, timeoutMs = SYNC_LOOPBACK_PROBE_TIMEOUT_MS, ): Promise { const checkedAt = new Date().toISOString(); @@ -105,22 +110,27 @@ export async function probeAdeLoopbackListener( const statusCode = response.statusCode ?? null; const statusMessage = response.statusMessage ?? null; const rawMarker = response.headers[SYNC_LOOPBACK_ADE_MARKER_HEADER]; - const markerValue = Array.isArray(rawMarker) ? rawMarker[0] : rawMarker; - const hasAdeMarker = markerValue === SYNC_LOOPBACK_ADE_MARKER_VALUE; - const statusOk = statusCode === 426 - && (statusMessage == null || statusMessage.toLowerCase() === "upgrade required"); - // A foreign/stale bare `ws` server also returns 426, so the status is - // necessary but not sufficient — the ADE marker must be present too. - const ok = statusOk && hasAdeMarker; + const observedMarker = Array.isArray(rawMarker) ? rawMarker[0] : rawMarker; + const markerValue = typeof observedMarker === "string" ? observedMarker : null; + const hasExpectedNonce = expectedNonce.length > 0 && markerValue === expectedNonce; + // A foreign/stale bare `ws` server and another ADE instance can both + // return 426. Only this process knows the identity it expects from the + // listener instance it just constructed. + const ok = statusCode === 426 && hasExpectedNonce; finish({ ok, statusCode, statusMessage, + markerValue, reason: ok ? null - : !statusOk + : statusCode !== 426 ? `Expected ADE 426 Upgrade Required on 127.0.0.1:${port}, received ${statusCode ?? "no status"}${statusMessage ? ` ${statusMessage}` : ""}.` - : `The listener on 127.0.0.1:${port} returned 426 but did not present the ADE loopback marker (${SYNC_LOOPBACK_ADE_MARKER_HEADER}: ${SYNC_LOOPBACK_ADE_MARKER_VALUE}); it is not an ADE sync host.`, + : markerValue == null || markerValue.length === 0 + ? `The listener on 127.0.0.1:${port} returned 426 but did not present a loopback identity in ${SYNC_LOOPBACK_ADE_MARKER_HEADER}.` + : expectedNonce.length === 0 + ? `The listener on 127.0.0.1:${port} cannot be ADE-validated because the expected loopback identity is empty.` + : `The listener on 127.0.0.1:${port} returned 426 but presented a different loopback identity — another process owns 127.0.0.1:${port}.`, }); }); request.setTimeout(timeoutMs, () => { @@ -131,6 +141,7 @@ export async function probeAdeLoopbackListener( ok: false, statusCode: null, statusMessage: null, + markerValue: null, reason: `ADE loopback probe failed on 127.0.0.1:${port}: ${error.message}`, }); }); @@ -139,9 +150,10 @@ export async function probeAdeLoopbackListener( export async function assertAdeLoopbackListener( port: number, - probe: (port: number) => Promise = probeAdeLoopbackListener, + expectedNonce: string, + probe: (port: number, expectedNonce: string) => Promise = probeAdeLoopbackListener, ): Promise { - const result = await probe(port); + const result = await probe(port, expectedNonce); if (!result.ok) { throw new LoopbackShadowedError( port, diff --git a/apps/ade-cli/src/services/sync/syncService.ts b/apps/ade-cli/src/services/sync/syncService.ts index 00142fa1f..80eb61e6b 100644 --- a/apps/ade-cli/src/services/sync/syncService.ts +++ b/apps/ade-cli/src/services/sync/syncService.ts @@ -188,7 +188,7 @@ type SyncServiceArgs = { */ dispatchDeeplinkUrl?: (url: string) => Promise<{ ok: boolean; message?: string }>; /** Test seam for self-owned host startup; production uses the HTTP 426 probe. */ - loopbackProbe?: (port: number) => Promise; + loopbackProbe?: (port: number, expectedNonce: string) => Promise; }; const DRAFT_FILE = "sync-peer-draft.json"; diff --git a/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts b/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts index 7ad9270dd..6efd5e749 100644 --- a/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts +++ b/apps/ade-cli/src/services/sync/syncTunnelClientService.test.ts @@ -91,17 +91,21 @@ describe("createSyncTunnelClientService", () => { }); const originalFetch = globalThis.fetch; globalThis.fetch = async () => new Response(null, { status: 204 }); + const expectedNonce = "c".repeat(32); + const loopbackProbe = vi.fn(async (port: number, receivedNonce: string) => ({ + ok: false, + port, + statusCode: 426, + statusMessage: "Upgrade Required", + markerValue: "d".repeat(32), + checkedAt: new Date().toISOString(), + reason: `foreign listener does not match ${receivedNonce}`, + })); const service = createSyncTunnelClientService({ getSyncPort: () => 8787, + getExpectedLoopbackNonce: () => expectedNonce, configStore: fakeStore(true, `http://127.0.0.1:${relayPort}`), - loopbackProbe: async (port) => ({ - ok: false, - port, - statusCode: 404, - statusMessage: "Not Found", - checkedAt: new Date().toISOString(), - reason: "foreign listener returned 404", - }), + loopbackProbe, }); try { @@ -111,6 +115,7 @@ describe("createSyncTunnelClientService", () => { }); expect(connections).toHaveLength(1); expect(connections[0]).toContain(`/host/${"a".repeat(32)}`); + expect(loopbackProbe).toHaveBeenCalledWith(8787, expectedNonce); expect(service.getStatus()).toMatchObject({ connected: true, activeTunnels: 0, diff --git a/apps/ade-cli/src/services/sync/syncTunnelClientService.ts b/apps/ade-cli/src/services/sync/syncTunnelClientService.ts index 4d9056203..6b8d0bfeb 100644 --- a/apps/ade-cli/src/services/sync/syncTunnelClientService.ts +++ b/apps/ade-cli/src/services/sync/syncTunnelClientService.ts @@ -46,11 +46,13 @@ type SyncTunnelClientArgs = { logger?: Logger; /** Local ADE sync WebSocket server port, or null when the host isn't up. */ getSyncPort: () => number | null; + /** Expected identity of the active in-process sync listener. */ + getExpectedLoopbackNonce?: () => string | null; configStore: SyncCloudRelayStore; /** Overrides the identity from configStore (e.g. a shared machine store). */ machineIdentity?: () => MachineIdentity | null; /** Test seam; production always uses the HTTP 426 loopback probe. */ - loopbackProbe?: (port: number) => Promise; + loopbackProbe?: (port: number, expectedNonce: string) => Promise; }; const BACKOFF_BASE_MS = 1_000; @@ -107,6 +109,7 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT let connected = false; let lastError: string | null = null; let validatedPort: number | null = null; + let validatedLoopbackNonce: string | null = null; let lastFailureAt: string | null = null; let lastSuccessAt: string | null = null; let claimed = false; @@ -210,13 +213,27 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT log.warn?.("sync_tunnel.no_sync_port", { connectionId }); return; } + const expectedLoopbackNonce = args.getExpectedLoopbackNonce?.() ?? null; + if (!expectedLoopbackNonce) { + validatedPort = null; + validatedLoopbackNonce = null; + recordFailure("Relay bridge refused because the ADE sync listener identity is unavailable."); + log.warn?.("sync_tunnel.no_loopback_identity", { connectionId, port }); + return; + } try { - const result = await assertAdeLoopbackListener(port, loopbackProbe); + const result = await assertAdeLoopbackListener( + port, + expectedLoopbackNonce, + loopbackProbe, + ); validatedPort = port; + validatedLoopbackNonce = expectedLoopbackNonce; lastError = null; lastSuccessAt = result.checkedAt; } catch (error) { validatedPort = null; + validatedLoopbackNonce = null; const reason = `Relay bridge refused because 127.0.0.1:${port} is not the ADE sync listener: ${error instanceof Error ? error.message : String(error)}`; recordFailure(reason); log.warn?.("sync_tunnel.loopback_validation_failed", { @@ -321,12 +338,16 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT getStatus(): SyncTunnelClientStatus { const { machineKey } = identity(); const currentPort = args.getSyncPort(); + const currentLoopbackNonce = args.getExpectedLoopbackNonce?.() ?? null; return { enabled: args.configStore.isEnabled(), connected, activeTunnels: tunnels.size, lastError, - relayBridgeValidated: currentPort != null && validatedPort === currentPort, + relayBridgeValidated: currentPort != null + && currentLoopbackNonce != null + && validatedPort === currentPort + && validatedLoopbackNonce === currentLoopbackNonce, validatedPort, lastFailureAt, lastSuccessAt,