diff --git a/apps/ade-cli/src/bootstrap.ts b/apps/ade-cli/src/bootstrap.ts index 8053af5e4..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 @@ -1588,6 +1590,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..6f7d29177 100644 --- a/apps/ade-cli/src/services/sync/sharedSyncListener.ts +++ b/apps/ade-cli/src/services/sync/sharedSyncListener.ts @@ -1,6 +1,16 @@ +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"; +import { + assertAdeLoopbackListener, + generateLoopbackNonce, + isLoopbackShadowedError, + probeAdeLoopbackListener, + writeAdeLoopbackUpgradeResponse, + 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 +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 @@ -133,10 +147,46 @@ 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, + candidateHttpServer: http.Server, +): Promise { + for (const client of candidateServer.clients) { + try { + client.terminate(); + } catch { + // ignore cleanup failures on a rejected candidate + } + } + // Closing the WebSocketServer only detaches its listeners from the + // externally-supplied http server; the http server owns the port and must be + // closed explicitly so a rejected candidate does not leak the bind. ws's + // close() for an external server resolves only after every client drains, so + // close the http server directly (forcing lingering sockets) instead. + try { + candidateServer.close(); + } catch { + // ignore + } + await new Promise((resolve) => { + if (!candidateHttpServer.listening) { + resolve(); + return; + } + try { + candidateHttpServer.close(() => resolve()); + candidateHttpServer.closeAllConnections?.(); + } catch { + resolve(); + } + }); +} + function rawDataBytes(data: RawData): number { if (typeof data === "string") return Buffer.byteLength(data, "utf8"); if (Buffer.isBuffer(data)) return data.length; @@ -152,20 +202,64 @@ export function createSharedSyncListener(options: { bindHost?: string; maxPayloadBytes?: number; parkedPeerGraceMs?: number; + 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 + // 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; 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 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); @@ -269,43 +363,93 @@ 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 (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((request, response) => { + writeAdeLoopbackUpgradeResponse(request, response, expectedLoopbackNonce); + }); 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) => { + 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 { + await validateLoopback(port); + cleanup(); + resolve(port); + } catch (error) { + cleanup(); + 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; + httpServer = candidateHttpServer; server.on("error", (error: unknown) => { logger.warn?.("sync_listener.server_error", { error: error instanceof Error ? error.message : String(error), @@ -313,42 +457,21 @@ 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 + 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; + 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 +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; @@ -441,24 +580,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 64820daf2..e448108ee 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"; @@ -1768,6 +1769,121 @@ 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", + markerValue: null, + 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("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, + expectedNonce: string, + ): Promise => + probeOk + ? { 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, + 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, + expectedNonce: string, + ): Promise => + probeOk + ? { 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, + 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({ @@ -3063,6 +3179,142 @@ 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("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" }); + 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/syncHostService.ts b/apps/ade-cli/src/services/sync/syncHostService.ts index 15b58640b..89b949043 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"; @@ -144,6 +145,15 @@ import { type SharedSyncListener, type SyncPeerHandoffSnapshot, } from "./sharedSyncListener"; +import { + assertAdeLoopbackListener, + generateLoopbackNonce, + isLoopbackShadowedError, + probeAdeLoopbackListener, + writeAdeLoopbackUpgradeResponse, + 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 +720,8 @@ type SyncHostServiceArgs = { * re-scanning a QR. */ getCloudRelayWssUrl?: () => string | null; + /** Test seam; production always uses the HTTP 426 loopback probe. */ + loopbackProbe?: (port: number, expectedNonce: string) => Promise; }; function sanitizeRemoteAddress(remoteAddress: string | null | undefined): string | null { @@ -1954,19 +1966,44 @@ 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. + // + // 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((request, response) => { + writeAdeLoopbackUpgradeResponse(request, response, expectedLoopbackNonce); + }); 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; + 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 +2878,58 @@ export function createSyncHostService(args: SyncHostServiceArgs) { } }; + 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 ( + !options?.force + && loopbackValidationStatus.port === port + && loopbackValidationStatus.loopbackAdeValidated + ) return; + 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 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 +5401,18 @@ 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); + 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.`); + } + await publishValidatedDiscovery(port); return port; } if (startupError) { @@ -5322,8 +5421,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 +5454,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 +5474,18 @@ 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 }); + // 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), + }); + }); } }, @@ -5399,8 +5508,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), + }); + }); } }, @@ -5453,6 +5572,10 @@ export function createSyncHostService(args: SyncHostServiceArgs) { return { ...tailnetDiscoveryStatus }; }, + getLoopbackValidationStatus(): SyncLoopbackValidationStatus { + return { ...loopbackValidationStatus }; + }, + getLanePresenceSnapshot(): Array<{ laneId: string; devicesOpen: DeviceMarker[] }> { return getLanePresenceSnapshot(); }, @@ -5621,12 +5744,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 new file mode 100644 index 000000000..b3ffc982b --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncLoopbackCollision.test.ts @@ -0,0 +1,436 @@ +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 { 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(() => { + 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 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(); + 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, + listener.getExpectedLoopbackNonce(), + )).resolves.toMatchObject({ + ok: false, + statusCode: 404, + }); + await expect(probeAdeLoopbackListener( + resolvedPort!, + listener.getExpectedLoopbackNonce(), + )).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, expectedNonce: string) => { + if (firstProbePort == null) { + firstProbePort = port; + return await firstProbe; + } + return await probeAdeLoopbackListener(port, expectedNonce); + }); + 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", + markerValue: null, + 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; + } + }); + + // 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; + } + }, + ); + + // 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 expectedNonce = adeListener.getExpectedLoopbackNonce(); + const foreignResult = await probeAdeLoopbackListener(foreign.port, expectedNonce); + expect(foreignResult).toMatchObject({ ok: false, statusCode: 426 }); + expect(foreignResult.reason).toMatch(/did not present a loopback identity/); + + const adeResult = await probeAdeLoopbackListener(adePort, expectedNonce); + 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, + expectedNonce: string, + ): Promise => { + if (shadowsRemaining > 0) { + shadowsRemaining -= 1; + shadowedPorts.push(port); + return { + ok: false, + 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, expectedNonce); + }); + 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", + markerValue: null, + 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..26504e272 --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncLoopbackProbe.test.ts @@ -0,0 +1,128 @@ +import http from "node:http"; +import { afterEach, describe, expect, it } from "vitest"; +import { + assertAdeLoopbackListener, + generateLoopbackNonce, + isLoopbackShadowedError, + probeAdeLoopbackListener, + SYNC_LOOPBACK_ADE_MARKER_HEADER, + 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 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(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 loopback identity", 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), + }); + response.end(body); + }); + const result = await probeAdeLoopbackListener(port, expectedNonce); + expect(result.ok).toBe(false); + expect(result.statusCode).toBe(426); + 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, expectedNonce)) + .rejects.toSatisfy(isLoopbackShadowedError); + }); + + 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]: "1", + }); + response.end(body); + }); + 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]: expectedNonce, + }); + response.end("nope"); + }); + 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 new file mode 100644 index 000000000..fb81de4e7 --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncLoopbackProbe.ts @@ -0,0 +1,165 @@ +import crypto from "node:crypto"; +import http from "node:http"; + +export const SYNC_LOOPBACK_PROBE_TIMEOUT_MS = 1_500; + +/** + * 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 function generateLoopbackNonce(): string { + return crypto.randomBytes(16).toString("hex"); +} + +/** + * 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, + 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` / + // `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]: nonce, + }); + response.end(body); +} + +export type SyncLoopbackProbeResult = { + ok: boolean; + port: number; + statusCode: number | null; + statusMessage: string | null; + markerValue: 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, + expectedNonce: string, + 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 rawMarker = response.headers[SYNC_LOOPBACK_ADE_MARKER_HEADER]; + 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 + : statusCode !== 426 + ? `Expected ADE 426 Upgrade Required on 127.0.0.1:${port}, received ${statusCode ?? "no status"}${statusMessage ? ` ${statusMessage}` : ""}.` + : 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, () => { + request.destroy(new Error(`Loopback ADE probe timed out after ${timeoutMs}ms.`)); + }); + request.once("error", (error) => { + finish({ + ok: false, + statusCode: null, + statusMessage: null, + markerValue: null, + reason: `ADE loopback probe failed on 127.0.0.1:${port}: ${error.message}`, + }); + }); + }); +} + +export async function assertAdeLoopbackListener( + port: number, + expectedNonce: string, + probe: (port: number, expectedNonce: string) => Promise = probeAdeLoopbackListener, +): Promise { + const result = await probe(port, expectedNonce); + 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..80eb61e6b 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, expectedNonce: string) => 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); @@ -783,14 +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 (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; @@ -801,10 +832,20 @@ export function createSyncService(args: SyncServiceArgs) { return; } catch (error) { lastError = error; + if (isLoopbackShadowedError(error)) { + shadowedPorts.add(attemptedPort === 0 ? error.port : 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 +1207,92 @@ 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." + : !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, + 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 +1315,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 +1472,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..6efd5e749 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,59 @@ 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 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, + }); + + 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(loopbackProbe).toHaveBeenCalledWith(8787, expectedNonce); + 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..6b8d0bfeb 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; }; @@ -37,9 +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, expectedNonce: string) => Promise; }; const BACKOFF_BASE_MS = 1_000; @@ -95,8 +108,18 @@ export function createSyncTunnelClientService(args: SyncTunnelClientArgs): SyncT let stopped = false; 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; 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 +168,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 +181,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 +196,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 +209,40 @@ 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; } + 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, + 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", { + 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 +250,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 +285,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 +337,20 @@ 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 + && currentLoopbackNonce != null + && validatedPort === currentPort + && validatedLoopbackNonce === currentLoopbackNonce, + 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"