diff --git a/apps/desktop/src/ipc/DesktopIpcHandlers.ts b/apps/desktop/src/ipc/DesktopIpcHandlers.ts index a16535c4a5eb..020d067f0615 100644 --- a/apps/desktop/src/ipc/DesktopIpcHandlers.ts +++ b/apps/desktop/src/ipc/DesktopIpcHandlers.ts @@ -20,6 +20,7 @@ import { setTailscaleServeEnabled, } from "./methods/serverExposure.ts"; import { + acquireSshPortForward, bootstrapSshBearerSession, disconnectSshEnvironment, discoverSshHosts, @@ -27,6 +28,7 @@ import { fetchSshEnvironmentDescriptor, fetchSshSessionState, issueSshWebSocketTicket, + releaseSshPortForward, resolveSshHost, resolveSshPasswordPrompt, } from "./methods/sshEnvironment.ts"; @@ -116,6 +118,8 @@ export const installDesktopIpcHandlers = Effect.fn("desktop.ipc.installHandlers" yield* ipc.handle(resolveSshHost); yield* ipc.handle(ensureSshEnvironment); yield* ipc.handle(disconnectSshEnvironment); + yield* ipc.handle(acquireSshPortForward); + yield* ipc.handle(releaseSshPortForward); yield* ipc.handle(fetchSshEnvironmentDescriptor); yield* ipc.handle(bootstrapSshBearerSession); yield* ipc.handle(fetchSshSessionState); diff --git a/apps/desktop/src/ipc/channels.ts b/apps/desktop/src/ipc/channels.ts index 098748dff124..9a76b9faf985 100644 --- a/apps/desktop/src/ipc/channels.ts +++ b/apps/desktop/src/ipc/channels.ts @@ -52,6 +52,8 @@ export const CLEAR_CONNECTION_CATALOG_CHANNEL = "desktop:clear-connection-catalo export const DISCOVER_SSH_HOSTS_CHANNEL = "desktop:discover-ssh-hosts"; export const RESOLVE_SSH_HOST_CHANNEL = "desktop:resolve-ssh-host"; export const ENSURE_SSH_ENVIRONMENT_CHANNEL = "desktop:ensure-ssh-environment"; +export const ACQUIRE_SSH_PORT_FORWARD_CHANNEL = "desktop:acquire-ssh-port-forward"; +export const RELEASE_SSH_PORT_FORWARD_CHANNEL = "desktop:release-ssh-port-forward"; export const DISCONNECT_SSH_ENVIRONMENT_CHANNEL = "desktop:disconnect-ssh-environment"; export const FETCH_SSH_ENVIRONMENT_DESCRIPTOR_CHANNEL = "desktop:fetch-ssh-environment-descriptor"; export const BOOTSTRAP_SSH_BEARER_SESSION_CHANNEL = "desktop:bootstrap-ssh-bearer-session"; diff --git a/apps/desktop/src/ipc/methods/sshEnvironment.test.ts b/apps/desktop/src/ipc/methods/sshEnvironment.test.ts index 0ed188ca61b5..b45d0417deea 100644 --- a/apps/desktop/src/ipc/methods/sshEnvironment.test.ts +++ b/apps/desktop/src/ipc/methods/sshEnvironment.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest"; -import { SshHttpBridgeError } from "@t3tools/ssh/errors"; +import { SshHttpBridgeError, SshPasswordPromptError, SshReadinessError } from "@t3tools/ssh/errors"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -8,10 +8,40 @@ import * as Option from "effect/Option"; import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; import { + acquireSshPortForward, + releaseSshPortForward, DesktopSshEnvironmentRequestError, fetchSshEnvironmentDescriptor, } from "./sshEnvironment.ts"; +import * as DesktopSshEnvironment from "../../ssh/DesktopSshEnvironment.ts"; +import * as DesktopSshPasswordPrompts from "../../ssh/DesktopSshPasswordPrompts.ts"; + +const forwardTarget = { + alias: "devbox", + hostname: "devbox.example.test", + username: "developer", + port: 22, +}; +const forwardLayer = ( + acquire: DesktopSshEnvironment.DesktopSshEnvironment["Service"]["acquirePortForward"], + releases: string[] = [], +) => + Layer.succeed( + DesktopSshEnvironment.DesktopSshEnvironment, + DesktopSshEnvironment.DesktopSshEnvironment.of({ + discoverHosts: () => Effect.die("unused"), + resolveHost: () => Effect.die("unused"), + ensureEnvironment: () => Effect.die("unused"), + disconnectEnvironment: () => Effect.die("unused"), + acquirePortForward: acquire, + releasePortForward: (leaseId) => + Effect.sync(() => { + releases.push(leaseId); + }), + }), + ); + function jsonResponse(request: HttpClientRequest.HttpClientRequest, body: unknown, status = 200) { return HttpClientResponse.fromWeb( request, @@ -34,6 +64,86 @@ function layerHttpClient( } describe("SSH environment IPC", () => { + it.effect("acquires distinct leases and passes repeated releases through", () => { + let count = 0; + const releases: string[] = []; + const layer = forwardLayer( + (target, port) => + Effect.sync(() => { + assert.deepEqual(target, forwardTarget); + assert.equal(port, 5173); + return { leaseId: `lease-${++count}`, localPort: 41773 }; + }), + releases, + ); + return Effect.gen(function* () { + assert.deepEqual( + yield* acquireSshPortForward.handler({ target: forwardTarget, remotePort: 5173 }), + { leaseId: "lease-1", localPort: 41773 }, + ); + assert.deepEqual( + yield* acquireSshPortForward.handler({ target: forwardTarget, remotePort: 5173 }), + { leaseId: "lease-2", localPort: 41773 }, + ); + yield* releaseSshPortForward.handler({ leaseId: "lease-1" }); + yield* releaseSshPortForward.handler({ leaseId: "lease-1" }); + assert.deepEqual(releases, ["lease-1", "lease-1"]); + }).pipe(Effect.provide(layer)); + }); + + it.effect("returns the same prompt-cancellation envelope as ensure", () => { + const layer = forwardLayer(() => + Effect.fail( + new SshPasswordPromptError({ + message: "SSH authentication cancelled for devbox.", + cause: new DesktopSshPasswordPrompts.DesktopSshPromptCancelledError({ + requestId: "prompt-1", + destination: "devbox", + }), + }), + ), + ); + return Effect.gen(function* () { + assert.deepEqual( + yield* acquireSshPortForward.handler({ target: forwardTarget, remotePort: 5173 }), + { + type: "ssh-password-prompt-cancelled", + message: "SSH authentication cancelled for devbox.", + }, + ); + }).pipe(Effect.provide(layer)); + }); + + it.effect("does not disguise startup failures as cancellation", () => { + const layer = forwardLayer(() => + Effect.fail(new SshReadinessError({ message: "Forward failed." })), + ); + return Effect.gen(function* () { + const exit = yield* Effect.exit( + acquireSshPortForward.handler({ target: forwardTarget, remotePort: 5173 }), + ); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + const error = Cause.findErrorOption(exit.cause); + assert.isTrue(Option.isSome(error)); + if (Option.isSome(error)) assert.instanceOf(error.value, SshReadinessError); + } + }).pipe(Effect.provide(layer)); + }); + + it.effect("rejects invalid remote ports before calling the manager", () => { + const layer = forwardLayer(() => Effect.die("invalid port reached manager")); + return Effect.gen(function* () { + for (const remotePort of [0, -1, 65536, 5173.5, "5173"]) { + const exit = yield* Effect.exit( + acquireSshPortForward.handler({ target: forwardTarget, remotePort }), + ); + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) assert.isFalse(Cause.hasDies(exit.cause)); + } + }).pipe(Effect.provide(layer)); + }); + it.effect("fetches and decodes the remote environment descriptor", () => { const requestUrls: string[] = []; const layer = layerHttpClient((request) => diff --git a/apps/desktop/src/ipc/methods/sshEnvironment.ts b/apps/desktop/src/ipc/methods/sshEnvironment.ts index e9535300104d..df0d0abbe94a 100644 --- a/apps/desktop/src/ipc/methods/sshEnvironment.ts +++ b/apps/desktop/src/ipc/methods/sshEnvironment.ts @@ -15,6 +15,9 @@ import { DesktopSshEnvironmentEnsureResultSchema, DesktopSshEnvironmentTargetSchema, DesktopSshHttpBaseUrlInputSchema, + DesktopSshPortForwardAcquireInputSchema, + DesktopSshPortForwardAcquireResultSchema, + DesktopSshPortForwardReleaseInputSchema, DesktopSshPasswordPromptCancelledType, DesktopSshPasswordPromptResolutionInputSchema, ExecutionEnvironmentDescriptor, @@ -157,6 +160,36 @@ export const disconnectSshEnvironment = DesktopIpc.makeIpcMethod({ }), }); +export const acquireSshPortForward = DesktopIpc.makeIpcMethod({ + channel: IpcChannels.ACQUIRE_SSH_PORT_FORWARD_CHANNEL, + payload: DesktopSshPortForwardAcquireInputSchema, + result: DesktopSshPortForwardAcquireResultSchema, + handler: Effect.fn("desktop.ipc.sshEnvironment.acquirePortForward")(function* ({ + target, + remotePort, + }) { + const sshEnvironment = yield* DesktopSshEnvironment.DesktopSshEnvironment; + return yield* sshEnvironment.acquirePortForward(target, remotePort).pipe( + Effect.catchIf(DesktopSshEnvironment.isDesktopSshPasswordPromptCancellation, (error) => + Effect.succeed({ + type: DesktopSshPasswordPromptCancelledType, + message: error.message, + }), + ), + ); + }), +}); + +export const releaseSshPortForward = DesktopIpc.makeIpcMethod({ + channel: IpcChannels.RELEASE_SSH_PORT_FORWARD_CHANNEL, + payload: DesktopSshPortForwardReleaseInputSchema, + result: Schema.Void, + handler: Effect.fn("desktop.ipc.sshEnvironment.releasePortForward")(function* ({ leaseId }) { + const sshEnvironment = yield* DesktopSshEnvironment.DesktopSshEnvironment; + yield* sshEnvironment.releasePortForward(leaseId); + }), +}); + export const fetchSshEnvironmentDescriptor = DesktopIpc.makeIpcMethod({ channel: IpcChannels.FETCH_SSH_ENVIRONMENT_DESCRIPTOR_CHANNEL, payload: DesktopSshHttpBaseUrlInputSchema, diff --git a/apps/desktop/src/preload.ts b/apps/desktop/src/preload.ts index 7d7269974bcd..aa5040e7a418 100644 --- a/apps/desktop/src/preload.ts +++ b/apps/desktop/src/preload.ts @@ -59,7 +59,7 @@ if (clientPlatform === "darwin") { window.addEventListener("resize", syncWindowControlInset); } -function unwrapEnsureSshEnvironmentResult(result: unknown) { +function unwrapSshPasswordPromptResult(result: T): T { if ( typeof result === "object" && result !== null && @@ -72,7 +72,7 @@ function unwrapEnsureSshEnvironmentResult(result: unknown) { : "SSH authentication cancelled."; throw new Error(message); } - return result as Awaited>; + return result; } contextBridge.exposeInMainWorld("desktopBridge", { @@ -142,7 +142,7 @@ contextBridge.exposeInMainWorld("desktopBridge", { discoverSshHosts: () => ipcRenderer.invoke(IpcChannels.DISCOVER_SSH_HOSTS_CHANNEL), resolveSshHost: (alias) => ipcRenderer.invoke(IpcChannels.RESOLVE_SSH_HOST_CHANNEL, alias), ensureSshEnvironment: async (target, options) => - unwrapEnsureSshEnvironmentResult( + unwrapSshPasswordPromptResult>>( await ipcRenderer.invoke(IpcChannels.ENSURE_SSH_ENVIRONMENT_CHANNEL, { target, ...(options === undefined ? {} : { options }), @@ -150,6 +150,15 @@ contextBridge.exposeInMainWorld("desktopBridge", { ), disconnectSshEnvironment: (target) => ipcRenderer.invoke(IpcChannels.DISCONNECT_SSH_ENVIRONMENT_CHANNEL, target), + acquireSshPortForward: async (target, remotePort) => + unwrapSshPasswordPromptResult>>( + await ipcRenderer.invoke(IpcChannels.ACQUIRE_SSH_PORT_FORWARD_CHANNEL, { + target, + remotePort, + }), + ), + releaseSshPortForward: (leaseId) => + ipcRenderer.invoke(IpcChannels.RELEASE_SSH_PORT_FORWARD_CHANNEL, { leaseId }), fetchSshEnvironmentDescriptor: (httpBaseUrl) => ipcRenderer.invoke(IpcChannels.FETCH_SSH_ENVIRONMENT_DESCRIPTOR_CHANNEL, { httpBaseUrl }), bootstrapSshBearerSession: (httpBaseUrl, credential) => diff --git a/apps/desktop/src/ssh/DesktopSshEnvironment.ts b/apps/desktop/src/ssh/DesktopSshEnvironment.ts index 26e307b0c250..01f26888ead0 100644 --- a/apps/desktop/src/ssh/DesktopSshEnvironment.ts +++ b/apps/desktop/src/ssh/DesktopSshEnvironment.ts @@ -61,6 +61,11 @@ export class DesktopSshEnvironment extends Context.Service< readonly disconnectEnvironment: ( target: DesktopSshEnvironmentTarget, ) => Effect.Effect; + readonly acquirePortForward: ( + target: DesktopSshEnvironmentTarget, + remotePort: number, + ) => Effect.Effect; + readonly releasePortForward: (leaseId: string) => Effect.Effect; } >()("@t3tools/desktop/ssh/DesktopSshEnvironment") {} @@ -157,6 +162,16 @@ export const make = Effect.gen(function* () { Effect.provide(runtimeContext), Effect.withSpan("desktop.ssh.disconnectEnvironment"), ), + acquirePortForward: (target, remotePort) => + manager + .acquirePortForward(target, remotePort) + .pipe( + Effect.provideService(SshAuth.SshPasswordPrompt, passwordPrompt), + Effect.provide(runtimeContext), + Effect.withSpan("desktop.ssh.acquirePortForward"), + ), + releasePortForward: (leaseId) => + manager.releasePortForward(leaseId).pipe(Effect.withSpan("desktop.ssh.releasePortForward")), }); }); diff --git a/apps/web/src/browser/browserTargetResolver.test.ts b/apps/web/src/browser/browserTargetResolver.test.ts index c2b3432402ed..367b326fa0f4 100644 --- a/apps/web/src/browser/browserTargetResolver.test.ts +++ b/apps/web/src/browser/browserTargetResolver.test.ts @@ -4,9 +4,15 @@ import { beforeEach, describe, expect, it, vi } from "vite-plus/test"; const readPreparedConnection = vi.fn(); vi.mock("~/state/session", () => ({ readPreparedConnection })); +const readSshEnvironmentTarget = vi.fn(() => null as unknown); +vi.mock("./sshPreviewForwards", () => ({ readSshEnvironmentTarget })); describe("browser target resolver", () => { - beforeEach(() => readPreparedConnection.mockReset()); + beforeEach(() => { + readPreparedConnection.mockReset(); + readSshEnvironmentTarget.mockReset(); + readSshEnvironmentTarget.mockReturnValue(null); + }); it("maps environment ports onto a private network host", async () => { readPreparedConnection.mockReturnValue({ httpBaseUrl: "http://192.168.1.25:3773" }); @@ -141,6 +147,25 @@ describe("browser target resolver", () => { ).toMatchObject({ resolvedUrl: "http://localhost:5173", resolutionKind: "direct" }); }); + it("keeps SSH environment ports remote-loopback, normalized, for forwarding", async () => { + readPreparedConnection.mockReturnValue({ httpBaseUrl: "http://127.0.0.1:41000" }); + readSshEnvironmentTarget.mockReturnValue({ alias: "devbox" }); + const { resolveBrowserNavigationTarget, resolveDiscoveredServerUrl } = + await import("./browserTargetResolver"); + const environmentId = EnvironmentId.make("environment-ssh"); + expect(resolveDiscoveredServerUrl(environmentId, "localhost:5173")).toBe( + "http://localhost:5173/", + ); + expect( + resolveBrowserNavigationTarget(environmentId, { kind: "environment-port", port: 5173 }), + ).toEqual({ + requestedUrl: "http://localhost:5173/", + resolvedUrl: "http://localhost:5173/", + resolutionKind: "ssh-forward", + environmentId, + }); + }); + it("normalizes schemeless localhost server-picker values", async () => { readPreparedConnection.mockReturnValue({ httpBaseUrl: "http://localhost:3773" }); const { resolveDiscoveredServerUrl } = await import("./browserTargetResolver"); diff --git a/apps/web/src/browser/browserTargetResolver.ts b/apps/web/src/browser/browserTargetResolver.ts index 4acb1e2b487f..4e0dbfac2244 100644 --- a/apps/web/src/browser/browserTargetResolver.ts +++ b/apps/web/src/browser/browserTargetResolver.ts @@ -8,6 +8,8 @@ import { isLocalLoopbackHost, isPrivateNetworkHost } from "@t3tools/shared/hostC import { readPreparedConnection } from "~/state/session"; +import { readSshEnvironmentTarget } from "./sshPreviewForwards"; + export { normalizeHostname, isLocalLoopbackHost, @@ -28,13 +30,25 @@ const resolveEnvironmentPortTarget = ( requestedUrl?: string, sourceUrl?: URL, ): PreviewUrlResolution => { + const protocol = target.protocol ?? "http"; + const path = target.path?.startsWith("/") ? target.path : `/${target.path ?? ""}`; + const loopbackUrl = requestedUrl ?? `${protocol}://localhost:${target.port}${path}`; + // An SSH environment's server URL is itself a local forward, so its host says + // nothing about where the port lives. The URL stays remote-loopback and + // navigation forwards it (see sshPreviewForwards). + if (readSshEnvironmentTarget(environmentId) !== null) { + return { + requestedUrl: loopbackUrl, + resolvedUrl: (sourceUrl ?? new URL(loopbackUrl)).toString(), + resolutionKind: "ssh-forward", + environmentId, + }; + } if (!isPrivateNetworkHost(environmentUrl.hostname)) { throw new Error( "This environment port needs the planned authenticated preview gateway; its server address is not directly private-network reachable.", ); } - const protocol = target.protocol ?? "http"; - const path = target.path?.startsWith("/") ? target.path : `/${target.path ?? ""}`; const normalizedEnvironmentHost = environmentUrl.hostname.replace(/^\[|\]$/g, ""); // Local loopback environments should advertise `localhost` so Chromium // dual-stack lookup can reach a Vite server bound only to ::1 or 127.0.0.1. @@ -51,7 +65,7 @@ const resolveEnvironmentPortTarget = ( resolved.port = String(target.port); } return { - requestedUrl: requestedUrl ?? `${protocol}://localhost:${target.port}${path}`, + requestedUrl: loopbackUrl, resolvedUrl: resolved.toString(), resolutionKind: isLocalLoopbackHost(normalizedEnvironmentHost) ? "direct" diff --git a/apps/web/src/browser/openFileInPreview.ts b/apps/web/src/browser/openFileInPreview.ts index 38ab15eb826b..4080f7b9fc6d 100644 --- a/apps/web/src/browser/openFileInPreview.ts +++ b/apps/web/src/browser/openFileInPreview.ts @@ -20,9 +20,17 @@ import { applyPreviewServerSnapshot, isPreviewSupportedInRuntime, rememberPreviewUrl, + settleOpenedPreviewForward, } from "~/previewStateStore"; import { useRightPanelStore } from "~/rightPanelStore"; +import { + acquirePreviewForward, + type PreviewForward, + isSshPreviewForwardError, + type SshPreviewForwardError, +} from "./sshPreviewForwards"; + import { browserDefaultOpenProfileId, browserDefaultOpenViewport, @@ -55,18 +63,25 @@ export async function openUrlInPreview(input: { readonly threadRef: ScopedThreadRef; readonly url: string; readonly openPreview: OpenPreviewMutation; -}): Promise> { +}): Promise> { const defaults = await resolveBrowserDefaults().catch( (cause: unknown) => new BrowserSettingsReadError({ cause }), ); if (defaults instanceof BrowserSettingsReadError) { return AsyncResult.failure(Cause.fail(defaults)); } + let forward: PreviewForward; + try { + forward = await acquirePreviewForward(input.threadRef.environmentId, input.url); + } catch (error) { + if (!isSshPreviewForwardError(error)) throw error; + return AsyncResult.failure(Cause.fail(error)); + } const result = await input.openPreview({ environmentId: input.threadRef.environmentId, input: { threadId: input.threadRef.threadId, - url: input.url, + url: forward.url, // Built here rather than via `openPreviewSession` because this path // maps the result differently, so the configured defaults have to be // applied explicitly or file/link opens would ignore them. @@ -74,8 +89,10 @@ export async function openUrlInPreview(input: { profileId: browserDefaultOpenProfileId(defaults), }, }); + if (result._tag === "Failure") settleOpenedPreviewForward(input.threadRef, forward, null); return mapAtomCommandResult(result, (snapshot) => { applyPreviewServerSnapshot(input.threadRef, snapshot); + settleOpenedPreviewForward(input.threadRef, forward, snapshot.tabId); rememberPreviewUrl(input.threadRef, input.url); useRightPanelStore.getState().openBrowser(input.threadRef, snapshot.tabId); }); @@ -98,7 +115,11 @@ export async function openFileInPreview(input: { }): Promise< AtomCommandResult< void, - AssetError | PreviewError | BrowserPreviewUnavailableError | BrowserSettingsReadError + | AssetError + | PreviewError + | BrowserPreviewUnavailableError + | BrowserSettingsReadError + | SshPreviewForwardError > > { if (!isPreviewSupportedInRuntime()) { diff --git a/apps/web/src/browser/sshPreviewForwards.test.ts b/apps/web/src/browser/sshPreviewForwards.test.ts new file mode 100644 index 000000000000..eeb548e45847 --- /dev/null +++ b/apps/web/src/browser/sshPreviewForwards.test.ts @@ -0,0 +1,215 @@ +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import * as Option from "effect/Option"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +import { + acquirePreviewForward, + navigateTabThroughForward, + releasePreviewForward, + releaseTabForward, + resetSshPreviewForwardsForTests, + settleOpenedForward, + SshPreviewForwardError, + toRemotePreviewUrl, +} from "./sshPreviewForwards"; + +const SSH_ENV = EnvironmentId.make("environment-ssh"); +const OTHER_SSH_ENV = EnvironmentId.make("environment-ssh-2"); +const LAN_ENV = EnvironmentId.make("environment-lan"); +const THREAD = { environmentId: SSH_ENV, threadId: ThreadId.make("thread-1") }; +const SSH_TARGET = { alias: "devbox", hostname: "devbox", username: null, port: null }; + +const catalogEntries = new Map([ + [ + SSH_ENV, + { + target: { _tag: "SshConnectionTarget" }, + profile: Option.some({ _tag: "SshConnectionProfile", target: SSH_TARGET }), + }, + ], + [ + OTHER_SSH_ENV, + { + target: { _tag: "SshConnectionTarget" }, + profile: Option.some({ _tag: "SshConnectionProfile", target: SSH_TARGET }), + }, + ], + [LAN_ENV, { target: { _tag: "BearerConnectionTarget" }, profile: Option.none() }], +]); + +vi.mock("~/connection/catalog", () => ({ environmentCatalog: { catalogValueAtom: "catalog" } })); +vi.mock("~/rpc/atomRegistry", () => ({ + appAtomRegistry: { get: () => ({ entries: catalogEntries }) }, +})); +vi.mock("~/state/session", () => ({ + readPreparedConnection: () => ({ httpBaseUrl: "http://127.0.0.1:41000" }), +})); + +interface PendingAcquire { + readonly remotePort: number; + readonly resolve: (localPort: number) => void; +} + +let pending: PendingAcquire[]; +let released: string[]; +let nextLease: number; + +beforeEach(() => { + pending = []; + released = []; + nextLease = 0; + vi.stubGlobal("window", { + desktopBridge: { + acquireSshPortForward: (_target: unknown, remotePort: number) => + new Promise((resolve) => { + pending.push({ + remotePort, + resolve: (localPort) => resolve({ leaseId: `lease-${++nextLease}`, localPort }), + }); + }), + releaseSshPortForward: async (leaseId: string) => { + released.push(leaseId); + }, + }, + }); +}); + +afterEach(() => { + resetSshPreviewForwardsForTests(); + vi.unstubAllGlobals(); +}); + +describe("acquirePreviewForward", () => { + it("rewrites a remote loopback URL onto the forwarded local port", async () => { + const forward = acquirePreviewForward(SSH_ENV, "http://127.0.0.1:5173/app?x=1#top"); + expect(pending.map((entry) => entry.remotePort)).toEqual([5173]); + pending[0]!.resolve(53001); + expect(await forward).toEqual({ + url: "http://localhost:53001/app?x=1#top", + leaseId: "lease-1", + }); + }); + + it("forwards any 127.0.0.0/8 address to the remote machine", async () => { + const forward = acquirePreviewForward(SSH_ENV, "http://127.0.0.2:5173/"); + expect(pending.map((entry) => entry.remotePort)).toEqual([5173]); + pending[0]!.resolve(5173); + expect((await forward).url).toBe("http://localhost:5173/"); + }); + + it("leaves non-SSH environments, public hosts, and the environment server alone", async () => { + for (const [environmentId, url] of [ + [LAN_ENV, "http://localhost:5173/"], + [SSH_ENV, "https://example.com/"], + [SSH_ENV, "http://127.0.0.1:41000/assets/file.html"], + ] as const) { + expect(await acquirePreviewForward(environmentId, url)).toEqual({ url, leaseId: null }); + } + expect(pending).toEqual([]); + }); + + it("maps forwarded ports back per environment, after the lease is released", async () => { + const first = acquirePreviewForward(SSH_ENV, "http://localhost:5173/"); + pending[0]!.resolve(53001); + releasePreviewForward(await first); + expect(toRemotePreviewUrl(SSH_ENV, "http://localhost:53001/x")).toBe("http://localhost:5173/x"); + expect(toRemotePreviewUrl(OTHER_SSH_ENV, "http://localhost:53001/x")).toBe( + "http://localhost:53001/x", + ); + void acquirePreviewForward(SSH_ENV, "http://localhost:53001/next"); + void acquirePreviewForward(OTHER_SSH_ENV, "http://localhost:53001/next"); + expect(pending.slice(1).map((entry) => entry.remotePort)).toEqual([5173, 53001]); + }); + + it("rejects with a typed error when the desktop cannot forward", async () => { + vi.stubGlobal("window", { + desktopBridge: { + acquireSshPortForward: async () => { + throw new Error("Permission denied (publickey)."); + }, + }, + }); + const error = await acquirePreviewForward(SSH_ENV, "http://localhost:5173/").catch( + (cause: unknown) => cause, + ); + expect(error).toBeInstanceOf(SshPreviewForwardError); + expect((error as SshPreviewForwardError).message).toBe( + "Could not forward remote port 5173: Permission denied (publickey).", + ); + }); +}); + +describe("navigateTabThroughForward", () => { + const navigate = (tabId: string, url: string, loaded: string[]): Promise => + navigateTabThroughForward({ + threadRef: THREAD, + tabId, + url, + navigate: async (resolved) => { + loaded.push(resolved); + }, + }); + + it("lets only the newest overlapping navigation load and keep its lease", async () => { + const loaded: string[] = []; + const older = navigate("tab-1", "http://localhost:3000/", loaded); + const newer = navigate("tab-1", "http://localhost:4000/", loaded); + pending[1]!.resolve(54000); + expect(await newer).toBe("http://localhost:54000/"); + pending[0]!.resolve(53000); + expect(await older).toBeNull(); + expect(loaded).toEqual(["http://localhost:54000/"]); + // The superseded lease is dropped; the tab keeps the newer one. + expect(released).toEqual(["lease-2"]); + }); + + it("releases the previous lease after the swap and the last one on tab close", async () => { + const loaded: string[] = []; + const first = navigate("tab-1", "http://localhost:3000/", loaded); + pending[0]!.resolve(53000); + await first; + const second = navigate("tab-1", "http://localhost:4000/", loaded); + pending[1]!.resolve(54000); + await second; + expect(released).toEqual(["lease-1"]); + releaseTabForward(THREAD, "tab-1"); + expect(released).toEqual(["lease-1", "lease-2"]); + }); + + it("reports a tab closed during its navigation as superseded", async () => { + const loaded: string[] = []; + const navigation = navigateTabThroughForward({ + threadRef: THREAD, + tabId: "tab-1", + url: "http://localhost:3000/", + navigate: async (resolved) => { + loaded.push(resolved); + releaseTabForward(THREAD, "tab-1"); + }, + }); + pending[0]!.resolve(53000); + expect(await navigation).toBeNull(); + expect(released).toEqual(["lease-1"]); + }); + + it("drops the lease when the tab closes during acquisition", async () => { + const loaded: string[] = []; + const navigation = navigate("tab-1", "http://localhost:3000/", loaded); + releaseTabForward(THREAD, "tab-1"); + pending[0]!.resolve(53000); + expect(await navigation).toBeNull(); + expect(loaded).toEqual([]); + expect(released).toEqual(["lease-1"]); + }); +}); + +describe("settleOpenedForward", () => { + it("gives the lease to the opened tab, or releases it when the open failed", async () => { + settleOpenedForward(THREAD, { url: "http://localhost:53000/", leaseId: "a" }, null); + expect(released).toEqual(["a"]); + settleOpenedForward(THREAD, { url: "http://localhost:53000/", leaseId: "b" }, "tab-7"); + expect(released).toEqual(["a"]); + releaseTabForward(THREAD, "tab-7"); + expect(released).toEqual(["a", "b"]); + }); +}); diff --git a/apps/web/src/browser/sshPreviewForwards.ts b/apps/web/src/browser/sshPreviewForwards.ts new file mode 100644 index 000000000000..f31b9e4bcce3 --- /dev/null +++ b/apps/web/src/browser/sshPreviewForwards.ts @@ -0,0 +1,248 @@ +import { scopedThreadKey } from "@t3tools/client-runtime/environment"; +import type { + DesktopSshEnvironmentTarget, + EnvironmentId, + ScopedThreadRef, +} from "@t3tools/contracts"; +import { isLocalLoopbackHost } from "@t3tools/shared/hostClassification"; +import { isLoopbackHost } from "@t3tools/shared/preview"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +import { environmentCatalog } from "~/connection/catalog"; +import { appAtomRegistry } from "~/rpc/atomRegistry"; +import { readPreparedConnection } from "~/state/session"; + +/** + * Loopback preview URLs on an SSH environment name ports on the remote host, so + * the desktop reaches them through an `ssh -L` forward. The desktop shares one + * forward per remote port across leases; this module tracks which lease each + * preview tab holds. A tab holds at most one lease, swapped when it navigates + * and released when the tab leaves the thread's preview state. The desktop tab + * itself may close and reopen on the same forwarded URL while the preview tab + * lives, so its lifetime is the wrong owner. + */ +export interface PreviewForward { + /** URL to load in the desktop tab. */ + readonly url: string; + readonly leaseId: string | null; +} + +export function readSshEnvironmentTarget( + environmentId: EnvironmentId, +): DesktopSshEnvironmentTarget | null { + const entry = appAtomRegistry.get(environmentCatalog.catalogValueAtom).entries.get(environmentId); + if (entry?.target._tag !== "SshConnectionTarget") return null; + const profile = Option.getOrNull(entry.profile); + return profile?._tag === "SshConnectionProfile" ? profile.target : null; +} + +const remoteLoopbackPort = (url: URL): number | null => { + if (url.protocol !== "http:" && url.protocol !== "https:") return null; + // All of 127.0.0.0/8 names the remote machine; the forward targets remote localhost. + if (!isLoopbackHost(url.hostname) && !isLocalLoopbackHost(url.hostname)) return null; + return Number(url.port || (url.protocol === "https:" ? 443 : 80)); +}; + +/** + * Local port → remote port per environment. The desktop reuses the remote port + * when it is free locally, so most entries are identity. Entries outlive their + * leases: forwarded URLs come back through history, recents, and server + * snapshots after the tab that created them is gone. + */ +const forwardedPorts = new Map>(); + +const isEnvironmentServer = (environmentId: EnvironmentId, url: URL): boolean => { + const connection = readPreparedConnection(environmentId); + if (connection === null) return false; + const server = new URL(connection.httpBaseUrl); + return isLoopbackHost(server.hostname) && server.port === url.port; +}; + +const remotePortFor = (environmentId: EnvironmentId, port: number): number => + forwardedPorts.get(environmentId)?.get(port) ?? port; + +/** Maps a URL loaded through a forward back to the remote URL it names. */ +export function toRemotePreviewUrl(environmentId: EnvironmentId, url: string): string { + const ports = forwardedPorts.get(environmentId); + if (ports === undefined) return url; + let parsed: URL; + try { + parsed = new URL(url); + } catch { + return url; + } + const loopbackPort = remoteLoopbackPort(parsed); + const remotePort = loopbackPort === null ? undefined : ports.get(loopbackPort); + if (remotePort === undefined || remotePort === loopbackPort) return url; + parsed.port = String(remotePort); + return parsed.toString(); +} + +export class SshPreviewForwardError extends Schema.TaggedError()( + "SshPreviewForwardError", + { remotePort: Schema.Number, detail: Schema.String, cause: Schema.optional(Schema.Defect()) }, +) { + override get message(): string { + return `Could not forward remote port ${this.remotePort}: ${this.detail}`; + } +} + +export const isSshPreviewForwardError = Schema.is(SshPreviewForwardError); + +/** + * Acquires a forward when `url` is a loopback URL on an SSH environment. + * Rejects with SshPreviewForwardError. + */ +export async function acquirePreviewForward( + environmentId: EnvironmentId, + url: string, +): Promise { + const target = readSshEnvironmentTarget(environmentId); + if (target === null) return { url, leaseId: null }; + let parsed: URL; + try { + parsed = new URL(url); + } catch { + return { url, leaseId: null }; + } + const loopbackPort = remoteLoopbackPort(parsed); + // The environment's own server is itself a local SSH forward (asset URLs). + if (loopbackPort === null || isEnvironmentServer(environmentId, parsed)) { + return { url, leaseId: null }; + } + const remotePort = remotePortFor(environmentId, loopbackPort); + const bridge = window.desktopBridge; + if (bridge === undefined) { + throw new SshPreviewForwardError({ + remotePort, + detail: "previewing ports on an SSH environment requires the desktop app.", + }); + } + const forward = await bridge.acquireSshPortForward(target, remotePort).catch((cause: unknown) => { + throw new SshPreviewForwardError({ + remotePort, + detail: cause instanceof Error ? cause.message : String(cause), + cause, + }); + }); + const ports = forwardedPorts.get(environmentId) ?? new Map(); + ports.set(forward.localPort, remotePort); + forwardedPorts.set(environmentId, ports); + // `localhost` keeps the page's origin host stable for cookies and OAuth + // callbacks; the forward itself only binds 127.0.0.1. + parsed.hostname = "localhost"; + parsed.port = String(forward.localPort); + return { url: parsed.toString(), leaseId: forward.leaseId }; +} + +const releaseLease = (leaseId: string | null): void => { + if (leaseId === null) return; + void window.desktopBridge?.releaseSshPortForward(leaseId).catch(() => undefined); +}; + +export function releasePreviewForward(forward: PreviewForward): void { + releaseLease(forward.leaseId); +} + +interface TabForward { + /** Increments per navigation; only the newest navigation may load or commit. */ + latest: number; + committed: number; + leaseId: string | null; +} + +const tabForwards = new Map(); + +const tabKey = (threadRef: ScopedThreadRef, tabId: string): string => + `${scopedThreadKey(threadRef)}\u0000${tabId}`; + +const tabForward = (key: string): TabForward => { + let current = tabForwards.get(key); + if (!current) { + current = { latest: 0, committed: 0, leaseId: null }; + tabForwards.set(key, current); + } + return current; +}; + +/** + * Hands a forward acquired for a new tab to the tab that now shows it, or + * releases it when the open failed. + */ +export function settleOpenedForward( + threadRef: ScopedThreadRef, + forward: PreviewForward | null, + tabId: string | null, +): void { + if (forward === null) return; + if (tabId === null) { + releasePreviewForward(forward); + return; + } + const state = tabForward(tabKey(threadRef, tabId)); + state.latest += 1; + commit(state, state.latest, forward.leaseId); +} + +function commit(state: TabForward, token: number, leaseId: string | null): void { + if (token <= state.committed) { + releaseLease(leaseId); + return; + } + const previous = state.leaseId; + state.committed = token; + state.leaseId = leaseId; + // Released after the swap: when both point at the same remote port the + // desktop keeps the shared ssh child alive across the handoff. + if (previous !== leaseId) releaseLease(previous); +} + +/** + * Navigates an existing desktop tab, forwarding SSH loopback URLs first. + * Returns the URL loaded, or null when a newer navigation of the same tab + * started while the forward was being acquired, or the tab closed. Navigate + * calls go out in token order, so a superseded navigation never loads last. + */ +export async function navigateTabThroughForward(input: { + readonly threadRef: ScopedThreadRef; + readonly tabId: string; + readonly url: string; + readonly navigate: (url: string) => Promise; +}): Promise { + const key = tabKey(input.threadRef, input.tabId); + const state = tabForward(key); + state.latest += 1; + const token = state.latest; + const forward = await acquirePreviewForward(input.threadRef.environmentId, input.url); + if (token !== state.latest || tabForwards.get(key) !== state) { + releasePreviewForward(forward); + return null; + } + try { + await input.navigate(forward.url); + } catch (error) { + releasePreviewForward(forward); + throw error; + } + if (tabForwards.get(key) !== state) { + releasePreviewForward(forward); + return null; + } + commit(state, token, forward.leaseId); + return forward.url; +} + +/** Called by the preview state store for every tab it drops, whatever dropped it. */ +export function releaseTabForward(threadRef: ScopedThreadRef, tabId: string): void { + const key = tabKey(threadRef, tabId); + const state = tabForwards.get(key); + if (!state) return; + tabForwards.delete(key); + releaseLease(state.leaseId); +} + +export function resetSshPreviewForwardsForTests(): void { + tabForwards.clear(); + forwardedPorts.clear(); +} diff --git a/apps/web/src/browser/useOpenLink.ts b/apps/web/src/browser/useOpenLink.ts index 2a1d122eedbf..071a4284075d 100644 --- a/apps/web/src/browser/useOpenLink.ts +++ b/apps/web/src/browser/useOpenLink.ts @@ -15,7 +15,10 @@ import { resolveBrowserLinkTargetPreference, resolveLinkTarget, } from "./browserLinkTarget"; +import { toastManager } from "~/components/ui/toast"; + import { BrowserSettingsReadError, openUrlInPreview } from "./openFileInPreview"; +import { isSshPreviewForwardError } from "./sshPreviewForwards"; const NO_MODIFIER = { metaKey: false, ctrlKey: false } as const; @@ -57,6 +60,15 @@ export function useOpenLink(threadRef: ScopedThreadRef | null | undefined): ( } const failure = squashAtomCommandFailure(result); if (failure instanceof BrowserSettingsReadError) throw failure; + // The system browser would load this machine's port, not the remote one. + if (isSshPreviewForwardError(failure)) { + toastManager.add({ + type: "error", + title: "Could not reach the remote port", + description: failure.message, + }); + return; + } console.error(result.cause); } const api = readLocalApi(); diff --git a/apps/web/src/components/preview/PreviewAutomationHosts.tsx b/apps/web/src/components/preview/PreviewAutomationHosts.tsx index 0a4b60be391f..a176370d10cc 100644 --- a/apps/web/src/components/preview/PreviewAutomationHosts.tsx +++ b/apps/web/src/components/preview/PreviewAutomationHosts.tsx @@ -29,6 +29,7 @@ import { applyPreviewServerSnapshot, readThreadPreviewState, reconcilePreviewServerSessions, + settleOpenedPreviewForward, updatePreviewServerSnapshot, useActivePreviewSessions, } from "~/previewStateStore"; @@ -38,6 +39,7 @@ import { usePreviewMiniPlayerStore, } from "~/previewMiniPlayerStore"; import { resolveBrowserNavigationTarget } from "~/browser/browserTargetResolver"; +import { acquirePreviewForward, navigateTabThroughForward } from "~/browser/sshPreviewForwards"; import { readActiveBrowserRecordingTargets, startBrowserRecording, @@ -453,11 +455,14 @@ function PreviewAutomationHost(props: { readonly environmentId: EnvironmentId }) tabId = activeTabId; if (!activeTabId) { const defaults = await resolveBrowserDefaults(); + const forward = resolvedInputUrl + ? await acquirePreviewForward(environmentId, resolvedInputUrl) + : null; const result = await open({ environmentId, input: { threadId: request.threadId, - ...(resolvedInputUrl ? { url: resolvedInputUrl } : {}), + ...(forward ? { url: forward.url } : {}), // An agent that didn't state a size gets the user's // configured default, same as a hand-opened tab. viewport: browserDefaultOpenViewport(defaults), @@ -465,10 +470,12 @@ function PreviewAutomationHost(props: { readonly environmentId: EnvironmentId }) }, }); if (result._tag === "Failure") { + settleOpenedPreviewForward(threadRef, forward, null); return raiseAtomCommandFailure(result); } const snapshot = result.value; applyPreviewServerSnapshot(threadRef, snapshot); + settleOpenedPreviewForward(threadRef, forward, snapshot.tabId); activeTabId = snapshot.tabId; activeSnapshot = snapshot; tabId = activeTabId; @@ -557,7 +564,16 @@ function PreviewAutomationHost(props: { readonly environmentId: EnvironmentId }) } if (reusedExistingTab && resolvedInputUrl && previewBridge) { assertPreviewRuntimeCurrent(threadRef, activeTabId, activeRuntimeTabId, request); - await previewBridge.navigate(activeRuntimeTabId, resolvedInputUrl); + const bridge = previewBridge; + const loadedUrl = await navigateTabThroughForward({ + threadRef, + tabId: activeTabId, + url: resolvedInputUrl, + navigate: (url) => bridge.navigate(activeRuntimeTabId, url), + }); + if (loadedUrl === null) { + throw new Error("A newer navigation of this preview tab replaced this one."); + } await waitForNavigationReadiness( threadRef, request.requestId, @@ -580,7 +596,15 @@ function PreviewAutomationHost(props: { readonly environmentId: EnvironmentId }) url: input.url!, }, ); - await ready.bridge.navigate(ready.runtimeTabId, resolution.resolvedUrl); + const loadedUrl = await navigateTabThroughForward({ + threadRef, + tabId: ready.tabId, + url: resolution.resolvedUrl, + navigate: (url) => ready.bridge.navigate(ready.runtimeTabId, url), + }); + if (loadedUrl === null) { + throw new Error("A newer navigation of this preview tab replaced this one."); + } await waitForNavigationReadiness( threadRef, request.requestId, diff --git a/apps/web/src/components/preview/PreviewView.tsx b/apps/web/src/components/preview/PreviewView.tsx index 810e6d502804..c05e90edc5a1 100644 --- a/apps/web/src/components/preview/PreviewView.tsx +++ b/apps/web/src/components/preview/PreviewView.tsx @@ -31,6 +31,13 @@ import { useThreadPreviewState, } from "~/previewStateStore"; import { resolveDiscoveredServerUrl } from "~/browser/browserTargetResolver"; +import { + isSshPreviewForwardError, + navigateTabThroughForward, + readSshEnvironmentTarget, + type SshPreviewForwardError, + toRemotePreviewUrl, +} from "~/browser/sshPreviewForwards"; import { useEnvironmentHttpBaseUrl } from "~/state/environments"; import { previewEnvironment } from "~/state/preview"; import { useAtomCommand } from "~/state/use-atom-command"; @@ -97,6 +104,14 @@ const localApi = typeof window === "undefined" ? null : ensureLocalApi(); * Single-tab preview surface: chrome row on top, one webview below, empty * state when no session exists for the thread. */ +const reportForwardFailure = (error: SshPreviewForwardError): void => { + toastManager.add({ + type: "error", + title: "Could not reach the remote port", + description: error.message, + }); +}; + export function PreviewView({ threadRef, tabId: requestedTabId, @@ -187,9 +202,21 @@ export function PreviewView({ const navigateToResolvedUrl = useCallback( async (resolvedUrl: string) => { - if (runtimeTabId && previewBridge) { - // The bridge mirrors the resolved URL back to the server. - await previewBridge.navigate(runtimeTabId, resolvedUrl); + if (tabId && runtimeTabId && previewBridge) { + const bridge = previewBridge; + // The bridge mirrors the loaded URL back to the server. + const loadedUrl = await navigateTabThroughForward({ + threadRef, + tabId, + url: resolvedUrl, + navigate: (url) => bridge.navigate(runtimeTabId, url), + }).catch((error: unknown) => { + // A forward fails before navigation, so no server `failed` event follows. + if (!isSshPreviewForwardError(error)) throw error; + reportForwardFailure(error); + return null; + }); + if (loadedUrl === null) return false; rememberPreviewUrl(threadRef, resolvedUrl); return true; } @@ -202,11 +229,13 @@ export function PreviewView({ title: "Unable to open browser", description: error.message, }); + } else if (isSshPreviewForwardError(error)) { + reportForwardFailure(error); } } return result._tag === "Success"; }, - [open, runtimeTabId, threadRef], + [open, runtimeTabId, tabId, threadRef], ); const handleSubmitUrl = useCallback( @@ -238,8 +267,23 @@ export function PreviewView({ ); const handleRefresh = useCallback(() => { + // An SSH tab's forward may have died with a disconnect; re-navigating + // through a fresh lease recovers it, and is a reload when it is alive. + const currentUrl = navStatus._tag === "Idle" ? null : navStatus.url; + if (currentUrl && readSshEnvironmentTarget(threadRef.environmentId) !== null) { + void navigateToResolvedUrl(toRemotePreviewUrl(threadRef.environmentId, currentUrl)).catch( + (error: unknown) => { + toastManager.add({ + type: "error", + title: "Unable to reload preview", + description: error instanceof Error ? error.message : "An error occurred.", + }); + }, + ); + return; + } if (previewBridge && runtimeTabId) void previewBridge.refresh(runtimeTabId); - }, [runtimeTabId]); + }, [navStatus, navigateToResolvedUrl, runtimeTabId, threadRef]); const handleZoomIn = useCallback(() => { if (previewBridge && runtimeTabId) void previewBridge.zoomIn(runtimeTabId); diff --git a/apps/web/src/components/preview/addBrowserSurface.ts b/apps/web/src/components/preview/addBrowserSurface.ts index e0cd83501201..212a47fdb8c3 100644 --- a/apps/web/src/components/preview/addBrowserSurface.ts +++ b/apps/web/src/components/preview/addBrowserSurface.ts @@ -5,6 +5,7 @@ import { import type { ScopedThreadRef } from "@t3tools/contracts"; import type { BrowserSettingsReadError, OpenPreviewMutation } from "~/browser/openFileInPreview"; +import type { SshPreviewForwardError } from "~/browser/sshPreviewForwards"; import { useRightPanelStore } from "~/rightPanelStore"; import { openPreviewSession } from "./openPreviewSession"; @@ -15,7 +16,7 @@ export async function addBrowserSurface(input: { readonly openPreview: OpenPreviewMutation; /** Omit to use the configured default profile. */ readonly profileId?: string | undefined; -}): Promise> { +}): Promise> { const result = await openPreviewSession({ openPreview: input.openPreview, threadRef: input.threadRef, diff --git a/apps/web/src/components/preview/openDiscoveredPort.ts b/apps/web/src/components/preview/openDiscoveredPort.ts index 288db101e7a5..7a3e9952e30f 100644 --- a/apps/web/src/components/preview/openDiscoveredPort.ts +++ b/apps/web/src/components/preview/openDiscoveredPort.ts @@ -6,6 +6,7 @@ import { import { resolveDiscoveredServerUrl } from "~/browser/browserTargetResolver"; import type { BrowserSettingsReadError, OpenPreviewMutation } from "~/browser/openFileInPreview"; +import type { SshPreviewForwardError } from "~/browser/sshPreviewForwards"; import { recordVisitForThread } from "~/browserHistoryStore"; import { useRightPanelStore } from "~/rightPanelStore"; import { openPreviewSession } from "./openPreviewSession"; @@ -14,7 +15,7 @@ export async function openDiscoveredPort(input: { readonly threadRef: ScopedThreadRef; readonly port: DiscoveredLocalServer; readonly openPreview: OpenPreviewMutation; -}): Promise> { +}): Promise> { const resolvedUrl = resolveDiscoveredServerUrl(input.threadRef.environmentId, input.port.url); const result = await openPreviewSession({ openPreview: input.openPreview, diff --git a/apps/web/src/components/preview/openPreviewSession.ts b/apps/web/src/components/preview/openPreviewSession.ts index 1b2f31a85da8..850a4f56cfa0 100644 --- a/apps/web/src/components/preview/openPreviewSession.ts +++ b/apps/web/src/components/preview/openPreviewSession.ts @@ -15,7 +15,17 @@ import { resolveBrowserDefaults, } from "~/browser/browserDefaults"; import { BrowserSettingsReadError } from "~/browser/openFileInPreview"; -import { applyPreviewServerSnapshot, rememberPreviewUrl } from "~/previewStateStore"; +import { + acquirePreviewForward, + type PreviewForward, + isSshPreviewForwardError, + type SshPreviewForwardError, +} from "~/browser/sshPreviewForwards"; +import { + applyPreviewServerSnapshot, + rememberPreviewUrl, + settleOpenedPreviewForward, +} from "~/previewStateStore"; interface OpenPreviewSessionInput { openPreview: (input: { @@ -32,7 +42,9 @@ interface OpenPreviewSessionInput { export async function openPreviewSession( input: OpenPreviewSessionInput, -): Promise> { +): Promise< + AtomCommandResult +> { // Resolved once: a tab opened before client settings hydrate would otherwise // be born at the schema defaults and never corrected. const defaults = await resolveBrowserDefaults().catch( @@ -41,20 +53,31 @@ export async function openPreviewSession( if (defaults instanceof BrowserSettingsReadError) { return AsyncResult.failure(Cause.fail(defaults)); } + let forward: PreviewForward | null = null; + if (input.url !== undefined) { + try { + forward = await acquirePreviewForward(input.threadRef.environmentId, input.url); + } catch (error) { + if (!isSshPreviewForwardError(error)) throw error; + return AsyncResult.failure(Cause.fail(error)); + } + } const result = await input.openPreview({ environmentId: input.threadRef.environmentId, input: { threadId: input.threadRef.threadId, - ...(input.url === undefined ? {} : { url: input.url }), + ...(forward === null ? {} : { url: forward.url }), viewport: input.viewport ?? browserDefaultOpenViewport(defaults), profileId: input.profileId ?? browserDefaultOpenProfileId(defaults), }, }); if (result._tag === "Failure") { + settleOpenedPreviewForward(input.threadRef, forward, null); return result; } const snapshot = result.value; applyPreviewServerSnapshot(input.threadRef, snapshot); + settleOpenedPreviewForward(input.threadRef, forward, snapshot.tabId); if (input.url !== undefined) { rememberPreviewUrl( input.threadRef, diff --git a/apps/web/src/components/preview/openTerminalLinkInPreview.test.ts b/apps/web/src/components/preview/openTerminalLinkInPreview.test.ts index 34c7192cd0e2..593d94e13ea4 100644 --- a/apps/web/src/components/preview/openTerminalLinkInPreview.test.ts +++ b/apps/web/src/components/preview/openTerminalLinkInPreview.test.ts @@ -11,8 +11,18 @@ import { vi.mock("~/previewStateStore", () => ({ applyPreviewServerSnapshot: vi.fn(), isPreviewSupportedInRuntime: () => true, + settleOpenedPreviewForward: vi.fn(), })); +const forwardMocks = vi.hoisted(() => ({ acquire: vi.fn() })); +const toastMocks = vi.hoisted(() => ({ add: vi.fn() })); + +vi.mock("~/browser/sshPreviewForwards", async (importOriginal) => ({ + ...(await importOriginal()), + acquirePreviewForward: forwardMocks.acquire, +})); +vi.mock("~/components/ui/toast", () => ({ toastManager: { add: toastMocks.add } })); + vi.mock("~/rightPanelStore", () => ({ useRightPanelStore: { getState: () => ({ openBrowser: vi.fn() }), @@ -61,6 +71,12 @@ beforeEach(() => { browserDefaultsMocks.resolve.mockReset(); browserDefaultsMocks.resolve.mockResolvedValue(hydratedDefaults); linkTargetMocks.preference.mockReturnValue("app"); + forwardMocks.acquire.mockReset(); + forwardMocks.acquire.mockImplementation(async (_environmentId: string, url: string) => ({ + url, + leaseId: null, + })); + toastMocks.add.mockReset(); }); afterEach(() => { @@ -96,6 +112,31 @@ describe("openTerminalLinkInPreview", () => { }, ); + it("reports a failed SSH forward instead of opening the local port externally", async () => { + const { SshPreviewForwardError } = await import("~/browser/sshPreviewForwards"); + forwardMocks.acquire.mockRejectedValueOnce( + new SshPreviewForwardError({ remotePort: 5173, detail: "Connection refused" }), + ); + const fallbackToBrowser = vi.fn(); + const openPreview = vi.fn(async () => AsyncResult.success(snapshot)); + + await openTerminalLinkInPreview({ + url: "http://localhost:5173/", + threadRef, + openPreview, + fallbackToBrowser, + forceBrowser: false, + }); + + expect(fallbackToBrowser).not.toHaveBeenCalled(); + expect(openPreview).not.toHaveBeenCalled(); + expect(toastMocks.add).toHaveBeenCalledWith( + expect.objectContaining({ + description: "Could not forward remote port 5173: Connection refused", + }), + ); + }); + it("opens in the system browser while that is the configured target", async () => { linkTargetMocks.preference.mockReturnValue("system"); const fallbackToBrowser = vi.fn(); diff --git a/apps/web/src/components/preview/openTerminalLinkInPreview.ts b/apps/web/src/components/preview/openTerminalLinkInPreview.ts index 23ef35a87829..5a314cf28ac3 100644 --- a/apps/web/src/components/preview/openTerminalLinkInPreview.ts +++ b/apps/web/src/components/preview/openTerminalLinkInPreview.ts @@ -9,8 +9,18 @@ import { } from "~/browser/browserDefaults"; import { isWebUrl, resolveBrowserLinkTargetPreference } from "~/browser/browserLinkTarget"; import type { OpenPreviewMutation } from "~/browser/openFileInPreview"; +import { + acquirePreviewForward, + isSshPreviewForwardError, + type PreviewForward, +} from "~/browser/sshPreviewForwards"; +import { toastManager } from "~/components/ui/toast"; import { recordVisitForThread } from "~/browserHistoryStore"; -import { applyPreviewServerSnapshot, isPreviewSupportedInRuntime } from "~/previewStateStore"; +import { + applyPreviewServerSnapshot, + isPreviewSupportedInRuntime, + settleOpenedPreviewForward, +} from "~/previewStateStore"; import { useRightPanelStore } from "~/rightPanelStore"; const terminalLinkErrorContext = { @@ -64,11 +74,28 @@ export async function openTerminalLinkInPreview( }; const defaults = await resolveBrowserDefaults(); + let forward: PreviewForward; + try { + forward = await acquirePreviewForward(input.threadRef.environmentId, input.url); + } catch (cause) { + // The system browser would load this machine's port, not the remote one. + if (isSshPreviewForwardError(cause)) { + toastManager.add({ + type: "error", + title: "Could not reach the remote port", + description: cause.message, + }); + return; + } + console.error(new TerminalLinkPreviewOpenError({ ...errorContext, cause })); + input.fallbackToBrowser(); + return; + } const result = await input.openPreview({ environmentId: input.threadRef.environmentId, input: { threadId: input.threadRef.threadId, - url: input.url, + url: forward.url, // Same reason as `openUrlInPreview`: this path handles its own result // mapping, so the configured defaults are applied explicitly. viewport: browserDefaultOpenViewport(defaults), @@ -76,6 +103,7 @@ export async function openTerminalLinkInPreview( }, }); if (result._tag === "Failure") { + settleOpenedPreviewForward(input.threadRef, forward, null); if (isAtomCommandInterrupted(result)) { return; } @@ -90,5 +118,6 @@ export async function openTerminalLinkInPreview( } recordVisitForThread(input.threadRef, input.url); applyPreviewServerSnapshot(input.threadRef, result.value); + settleOpenedPreviewForward(input.threadRef, forward, result.value.tabId); useRightPanelStore.getState().openBrowser(input.threadRef, result.value.tabId); } diff --git a/apps/web/src/previewStateStore.test.ts b/apps/web/src/previewStateStore.test.ts index f9956f9c02c8..93db06727fba 100644 --- a/apps/web/src/previewStateStore.test.ts +++ b/apps/web/src/previewStateStore.test.ts @@ -5,7 +5,19 @@ import { type PreviewSessionSnapshot, ThreadId, } from "@t3tools/contracts"; -import { beforeEach, describe, expect, it } from "vite-plus/test"; +import { beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +const { releaseTabForward, settleOpenedForward } = vi.hoisted(() => ({ + releaseTabForward: vi.fn(), + settleOpenedForward: vi.fn(), +})); +// The forward on local port 53001 carries remote port 5173. +vi.mock("./browser/sshPreviewForwards", () => ({ + releaseTabForward, + settleOpenedForward, + toRemotePreviewUrl: (_environmentId: string, url: string) => + url.replace("localhost:53001", "localhost:5173"), +})); import { __testing, @@ -19,6 +31,7 @@ import { reconcilePreviewServerSessions, rememberPreviewUrl, resetPreviewStateForTests, + settleOpenedPreviewForward, setActivePreviewTab, updatePreviewServerSnapshot, } from "./previewStateStore"; @@ -227,6 +240,41 @@ describe("previewStateStore (single-tab)", () => { expect(state.recentlySeenUrls).toContain("http://localhost:5173/"); }); + it("releases the SSH forward of every tab that leaves the sessions", () => { + releaseTabForward.mockClear(); + applyPreviewServerSnapshot(ref, makeSnapshot({ tabId: "tab_a" })); + applyPreviewServerSnapshot(ref, makeSnapshot({ tabId: "tab_b" })); + beginPreviewSessionClose(ref, "tab_a"); + expect(releaseTabForward.mock.calls).toEqual([[ref, "tab_a"]]); + applyPreviewServerSnapshot(ref, null); + expect(releaseTabForward.mock.calls).toEqual([ + [ref, "tab_a"], + [ref, "tab_b"], + ]); + }); + + it("releases a forward whose tab closed while its open was in flight", () => { + settleOpenedForward.mockClear(); + const forward = { url: "http://localhost:5173/", leaseId: "lease-1" }; + applyPreviewServerSnapshot(ref, makeSnapshot({ tabId: "tab_open" })); + settleOpenedPreviewForward(ref, forward, "tab_open"); + settleOpenedPreviewForward(ref, forward, "tab_closed"); + expect(settleOpenedForward.mock.calls).toEqual([ + [ref, forward, "tab_open"], + [ref, forward, null], + ]); + }); + + it("remembers the remote URL of a forwarded SSH preview", () => { + applyPreviewServerSnapshot( + ref, + makeSnapshot({ + navStatus: { _tag: "Success", url: "http://localhost:53001/app", title: "" }, + }), + ); + expect(readThreadPreviewState(ref).recentlySeenUrls).toEqual(["http://localhost:5173/app"]); + }); + it("optimistically removes a session before the server close event arrives", () => { const first = makeSnapshot({ tabId: "tab_a" }); const second = makeSnapshot({ diff --git a/apps/web/src/previewStateStore.ts b/apps/web/src/previewStateStore.ts index 26be8c497740..3177b8dbd8ae 100644 --- a/apps/web/src/previewStateStore.ts +++ b/apps/web/src/previewStateStore.ts @@ -18,6 +18,12 @@ import { import { Atom } from "effect/reactivity"; import { PREVIEW_RECENT_URL_LIMIT } from "./components/preview/previewConstants"; +import { + type PreviewForward, + releaseTabForward, + settleOpenedForward, + toRemotePreviewUrl, +} from "./browser/sshPreviewForwards"; import { appAtomRegistry } from "./rpc/atomRegistry"; export interface DesktopPreviewOverlay { @@ -106,18 +112,37 @@ function syncActivePreviewThread(threadKey: string, state: ThreadPreviewState): }); } +/** + * Snapshot URLs of SSH previews carry the local forward port. Recents keep the + * remote URL so reopening one later forwards the right port. + */ +function withRemoteRecentUrls( + ref: ScopedThreadRef, + current: ThreadPreviewState, + next: ThreadPreviewState, +): ThreadPreviewState { + if (next.recentlySeenUrls === current.recentlySeenUrls) return next; + const mapped = next.recentlySeenUrls.map((url) => toRemotePreviewUrl(ref.environmentId, url)); + if (mapped.every((url, index) => url === next.recentlySeenUrls[index])) return next; + return { ...next, recentlySeenUrls: [...new Set(mapped)] }; +} + function updateThreadPreviewState( ref: ScopedThreadRef, update: (current: ThreadPreviewState) => ThreadPreviewState, ): void { const threadKey = scopedThreadKey(ref); const atom = previewStateAtom(threadKey); - let nextState = appAtomRegistry.get(atom); + const previous = appAtomRegistry.get(atom); + let nextState = previous; const changed = appAtomRegistry.modify(atom, (current) => { - nextState = update(current); + nextState = withRemoteRecentUrls(ref, current, update(current)); return [nextState !== current, nextState]; }); if (!changed) return; + for (const tabId of Object.keys(previous.sessions)) { + if (!nextState.sessions[tabId]) releaseTabForward(ref, tabId); + } changedPreviewThreadKeys.add(threadKey); syncActivePreviewThread(threadKey, nextState); } @@ -451,6 +476,19 @@ export function setActivePreviewTab(ref: ScopedThreadRef, tabId: string): void { }); } +/** + * Gives an opened tab its forward. A tab closed while its open was in flight is + * already gone from the sessions, so the forward is released instead. + */ +export function settleOpenedPreviewForward( + ref: ScopedThreadRef, + forward: PreviewForward | null, + tabId: string | null, +): void { + const open = tabId !== null && readThreadPreviewState(ref).sessions[tabId] !== undefined; + settleOpenedForward(ref, forward, open ? tabId : null); +} + export function rememberPreviewUrl(ref: ScopedThreadRef, url: string): void { if (url.trim().length === 0) return; updateThreadPreviewState(ref, (current) => ({ diff --git a/packages/contracts/src/ipc.ts b/packages/contracts/src/ipc.ts index db2cc2a36cac..c78854b9ee25 100644 --- a/packages/contracts/src/ipc.ts +++ b/packages/contracts/src/ipc.ts @@ -448,6 +448,26 @@ export const DesktopSshEnvironmentEnsureResultSchema = Schema.Union([ DesktopSshPasswordPromptCancelledResultSchema, ]); +export const DesktopSshPortForwardSchema = Schema.Struct({ + leaseId: Schema.String, + localPort: Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 65535 })), +}); +export type DesktopSshPortForward = typeof DesktopSshPortForwardSchema.Type; + +export const DesktopSshPortForwardAcquireInputSchema = Schema.Struct({ + target: DesktopSshEnvironmentTargetSchema, + remotePort: Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 65535 })), +}); + +export const DesktopSshPortForwardAcquireResultSchema = Schema.Union([ + DesktopSshPortForwardSchema, + DesktopSshPasswordPromptCancelledResultSchema, +]); + +export const DesktopSshPortForwardReleaseInputSchema = Schema.Struct({ + leaseId: Schema.String, +}); + export const DesktopSshHttpBaseUrlInputSchema = Schema.Struct({ httpBaseUrl: Schema.String, }); @@ -1175,6 +1195,11 @@ export interface DesktopBridge { options?: { issuePairingToken?: boolean }, ) => Promise; disconnectSshEnvironment: (target: DesktopSshEnvironmentTarget) => Promise; + acquireSshPortForward: ( + target: DesktopSshEnvironmentTarget, + remotePort: number, + ) => Promise; + releaseSshPortForward: (leaseId: string) => Promise; fetchSshEnvironmentDescriptor: (httpBaseUrl: string) => Promise; bootstrapSshBearerSession: ( httpBaseUrl: string, diff --git a/packages/contracts/src/previewAutomation.ts b/packages/contracts/src/previewAutomation.ts index 734aa1bb53df..e9d8bc955c74 100644 --- a/packages/contracts/src/previewAutomation.ts +++ b/packages/contracts/src/previewAutomation.ts @@ -948,7 +948,7 @@ export type PreviewAutomationError = typeof PreviewAutomationError.Type; export const PreviewUrlResolution = Schema.Struct({ requestedUrl: Schema.String, resolvedUrl: Schema.String, - resolutionKind: Schema.Literals(["direct", "direct-private-network"]), + resolutionKind: Schema.Literals(["direct", "direct-private-network", "ssh-forward"]), environmentId: EnvironmentId, }); export type PreviewUrlResolution = typeof PreviewUrlResolution.Type; diff --git a/packages/ssh/src/portForward.test.ts b/packages/ssh/src/portForward.test.ts new file mode 100644 index 000000000000..b9be84e2a5db --- /dev/null +++ b/packages/ssh/src/portForward.test.ts @@ -0,0 +1,565 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as NetService from "@t3tools/shared/Net"; +import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Scope from "effect/Scope"; +import * as Sink from "effect/Sink"; +import * as Stream from "effect/Stream"; +import { TestClock } from "effect/testing"; +import { HttpClient, HttpClientResponse } from "effect/http"; +import { ChildProcessSpawner } from "effect/process"; + +import * as SshAuth from "./auth.ts"; +import * as SshTunnel from "./tunnel.ts"; +import { SshCommandError, SshReadinessError } from "./errors.ts"; + +const target = { + alias: "devbox", + hostname: "devbox.example.test", + username: "developer", + port: 22, +}; +const ready = + "debug1: Local forwarding listening on 127.0.0.1 port 41773.\ndebug1: Entering interactive session.\n"; +const bytes = (text: string) => new TextEncoder().encode(text); + +const fixture = ( + options: { + readonly stderr?: Stream.Stream; + readonly running?: Effect.Effect; + readonly spawnGate?: Effect.Effect; + readonly stopGate?: Effect.Effect; + readonly exitCode?: Effect.Effect; + readonly preferredAvailable?: boolean; + readonly portInUseAfterProbe?: number; + readonly collisionStderr?: Stream.Stream; + readonly listening?: boolean; + /** A local server holds the preferred port on ::1 only. */ + readonly ipv6Listener?: boolean; + } = {}, +) => { + const commands: ReadonlyArray[] = []; + let killed = 0; + const ipv4Available = (port: number) => + options.portInUseAfterProbe === port && + commands.some( + (args) => args.includes("-N") && args.some((arg) => arg.startsWith(`127.0.0.1:${port}:`)), + ) + ? false + : (options.preferredAvailable ?? false); + let nextPort = 41773; + const spawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + const args = command._tag === "StandardCommand" ? command.args : []; + commands.push(args); + const tunnel = args.includes("-N"); + const bindCollision = + tunnel && + options.portInUseAfterProbe !== undefined && + args.some((arg) => arg.startsWith(`127.0.0.1:${options.portInUseAfterProbe}:`)); + if (tunnel && options.spawnGate) yield* options.spawnGate; + if (!tunnel && !args.includes("-G") && options.stopGate) yield* options.stopGate; + const stopped = yield* Deferred.make(); + const stdout = Stream.make(bytes("hostname devbox.example.test\nuser developer\nport 22\n")); + return ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(123), + stdout, + stderr: bindCollision + ? (options.collisionStderr ?? + Stream.make( + bytes( + `debug1: Local forwarding listening on 127.0.0.1 port ${options.portInUseAfterProbe}.\nbind [127.0.0.1]:${options.portInUseAfterProbe}: Address already in use\n`, + ), + )) + : tunnel + ? (options.stderr ?? Stream.make(bytes(ready))) + : Stream.empty, + all: stdout, + exitCode: bindCollision + ? Effect.succeed(ChildProcessSpawner.ExitCode(255)) + : tunnel + ? (options.exitCode ?? Deferred.await(stopped)) + : Effect.succeed(ChildProcessSpawner.ExitCode(0)), + isRunning: bindCollision + ? Effect.succeed(false) + : tunnel + ? (options.running ?? Effect.succeed(true)) + : Effect.succeed(false), + kill: () => + Effect.sync(() => { + killed++; + }).pipe( + Effect.andThen(Deferred.succeed(stopped, ChildProcessSpawner.ExitCode(143))), + Effect.asVoid, + ), + stdin: Sink.drain, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + unref: Effect.succeed(Effect.void), + }); + }), + ); + const layer = Layer.mergeAll( + NodeServices.layer, + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner), + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.succeed(HttpClientResponse.fromWeb(request, new Response("", { status: 200 }))), + ), + ), + Layer.succeed( + NetService.NetService, + NetService.NetService.of({ + canListenOnHost: (port) => Effect.succeed(ipv4Available(port)), + isPortAvailableOnLoopback: (port) => + Effect.succeed(ipv4Available(port) && options.ipv6Listener !== true), + hasListenerOnHost: () => Effect.succeed(options.listening ?? true), + reserveLoopbackPort: () => Effect.sync(() => nextPort++), + findAvailablePort: (port) => Effect.succeed(port), + }), + ), + SshAuth.SshPasswordPrompt.disabledLayer, + SshTunnel.SshEnvironmentManager.layer(), + ); + return { + layer, + commands, + kills: () => killed, + spawns: () => commands.filter((args) => args.includes("-N")).length, + }; +}; + +describe("SSH preview port forwards", () => { + it.effect("shares concurrent acquires, releases idempotently, and binds only loopback", () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const [a, b] = yield* Effect.all( + [manager.acquirePortForward(target, 5173), manager.acquirePortForward(target, 5173)], + { concurrency: "unbounded" }, + ); + assert.equal(a.localPort, b.localPort); + assert.notEqual(a.leaseId, b.leaseId); + assert.equal(f.spawns(), 1); + assert.include( + f.commands.find((args) => args.includes("-N")) ?? [], + "127.0.0.1:41773:localhost:5173", + ); + yield* manager.releasePortForward(a.leaseId); + yield* manager.releasePortForward(a.leaseId); + assert.equal(f.kills(), 0); + yield* manager.releasePortForward(b.leaseId); + assert.equal(f.kills(), 1); + const c = yield* manager.acquirePortForward(target, 5173); + assert.equal(f.spawns(), 2); + yield* manager.releasePortForward(c.leaseId); + yield* manager.releasePortForward("unknown"); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect.each([ + { available: true, remotePort: 5173, localPort: 5173 }, + { available: false, remotePort: 5173, localPort: 41773 }, + { available: true, remotePort: 80, localPort: 41773 }, + ])( + "prefers remote port $remotePort when available=$available, using local $localPort", + ({ available, remotePort, localPort }) => { + const f = fixture({ preferredAvailable: available }); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const lease = yield* manager.acquirePortForward(target, remotePort); + assert.equal(lease.localPort, localPort); + assert.equal(f.spawns(), 1); + assert.include( + f.commands.find((args) => args.includes("-N")) ?? [], + `127.0.0.1:${localPort}:localhost:${remotePort}`, + ); + yield* manager.releasePortForward(lease.leaseId); + assert.equal(f.kills(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }, + ); + + it.effect("falls back when a local server holds the preferred port on ::1 only", () => { + const f = fixture({ preferredAvailable: true, ipv6Listener: true }); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const lease = yield* manager.acquirePortForward(target, 5173); + assert.equal(lease.localPort, 41773); + assert.equal(f.spawns(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect("falls back once if another listener takes the preferred port after the probe", () => { + const f = fixture({ preferredAvailable: true, portInUseAfterProbe: 5173 }); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const lease = yield* manager.acquirePortForward(target, 5173); + assert.equal(lease.localPort, 41773); + assert.equal(f.spawns(), 2); + assert.equal(f.kills(), 1); + const commands = f.commands.filter((args) => args.includes("-N")); + assert.include(commands[0] ?? [], "127.0.0.1:5173:localhost:5173"); + assert.include(commands[1] ?? [], "127.0.0.1:41773:localhost:5173"); + yield* manager.releasePortForward(lease.leaseId); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect("preferred-port fallback does not depend on delivery of the final stderr chunk", () => { + const f = fixture({ + preferredAvailable: true, + portInUseAfterProbe: 5173, + collisionStderr: Stream.fromEffect(Effect.never), + }); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const lease = yield* manager.acquirePortForward(target, 5173); + assert.equal(lease.localPort, 41773); + assert.equal(f.spawns(), 2); + assert.equal(f.kills(), 1); + yield* manager.releasePortForward(lease.leaseId); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect("rejects a live child with the session marker but no loopback listener", () => { + const f = fixture({ listening: false }); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const error = yield* manager.acquirePortForward(target, 5173).pipe(Effect.flip); + assert.instanceOf(error, SshReadinessError); + assert.equal(error.message, "SSH forward stopped before becoming ready."); + assert.equal(f.kills(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect("remote stop stays interruptible after local forwards are torn down", () => + Effect.gen(function* () { + const stopping = yield* Deferred.make(); + const f = fixture({ + stopGate: Deferred.succeed(stopping, undefined).pipe(Effect.andThen(Effect.never)), + }); + yield* Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const old = yield* manager.acquirePortForward(target, 5173); + const disconnect = yield* Effect.forkChild(manager.disconnectEnvironment(target)); + yield* Deferred.await(stopping); + yield* Fiber.interrupt(disconnect); + assert.isTrue(Exit.isFailure(yield* Fiber.await(disconnect))); + assert.equal(f.kills(), 1); + yield* manager.releasePortForward(old.leaseId); + const fresh = yield* manager.acquirePortForward(target, 5173); + yield* manager.releasePortForward(fresh.leaseId); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }), + ); + + it.effect("keeps different remote ports independent", () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const a = yield* manager.acquirePortForward(target, 5173); + const b = yield* manager.acquirePortForward(target, 8080); + assert.notEqual(a.localPort, b.localPort); + yield* manager.releasePortForward(a.leaseId); + assert.equal(f.kills(), 1); + yield* manager.releasePortForward(b.leaseId); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect("serializes the last release against an in-flight reuse check", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const resume = yield* Deferred.make(); + let checks = 0; + const f = fixture({ + running: Effect.suspend(() => + ++checks === 1 + ? Effect.succeed(true) + : Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + Effect.as(true), + ), + ), + }); + yield* Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const a = yield* manager.acquirePortForward(target, 5173); + const acquire = yield* Effect.forkChild(manager.acquirePortForward(target, 5173)); + yield* Deferred.await(entered); + const release = yield* Effect.forkChild(manager.releasePortForward(a.leaseId)); + yield* Deferred.succeed(resume, undefined); + const b = yield* Fiber.join(acquire); + yield* Fiber.join(release); + assert.equal(f.kills(), 0); + yield* manager.releasePortForward(b.leaseId); + assert.equal(f.kills(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }), + ); + + it.effect("disconnect tears down acquired forwards and tolerates late releases", () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const a = yield* manager.acquirePortForward(target, 5173); + const b = yield* manager.acquirePortForward(target, 8080); + yield* manager.disconnectEnvironment(target); + assert.equal(f.kills(), 2); + yield* manager.releasePortForward(a.leaseId); + yield* manager.releasePortForward(b.leaseId); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect.each(["disconnect", "scope close", "interrupt"] as const)( + "%s cancels pending creation and closes its child", + (action) => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const f = fixture({ + stderr: Stream.fromEffect( + Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)), + ), + }); + const scope = yield* Scope.make("sequential"); + const context = yield* Layer.buildWithScope(f.layer, scope); + const manager = Context.get(context, SshTunnel.SshEnvironmentManager); + const acquire = yield* Effect.forkChild( + manager.acquirePortForward(target, 5173).pipe(Effect.provide(context)), + ); + yield* Deferred.await(entered); + if (action === "disconnect") + yield* manager.disconnectEnvironment(target).pipe(Effect.provide(context)); + else if (action === "scope close") yield* Scope.close(scope, Exit.void); + else yield* Fiber.interrupt(acquire); + const exit = yield* Fiber.await(acquire); + assert.isTrue(Exit.isFailure(exit)); + assert.equal(f.kills(), 1); + yield* Scope.close(scope, Exit.void); + assert.equal(f.kills(), 1); + }), + ); + + it.effect("scope close interrupts creation before the spawner returns a child", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const f = fixture({ + spawnGate: Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)), + }); + const scope = yield* Scope.make("sequential"); + const context = yield* Layer.buildWithScope(f.layer, scope); + const manager = Context.get(context, SshTunnel.SshEnvironmentManager); + const acquire = yield* Effect.forkChild( + manager.acquirePortForward(target, 5173).pipe(Effect.provide(context)), + ); + yield* Deferred.await(entered); + yield* Scope.close(scope, Exit.void); + assert.isTrue(Exit.isFailure(yield* Fiber.await(acquire))); + assert.equal(f.kills(), 0); + }), + ); + + it.effect("disconnect rejects acquires begun while remote stop is in flight", () => + Effect.gen(function* () { + const stopping = yield* Deferred.make(); + const stopped = yield* Deferred.make(); + const f = fixture({ + stopGate: Deferred.succeed(stopping, undefined).pipe( + Effect.andThen(Deferred.await(stopped)), + ), + }); + yield* Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + yield* manager.acquirePortForward(target, 5173); + const disconnect = yield* Effect.forkChild(manager.disconnectEnvironment(target)); + yield* Deferred.await(stopping); + const acquire = yield* Effect.forkChild(manager.acquirePortForward(target, 5173)); + const result = yield* Fiber.await(acquire); + yield* Deferred.succeed(stopped, undefined); + yield* Fiber.join(disconnect); + assert.isTrue(Exit.isFailure(result)); + assert.equal(f.spawns(), 1); + const fresh = yield* manager.acquirePortForward(target, 5173); + yield* manager.releasePortForward(fresh.leaseId); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }), + ); + + it.effect("continues draining child stderr after returning a lease", () => + Effect.gen(function* () { + const write = yield* Deferred.make(); + const drained = yield* Deferred.make(); + const f = fixture({ + stderr: Stream.make(bytes(ready)).pipe( + Stream.concat( + Stream.fromEffect( + Deferred.await(write).pipe( + Effect.andThen(Deferred.succeed(drained, undefined)), + Effect.as(bytes("debug1: Connection established.\n")), + ), + ), + ), + ), + }); + yield* Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const lease = yield* manager.acquirePortForward(target, 5173); + yield* Deferred.succeed(write, undefined); + yield* Deferred.await(drained); + yield* manager.releasePortForward(lease.leaseId); + assert.equal(f.kills(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }), + ); + + it.effect("replaces a dead child rather than returning a stale lease", () => { + let running = true; + const f = fixture({ running: Effect.sync(() => running) }); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const old = yield* manager.acquirePortForward(target, 5173); + running = false; + // The second spawn resets its readiness check to a live process. + // The old running check observes false, then the reserve call restores true. + const net = yield* NetService.NetService; + const replacement = yield* manager.acquirePortForward(target, 5173).pipe( + Effect.provideService(NetService.NetService, { + ...net, + reserveLoopbackPort: () => + Effect.sync(() => { + running = true; + return 41774; + }), + }), + ); + assert.notEqual(old.localPort, replacement.localPort); + assert.equal(f.spawns(), 2); + yield* manager.releasePortForward(old.leaseId); + assert.equal(f.kills(), 1); + yield* manager.releasePortForward(replacement.leaseId); + assert.equal(f.kills(), 2); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect("a foreign listener cannot prove readiness even after SSH's pre-bind log", () => { + const f = fixture({ + stderr: Stream.make( + bytes( + "debug1: Local forwarding listening on 127.0.0.1 port 41773.\nbind [127.0.0.1]:41773: Address already in use\n", + ), + ), + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(255)), + running: Effect.succeed(false), + }); + return Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const error = yield* manager.acquirePortForward(target, 5173).pipe(Effect.flip); + assert.instanceOf(error, SshCommandError); + assert.equal(f.kills(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }); + + it.effect("child exit fails immediately even while stderr remains open", () => + Effect.gen(function* () { + const exitCode = yield* Deferred.make(); + const drained = yield* Deferred.make(); + const f = fixture({ + stderr: Stream.make( + bytes("ssh: connect to host devbox port 22: Connection refused\n"), + ).pipe( + Stream.concat( + Stream.fromEffect( + Deferred.succeed(drained, undefined).pipe(Effect.andThen(Effect.never)), + ), + ), + ), + exitCode: Deferred.await(exitCode), + }); + yield* Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const acquire = yield* Effect.forkChild( + manager.acquirePortForward(target, 5173).pipe(Effect.flip), + ); + yield* Deferred.await(drained); + yield* Deferred.succeed(exitCode, ChildProcessSpawner.ExitCode(255)); + const error = yield* Fiber.join(acquire); + assert.instanceOf(error, SshCommandError); + if (error instanceof SshCommandError) { + assert.equal(error.exitCode, 255); + assert.include(error.stderr, "Connection refused"); + assert.include(error.message, "Connection refused"); + } + assert.equal(f.kills(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }), + ); + + it.effect("readiness timeout includes a bounded tail of the last stderr lines", () => + Effect.gen(function* () { + const drained = yield* Deferred.make(); + const diagnostic = + "FIRST-DIAGNOSTIC\n" + + "x".repeat(20_000) + + "\n" + + Array.from( + { length: 20 }, + (_, index) => `debug1: Waiting for authentication ${index}`, + ).join("\n") + + "\n"; + const f = fixture({ + stderr: Stream.make(bytes(diagnostic)).pipe( + Stream.concat( + Stream.fromEffect( + Deferred.succeed(drained, undefined).pipe(Effect.andThen(Effect.never)), + ), + ), + ), + }); + yield* Effect.gen(function* () { + const manager = yield* SshTunnel.SshEnvironmentManager; + const acquire = yield* Effect.forkChild( + manager.acquirePortForward(target, 5173).pipe(Effect.flip), + ); + yield* Deferred.await(drained); + yield* TestClock.adjust(20_000); + const error = yield* Fiber.join(acquire); + assert.instanceOf(error, SshReadinessError); + assert.include(error.message, "did not become ready"); + assert.include(error.message, "Waiting for authentication 19"); + assert.notInclude(error.message, "Waiting for authentication 11"); + assert.notInclude(error.message, "FIRST-DIAGNOSTIC"); + assert.isBelow(error.message.length, 16_500); + assert.equal(f.kills(), 1); + }).pipe(Effect.provide(f.layer), Effect.scoped); + }), + ); + + it.effect("manager close rejects future acquisitions", () => + Effect.gen(function* () { + const f = fixture(); + const scope = yield* Scope.make("sequential"); + const context = yield* Layer.buildWithScope(f.layer, scope); + const manager = Context.get(context, SshTunnel.SshEnvironmentManager); + const lease = yield* manager.acquirePortForward(target, 5173).pipe(Effect.provide(context)); + yield* Scope.close(scope, Exit.void); + assert.equal(f.kills(), 1); + const error = yield* manager + .acquirePortForward(target, 5173) + .pipe(Effect.provide(context), Effect.flip); + assert.instanceOf(error, SshReadinessError); + yield* manager.releasePortForward(lease.leaseId); + }), + ); +}); diff --git a/packages/ssh/src/tunnel.test.ts b/packages/ssh/src/tunnel.test.ts index 3cf8c62c5b50..f69b1717b315 100644 --- a/packages/ssh/src/tunnel.test.ts +++ b/packages/ssh/src/tunnel.test.ts @@ -545,6 +545,8 @@ describe("ssh tunnel scripts", () => { assert.equal(first.httpBaseUrl, "http://127.0.0.1:41773/"); const firstTunnelArgs = spawnedCommands.find((args) => args.includes("-N")); assert.isDefined(firstTunnelArgs); + assert.include(firstTunnelArgs, "-L"); + assert.include(firstTunnelArgs, "127.0.0.1:41773:127.0.0.1:3773"); assert.include(firstTunnelArgs, "ControlMaster=no"); assert.include(firstTunnelArgs, "ControlPath=none"); assert.include(firstTunnelArgs, "ControlPersist=no"); diff --git a/packages/ssh/src/tunnel.ts b/packages/ssh/src/tunnel.ts index c77fe13cb818..066d232bffa3 100644 --- a/packages/ssh/src/tunnel.ts +++ b/packages/ssh/src/tunnel.ts @@ -11,9 +11,11 @@ import * as KeyedLock from "@t3tools/shared/KeyedLock"; import * as NetService from "@t3tools/shared/Net"; import { extractJsonObject, fromLenientJson } from "@t3tools/shared/schemaJson"; import { satisfiesSemverRange } from "@t3tools/shared/semver"; +import * as Deferred from "effect/Deferred"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; @@ -96,6 +98,16 @@ interface SshTunnelEntry { readonly scope: Scope.Closeable; } +interface SshPortForwardEntry extends SshTunnelEntry { + readonly connectionKey: string; + readonly leases: Set; +} + +export interface SshPortForwardLease { + readonly leaseId: string; + readonly localPort: number; +} + type SshEnvironmentEffectContext = | ChildProcessSpawner.ChildProcessSpawner | FileSystem.FileSystem @@ -158,6 +170,11 @@ export interface SshEnvironmentManagerShape { SshEnvironmentEffectError, SshEnvironmentEffectContext >; + readonly acquirePortForward: ( + target: DesktopSshEnvironmentTarget, + remotePort: number, + ) => Effect.Effect; + readonly releasePortForward: (leaseId: string) => Effect.Effect; readonly disconnectEnvironment: ( target: DesktopSshEnvironmentTarget, ) => Effect.Effect; @@ -1105,6 +1122,8 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: readonly authOptions: SshAuth.SshAuthOptions; readonly remoteServerKind: "external" | "managed" | null; readonly scope: Scope.Closeable; + readonly readiness?: "forward"; + readonly remoteHost: "127.0.0.1" | "localhost"; }): Effect.fn.Return< SshTunnelEntry, SshCommandError | SshInvalidTargetError | SshReadinessError, @@ -1140,6 +1159,7 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: }), "-o", "ExitOnForwardFailure=yes", + ...(input.readiness === "forward" ? ["-o", "LogLevel=DEBUG1"] : []), "-o", "ControlMaster=no", "-o", @@ -1153,7 +1173,7 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: "-n", "-N", "-L", - `${input.localPort}:127.0.0.1:${input.remotePort}`, + `127.0.0.1:${input.localPort}:${input.remoteHost}:${input.remotePort}`, hostSpec, ]; const sshCommand = yield* resolveSshCommand; @@ -1168,7 +1188,7 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: remoteServerKind: input.remoteServerKind, httpBaseUrl: input.httpBaseUrl, }); - const child = yield* spawner + const spawn = spawner .spawn( ChildProcess.make(sshCommand, args, { env: childEnvironment, @@ -1195,6 +1215,24 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: }), ), ); + const child = + input.readiness === "forward" + ? yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const child = yield* restore(spawn); + yield* Scope.addFinalizer( + scope, + child + .kill({ + killSignal: "SIGTERM", + forceKillAfter: TUNNEL_SHUTDOWN_TIMEOUT_MS, + }) + .pipe(Effect.ignore), + ); + return child; + }), + ) + : yield* spawn; yield* Effect.logDebug("ssh.tunnel.spawn.succeeded", { ...sshTargetLogFields(input.resolvedTarget), command: tunnelCommand, @@ -1214,10 +1252,47 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: process: child, scope, }; - const exitFailure = Effect.all( - [collectProcessOutput(child.stderr), child.exitCode.pipe(Effect.map(Number))], - { concurrency: "unbounded" }, - ).pipe( + // OpenSSH logs "Local forwarding listening" before bind(). The session marker + // is emitted only after ExitOnForwardFailure has checked all local forwards. + const forwardReady = yield* Deferred.make(); + let stderrTail = ""; + if (input.readiness === "forward") { + // Drain independently: child exit must not wait on a descendant holding stderr open. + yield* Effect.forkIn( + child.stderr.pipe( + Stream.decodeText(), + Stream.tap((chunk) => + Effect.sync(() => { + stderrTail = (stderrTail + chunk).slice(-16_384); + }), + ), + Stream.splitLines, + Stream.tap((line) => + line === "debug1: Entering interactive session." + ? Deferred.succeed(forwardReady, undefined) + : Effect.void, + ), + Stream.runDrain, + Effect.tapError((cause) => + Deferred.fail( + forwardReady, + new SshReadinessError({ + message: "Failed to read SSH forward diagnostics.", + cause, + }), + ), + ), + ), + scope, + ); + } + const exitObservation = + input.readiness === "forward" + ? child.exitCode.pipe(Effect.map((exitCode) => [stderrTail, Number(exitCode)] as const)) + : Effect.all([collectProcessOutput(child.stderr), child.exitCode.pipe(Effect.map(Number))], { + concurrency: "unbounded", + }); + const exitFailure = exitObservation.pipe( Effect.mapError( (cause) => new SshCommandError({ @@ -1253,13 +1328,39 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: }).pipe(Effect.andThen(Effect.fail(error))); }), ); - yield* Effect.raceFirst( - waitForHttpReady({ - baseUrl: input.httpBaseUrl, - timeoutMs: SSH_READY_TIMEOUT_MS, - }), - exitFailure, - ).pipe( + // Keep draining DEBUG1 output for the lifetime of the child, not just startup. + const monitor = input.readiness === "forward" ? yield* Effect.forkIn(exitFailure, scope) : null; + const readiness: Effect.Effect< + void, + SshReadinessError, + NetService.NetService | HttpClient.HttpClient + > = + input.readiness === "forward" + ? Deferred.await(forwardReady).pipe( + Effect.timeoutOrElse({ + duration: SSH_READY_TIMEOUT_MS, + orElse: () => + Effect.fail( + new SshReadinessError({ + message: `SSH forward on 127.0.0.1:${input.localPort} did not become ready.${stderrTail.trim() ? `\n${stderrTail.trim().split(/\r?\n/u).slice(-8).join("\n")}` : ""}`, + }), + ), + }), + Effect.andThen( + Effect.gen(function* () { + const net = yield* NetService.NetService; + const listening = yield* net.hasListenerOnHost(input.localPort, "127.0.0.1"); + const running = yield* child.isRunning.pipe(Effect.orElseSucceed(() => false)); + if (!listening || !running) { + return yield* new SshReadinessError({ + message: "SSH forward stopped before becoming ready.", + }); + } + }), + ), + ) + : waitForHttpReady({ baseUrl: input.httpBaseUrl, timeoutMs: SSH_READY_TIMEOUT_MS }); + yield* Effect.raceFirst(readiness, monitor === null ? exitFailure : Fiber.join(monitor)).pipe( Effect.tap(() => Effect.logInfo("ssh.tunnel.ready", { ...sshTargetLogFields(input.resolvedTarget), @@ -1278,7 +1379,9 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: net.canListenOnHost(input.localPort, "127.0.0.1"), ); const remoteLogTailExit = yield* Effect.exit( - readRemoteServerLogTail(input.resolvedTarget, input.authOptions), + input.readiness === "forward" + ? Effect.succeed("") + : readRemoteServerLogTail(input.resolvedTarget, input.authOptions), ); const processRunning = Exit.isSuccess(processRunningExit) ? processRunningExit.value : null; const localPortAvailable = Exit.isSuccess(localPortAvailableExit) @@ -1311,7 +1414,7 @@ const startSshTunnel = Effect.fn("ssh/tunnel.startSshTunnel")(function* (input: }), ), Effect.onExit((exit) => - Exit.isSuccess(exit) + Exit.isSuccess(exit) || input.readiness === "forward" ? Effect.void : child .kill({ @@ -1331,6 +1434,41 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma const tunnels = new Map(); const targetLocks = yield* KeyedLock.make(); const authSecrets = new Map(); + const portForwards = new Map(); + const leases = new Map(); + const pendingForwards = new Map< + string, + { + readonly connectionKey: string; + readonly cancelled: Deferred.Deferred; + readonly finished: Deferred.Deferred; + } + >(); + const disconnecting = new Map(); + const targetGenerations = new Map(); + let forwardsClosed = false; + let nextLeaseId = 0; + + const closePortForward = (entry: SshPortForwardEntry) => + Effect.gen(function* () { + if (portForwards.get(entry.key) === entry) portForwards.delete(entry.key); + for (const leaseId of entry.leases) leases.delete(leaseId); + entry.leases.clear(); + yield* Scope.close(entry.scope, Exit.void).pipe(Effect.ignore); + }).pipe(Effect.uninterruptible); + + const cancelPendingForwards = (connectionKey?: string) => + Effect.forEach( + [...pendingForwards.values()].filter( + (pending) => connectionKey === undefined || pending.connectionKey === connectionKey, + ), + (pending) => + Deferred.fail( + pending.cancelled, + new SshReadinessError({ message: "SSH forward creation was cancelled." }), + ).pipe(Effect.andThen(Deferred.await(pending.finished))), + { concurrency: "unbounded", discard: true }, + ); // Keep one lock per target so reconnect cannot reuse a server while stop is pending. const withTargetLock = Effect.fn("ssh/tunnel.withTargetLock")(function* ( @@ -1360,7 +1498,19 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma yield* Scope.addFinalizer( managerScope, - Effect.sync(() => [...tunnels.values()]).pipe( + Effect.sync(() => { + forwardsClosed = true; + }).pipe( + Effect.andThen(Effect.suspend(() => cancelPendingForwards())), + Effect.andThen( + Effect.suspend(() => + Effect.forEach([...portForwards.values()], closePortForward, { + concurrency: "unbounded", + discard: true, + }), + ), + ), + Effect.andThen(Effect.sync(() => [...tunnels.values()])), Effect.flatMap((entries) => Effect.forEach(entries, closeTunnelEntry, { concurrency: "unbounded" }), ), @@ -1529,6 +1679,7 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma wsBaseUrl, authOptions, remoteServerKind: remoteLaunch.remoteServerKind, + remoteHost: "127.0.0.1", scope: entryScope, }), }).pipe( @@ -1713,6 +1864,196 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma ); }); + const acquirePortForward = Effect.fn("ssh/tunnel.acquirePortForward")(function* ( + target: DesktopSshEnvironmentTarget, + remotePort: number, + ) { + if (!Number.isInteger(remotePort) || remotePort < 1 || remotePort > 65535) { + return yield* new SshInvalidTargetError({ + message: "SSH forward port must be an integer between 1 and 65535.", + }); + } + const baseResolved = yield* resolveSshTarget(target.alias || target.hostname); + const resolvedTarget = { + ...baseResolved, + ...(target.username !== null ? { username: target.username } : {}), + ...(target.port !== null ? { port: target.port } : {}), + }; + const connectionKey = targetConnectionKey(resolvedTarget); + if (forwardsClosed || disconnecting.has(connectionKey)) { + return yield* new SshReadinessError({ message: "SSH environment is closing." }); + } + const key = `${connectionKey}:${remotePort}`; + const generation = targetGenerations.get(connectionKey) ?? 0; + let acquiredLeaseId: string | null = null; + return yield* withTargetLock( + connectionKey, + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + if ( + forwardsClosed || + disconnecting.has(connectionKey) || + generation !== (targetGenerations.get(connectionKey) ?? 0) + ) { + return yield* new SshReadinessError({ message: "SSH environment is closing." }); + } + let entry = portForwards.get(key); + if (entry !== undefined) { + const running = yield* restore(entry.process.isRunning).pipe( + Effect.orElseSucceed(() => false), + ); + if (!running) { + yield* closePortForward(entry); + entry = undefined; + } + } + if (entry === undefined) { + if ( + forwardsClosed || + disconnecting.has(connectionKey) || + generation !== (targetGenerations.get(connectionKey) ?? 0) + ) { + return yield* new SshReadinessError({ message: "SSH environment is closing." }); + } + // Register pending ownership atomically with the closed-state check. + const scope = Scope.makeUnsafe("sequential"); + const cancelled = Deferred.makeUnsafe(); + const finished = Deferred.makeUnsafe(); + const pending = { connectionKey, cancelled, finished }; + pendingForwards.set(key, pending); + const exit = yield* Effect.exit( + restore( + Effect.raceFirst( + Effect.gen(function* () { + const net = yield* NetService.NetService; + // Both loopback families: the tab loads `localhost`, which may + // resolve to a local ::1 server on the same port before ours. + const preferRemotePort = + remotePort >= 1024 && (yield* net.isPortAvailableOnLoopback(remotePort)); + const localPort = preferRemotePort + ? remotePort + : yield* reserveLocalTunnelPort(); + const startForward = (port: number) => + Effect.uninterruptibleMask((restoreStart) => + Effect.gen(function* () { + const attemptScope = yield* Scope.fork(scope, "sequential"); + return yield* restoreStart( + runWithSshAuth({ + key: connectionKey, + target: resolvedTarget, + operation: (authOptions) => + startSshTunnel({ + key, + resolvedTarget, + remotePort, + localPort: port, + httpBaseUrl: `http://127.0.0.1:${port}/`, + wsBaseUrl: `ws://127.0.0.1:${port}/`, + remoteServerKind: null, + readiness: "forward", + remoteHost: "localhost", + authOptions, + scope: attemptScope, + }), + }), + ).pipe( + Effect.onExit((attemptExit) => + Exit.isFailure(attemptExit) + ? Scope.close(attemptScope, Exit.void).pipe(Effect.ignore) + : Effect.void, + ), + ); + }), + ); + return yield* startForward(localPort).pipe( + Effect.catchTags({ + SshCommandError: (error) => + Effect.gen(function* () { + // The port probe cannot reserve the port for ssh. Re-probe + // after an exited preferred attempt, independent of stderr + // delivery order or localized bind diagnostics. + if ( + !preferRemotePort || + error.exitCode === null || + (yield* net.isPortAvailableOnLoopback(localPort)) + ) { + return yield* error; + } + return yield* reserveLocalTunnelPort().pipe( + Effect.flatMap(startForward), + ); + }), + }), + ); + }), + Deferred.await(cancelled), + ), + ), + ); + if (Exit.isFailure(exit)) { + yield* Scope.close(scope, Exit.void).pipe(Effect.ignore); + pendingForwards.delete(key); + yield* Deferred.succeed(finished, undefined); + return yield* Effect.failCause(exit.cause); + } + if ( + forwardsClosed || + disconnecting.has(connectionKey) || + generation !== (targetGenerations.get(connectionKey) ?? 0) + ) { + yield* Scope.close(scope, Exit.void).pipe(Effect.ignore); + pendingForwards.delete(key); + yield* Deferred.succeed(finished, undefined); + return yield* new SshReadinessError({ + message: "SSH environment closed during forward creation.", + }); + } + entry = { ...exit.value, scope, connectionKey, leases: new Set() }; + portForwards.set(key, entry); + pendingForwards.delete(key); + yield* Deferred.succeed(finished, undefined); + } + // Scope close can run while an existing child's running check is suspended. + if ( + forwardsClosed || + disconnecting.has(connectionKey) || + generation !== (targetGenerations.get(connectionKey) ?? 0) || + portForwards.get(key) !== entry + ) { + return yield* new SshReadinessError({ message: "SSH environment is closing." }); + } + const leaseId = `ssh-port-forward-${++nextLeaseId}`; + entry.leases.add(leaseId); + leases.set(leaseId, entry); + acquiredLeaseId = leaseId; + return { leaseId, localPort: entry.localPort }; + }), + ), + ).pipe( + Effect.onExit((exit) => + Exit.isFailure(exit) && acquiredLeaseId !== null + ? releasePortForward(acquiredLeaseId) + : Effect.void, + ), + ); + }); + + const releasePortForward = Effect.fn("ssh/tunnel.releasePortForward")(function* ( + leaseId: string, + ) { + const entry = leases.get(leaseId); + if (entry === undefined) return; + yield* withTargetLock( + entry.connectionKey, + Effect.gen(function* () { + if (leases.get(leaseId) !== entry) return; + leases.delete(leaseId); + entry.leases.delete(leaseId); + if (entry.leases.size === 0) yield* closePortForward(entry); + }).pipe(Effect.uninterruptible), + ); + }); + const disconnectEnvironment = Effect.fn("ssh/tunnel.disconnectEnvironment")(function* ( target: DesktopSshEnvironmentTarget, ): Effect.fn.Return { @@ -1724,36 +2065,65 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma ...(target.port !== null ? { port: target.port } : {}), }; const key = targetConnectionKey(resolvedTarget); - yield* withTargetLock( - key, - Effect.gen(function* () { - const entry = tunnels.get(key) ?? null; - yield* Effect.logDebug("ssh.environment.disconnect.targetResolved", { - ...sshTargetLogFields(resolvedTarget), - key, - hasTunnel: entry !== null, - }); - if (entry !== null) { - // Explicit disconnect owns the remote stop so its failure reaches the caller. - yield* Effect.gen(function* () { - tunnels.delete(key); - yield* closeTunnelEntry(entry); - }).pipe(Effect.uninterruptible); - } - yield* runWithSshAuth({ - key, - target: resolvedTarget, - operation: (authOptions) => stopRemoteServer(resolvedTarget, authOptions), - }); - yield* Effect.logInfo("ssh.environment.disconnect.succeeded", { - ...sshTargetLogFields(resolvedTarget), - key, - }); + yield* Effect.uninterruptibleMask((restoreDisconnect) => + Effect.suspend(() => { + targetGenerations.set(key, (targetGenerations.get(key) ?? 0) + 1); + disconnecting.set(key, (disconnecting.get(key) ?? 0) + 1); + return Effect.gen(function* () { + yield* cancelPendingForwards(key); + yield* withTargetLock( + key, + Effect.gen(function* () { + yield* Effect.forEach( + [...portForwards.values()].filter((entry) => entry.connectionKey === key), + closePortForward, + { discard: true }, + ); + const entry = tunnels.get(key) ?? null; + yield* Effect.logDebug("ssh.environment.disconnect.targetResolved", { + ...sshTargetLogFields(resolvedTarget), + key, + hasTunnel: entry !== null, + }); + if (entry !== null) { + // Explicit disconnect owns the remote stop so its failure reaches the caller. + yield* Effect.gen(function* () { + tunnels.delete(key); + yield* closeTunnelEntry(entry); + }).pipe(Effect.uninterruptible); + } + yield* restoreDisconnect( + runWithSshAuth({ + key, + target: resolvedTarget, + operation: (authOptions) => stopRemoteServer(resolvedTarget, authOptions), + }), + ); + yield* Effect.logInfo("ssh.environment.disconnect.succeeded", { + ...sshTargetLogFields(resolvedTarget), + key, + }); + }), + ); + }).pipe( + Effect.ensuring( + Effect.sync(() => { + const remaining = (disconnecting.get(key) ?? 1) - 1; + if (remaining === 0) disconnecting.delete(key); + else disconnecting.set(key, remaining); + }), + ), + ); }), ); }); - return SshEnvironmentManager.of({ ensureEnvironment, disconnectEnvironment }); + return SshEnvironmentManager.of({ + ensureEnvironment, + disconnectEnvironment, + acquirePortForward, + releasePortForward, + }); }); /**