Add a direct SSH mode without ControlMaster

apps/review-desktop/code-oss/src/vs/review/electron-main/remote/reviewRemoteHost.ts

@@ -13,7 +13,8 @@ import { installedAttachScript, parseRemoteAttach, reviewRemoteAttachScript, typ
 import type { ReviewRemoteInstallConsent } from "./reviewRemoteInstallConsent.js";
 import type { ReviewRemoteInstallInput, ReviewRemoteInstallResult } from "./reviewRemoteInstaller.js";
 import { reviewRemoteVersionDir } from "./reviewRemoteInstallScript.js";
-import { judgeRemote, missingInstallTools, probeRemote, type ReviewRemoteProbe, type ReviewRemoteTarget } from "./reviewRemoteProbe.js";
+import { judgeRemote, missingInstallTools, probeRemote, REVIEW_REMOTE_PROBE_TIMEOUT, type ReviewRemoteProbe, type ReviewRemoteTarget } from "./reviewRemoteProbe.js";
+import { socksAnswers, socksForward, type ReviewSocksForward } from "./reviewSocksForward.js";
 import {
 	sshCancelForwardArgs,
 	sshCheckArgs,
@@ -301,6 +302,8 @@ export class ReviewRemoteHost {
 	private master: SshChildProcess | undefined;
 	private readonly closing = new Map<SshChildProcess, Promise<void>>();
 	private forwarded: { local: number; remote: number } | undefined;
+	private socksPort: number | undefined;
+	private readonly socks = new Map<number, ReviewSocksForward>();
 	private language: { local: number; remote: number; connectionToken: string; commit: string } | undefined;
 	private pendingAttaches = 0;
 	private cancelPending: (() => void) | undefined;
@@ -331,6 +334,10 @@ export class ReviewRemoteHost {
 		this.reported = { alias: this.alias };
 	}
 
+	private get direct(): boolean {
+		return !this.options.session.controlPath;
+	}
+
 	get state(): ReviewGatewayHost {
 		return this.reported;
 	}
@@ -390,9 +397,9 @@ export class ReviewRemoteHost {
 		this.reattaching = true;
 		this.options.log(`${this.alias}: ${reason}; attaching again.`);
 		try {
-			const check = await this.run(sshCheckArgs(this.options.session, env), this.timeouts.operation);
+			const check = await this.masterAnswers(env);
 			if (stale()) return;
-			if (check.code !== 0) throw unreachable(`The SSH connection to ${this.alias} did not answer. ${firstLines(check.stderr)}`.trim());
+			if (!check.ok) throw unreachable(`The SSH connection to ${this.alias} did not answer. ${firstLines(check.stderr)}`.trim());
 			await this.establish(env, stale, { script: this.attachScript, cli: this.cli }, reuse);
 		} catch (error) {
 			if (stale()) return;
@@ -447,6 +454,7 @@ export class ReviewRemoteHost {
 		this.installing?.abort();
 		const master = this.master;
 		this.master = undefined;
+		this.closeForwards();
 		await this.close(master);
 	}
 
@@ -509,9 +517,10 @@ export class ReviewRemoteHost {
 			this.env = env;
 			await Promise.all(this.closing.values());
 			// None of this host's masters runs now, but one killed by a signal left its socket, and ssh does not multiplex on a path that exists.
-			await rm(this.options.session.controlPath, { force: true });
+			if (this.options.session.controlPath) await rm(this.options.session.controlPath, { force: true });
+			const socksPort = this.direct ? await this.options.freePort() : undefined;
 			if (stale()) return;
-			const master = this.startMaster(env);
+			const master = this.startMaster(env, socksPort);
 			await this.waitForMaster(master, env, stale);
 			const prepared = await this.prepareAttach(env, stale);
 			if (prepared === undefined) return;
@@ -537,7 +546,7 @@ export class ReviewRemoteHost {
 		const language = await this.forwardLanguage(env, attach, stale);
 		if (stale()) return;
 		for (const forward of [kept ? undefined : old, oldLanguage]) {
-			if (forward) await this.run(sshCancelForwardArgs(this.options.session, forward.local, forward.remote, env), this.timeouts.operation);
+			if (forward) await this.cancelForward(env, forward);
 			if (stale()) return;
 		}
 		this.attachScript = prepared.script;
@@ -554,10 +563,10 @@ export class ReviewRemoteHost {
 		return this.failures++;
 	}
 
-	private startMaster(env: NodeJS.ProcessEnv): SshChildProcess {
+	private startMaster(env: NodeJS.ProcessEnv, socksPort: number | undefined): SshChildProcess {
 		let master: SshChildProcess;
 		try {
-			master = this.options.spawn(sshMasterArgs(this.options.session, env), {
+			master = this.options.spawn(sshMasterArgs(this.options.session, env, socksPort), {
 				env,
 				detached: true,
 				stdio: ["ignore", "ignore", "pipe"],
@@ -566,6 +575,7 @@ export class ReviewRemoteHost {
 			throw (error as NodeJS.ErrnoException).code === "ENOENT" ? unreachable(OPENSSH_NEEDED) : error;
 		}
 		this.master = master;
+		this.socksPort = socksPort;
 		this.masterStderr = "";
 		master.stderr?.setEncoding("utf8");
 		master.stderr?.on("data", (chunk: string) => (this.masterStderr = (this.masterStderr + chunk).slice(-OUTPUT_LIMIT)));
@@ -601,9 +611,9 @@ export class ReviewRemoteHost {
 		let since = Date.now();
 		for (;;) {
 			if (stale() || gone(master)) throw unreachable("The SSH connection ended.");
-			const check = await this.run(sshCheckArgs(this.options.session, env), this.timeouts.operation);
+			const check = await this.masterAnswers(env);
 			if (stale()) throw unreachable("The SSH connection ended.");
-			if (check.code === 0) return;
+			if (check.ok) return;
 			if (this.prompts > 0) since = Date.now();
 			if (Date.now() - since > this.timeouts.connect) {
 				throw unreachable(`${this.alias} did not connect within ${this.timeouts.connect / 1000} seconds. ${firstLines(this.masterStderr)}`.trim());
@@ -612,14 +622,22 @@ export class ReviewRemoteHost {
 		}
 	}
 
+	private async masterAnswers(env: NodeJS.ProcessEnv): Promise<{ ok: boolean; stderr: string }> {
+		if (this.direct) return { ok: this.socksPort !== undefined && (await socksAnswers(this.socksPort)), stderr: firstLines(this.masterStderr) };
+		const check = await this.run(sshCheckArgs(this.options.session, env), this.timeouts.operation);
+		return { ok: check.code === 0, stderr: check.stderr };
+	}
+
 	private async prepareAttach(env: NodeJS.ProcessEnv, stale: () => boolean): Promise<PreparedAttach | undefined> {
 		const onPath = { script: reviewRemoteAttachScript };
 		const flow = this.options.install;
 		const mode = flow?.mode() ?? "never";
 		if (!flow || mode === "never") return onPath;
 		const origin = flow.origin();
 		const script = await flow.script();
-		const probed = await probeRemote({ session: this.options.session, spawn: this.options.spawn, env, script, releases: origin && `${origin}/` });
+		// A direct probe logs in afresh, so it gets the connect allowance too.
+		const timeout = this.direct ? REVIEW_REMOTE_PROBE_TIMEOUT + this.timeouts.connect : undefined;
+		const probed = await probeRemote({ session: this.options.session, spawn: this.options.spawn, env, script, releases: origin && `${origin}/`, timeout });
 		if (stale()) return;
 		if ("error" in probed) throw unreachable(probed.error);
 		const { probe } = probed;
@@ -799,10 +817,8 @@ export class ReviewRemoteHost {
 
 	private async forward(env: NodeJS.ProcessEnv, attach: ReviewRemoteAttach, stale: () => boolean): Promise<string> {
 		this.forwarded = undefined;
-		const port = await this.options.freePort();
-		if (stale()) throw unreachable("The SSH connection ended.");
-		const forward = await this.run(sshForwardArgs(this.options.session, port, attach.port, env), this.timeouts.operation);
-		if (forward.code !== 0) throw unreachable(`Could not forward a local port to ${this.alias}: ${firstLines(forward.stderr) || `ssh exited with code ${forward.code}`}`);
+		const port = await this.openForward(env, attach.port, stale);
+		if (typeof port !== "number") throw unreachable(`Could not forward a local port to ${this.alias}: ${port.error}`);
 		try {
 			await probeHealth(port, this.timeouts.operation);
 		} catch (error) {
@@ -820,21 +836,51 @@ export class ReviewRemoteHost {
 		if (!server) return unavailable(attach.languageServerDetail ?? "This host has no VS Code server.");
 		const mismatch = languageCommitMismatch(this.alias, server.commit, this.options.desktopCommit);
 		if (mismatch) return unavailable(mismatch);
-		const port = await this.options.freePort();
-		if (stale()) throw unreachable("The SSH connection ended.");
-		const forward = await this.run(sshForwardArgs(this.options.session, port, server.port, env), this.timeouts.operation);
-		if (forward.code !== 0) return unavailable(`Could not forward a local port to the VS Code server on ${this.alias}: ${firstLines(forward.stderr) || `ssh exited with code ${forward.code}`}`);
+		const port = await this.openForward(env, server.port, stale);
+		if (typeof port !== "number") return unavailable(`Could not forward a local port to the VS Code server on ${this.alias}: ${port.error}`);
 		try {
 			const commit = await probeVersion(port, this.timeouts.operation);
 			if (commit !== server.commit) throw new Error(`it reports ${commit}, not ${server.commit}`);
 		} catch (error) {
-			await this.run(sshCancelForwardArgs(this.options.session, port, server.port, env), this.timeouts.operation);
+			await this.cancelForward(env, { local: port, remote: server.port });
 			return unavailable(`The VS Code server on ${this.alias} did not answer through the forward: ${(error as Error).message}.`);
 		}
 		this.language = { local: port, remote: server.port, connectionToken: server.connectionToken, commit: server.commit };
 		return { languageFeatures: true };
 	}
 
+	/** A loopback port that reaches `remote` on the host. */
+	private async openForward(env: NodeJS.ProcessEnv, remote: number, stale: () => boolean): Promise<number | { error: string }> {
+		if (this.direct) {
+			const socksPort = this.socksPort;
+			if (socksPort === undefined) throw unreachable("The SSH connection ended.");
+			const forward = await socksForward(socksPort, remote);
+			if (stale()) {
+				await forward.close();
+				throw unreachable("The SSH connection ended.");
+			}
+			this.socks.set(forward.port, forward);
+			return forward.port;
+		}
+		const port = await this.options.freePort();
+		if (stale()) throw unreachable("The SSH connection ended.");
+		const forward = await this.run(sshForwardArgs(this.options.session, port, remote, env), this.timeouts.operation);
+		return forward.code === 0 ? port : { error: firstLines(forward.stderr) || `ssh exited with code ${forward.code}` };
+	}
+
+	private async cancelForward(env: NodeJS.ProcessEnv, { local, remote }: { local: number; remote: number }): Promise<void> {
+		const socks = this.socks.get(local);
+		this.socks.delete(local);
+		if (socks) await socks.close();
+		else if (!this.direct) await this.run(sshCancelForwardArgs(this.options.session, local, remote, env), this.timeouts.operation);
+	}
+
+	private closeForwards(): void {
+		for (const forward of this.socks.values()) void forward.close();
+		this.socks.clear();
+		this.socksPort = undefined;
+	}
+
 	private facts(): Pick<ReviewGatewayHost, "declined" | "installFailure"> {
 		return { ...(this.declined && { declined: true as const }), ...(this.installFailure && { installFailure: this.installFailure }) };
 	}
@@ -863,6 +909,7 @@ export class ReviewRemoteHost {
 		this.master = undefined;
 		this.forwarded = undefined;
 		this.language = undefined;
+		this.closeForwards();
 		this.installing?.abort();
 		void this.close(master);
 	}
@@ -873,7 +920,7 @@ export class ReviewRemoteHost {
 		if (pending) return pending;
 		const closed = (async () => {
 			// A master still connecting has no socket to take -O exit.
-			const asked = this.env && (await this.run(sshCloseArgs(this.options.session, this.env), this.timeouts.close)).code === 0;
+			const asked = !this.direct && this.env && (await this.run(sshCloseArgs(this.options.session, this.env), this.timeouts.close)).code === 0;
 			if (!asked || !(await exitedWithin(master, this.timeouts.close))) master.kill("SIGTERM");
 			await exitedWithin(master, this.timeouts.close);
 		})().finally(() => this.closing.delete(master));

apps/review-desktop/code-oss/src/vs/review/electron-main/remote/reviewRemoteHosts.ts

@@ -23,15 +23,18 @@ import { uninstallRemote } from "./reviewRemoteUninstall.js";
 import {
 	prepareSshControlDirectory,
 	reviewSshInstancePrefix,
+	reviewSshDirectSession,
 	reviewSshSession,
 	sshCloseArgs,
+	type ReviewSshSession,
 } from "./reviewSshCommand.js";
 import { validateSshAlias } from "../../common/reviewSshAlias.js";
 
 
 export interface ReviewRemoteHostsOptions {
 	readonly spawn: SpawnSsh;
-	readonly controlDirectory: string;
+	/** Absent: the connections are direct, and nothing prompts. */
+	readonly controlDirectory?: string;
 	readonly instance: string;
 	environment(): Promise<NodeJS.ProcessEnv>;
 	createAskpass(input: {
@@ -198,7 +201,7 @@ export class ReviewRemoteHosts {
 			if (!this.options.install) throw new Error("This Desktop cannot install or remove Whiteboard on hosts.");
 			const env = { ...(await this.options.environment()), ...(await this.askpass()).env(alias) };
 			const script = await this.options.install.script();
-			await uninstallRemote({ session: reviewSshSession(alias, this.options.controlDirectory, this.options.instance), spawn: this.options.spawn, env, script });
+			await uninstallRemote({ session: this.session(alias), spawn: this.options.spawn, env, script });
 		} catch (error) {
 			this.options.log(`${alias}: Whiteboard was not removed; the host waits for the setting.`);
 			throw error;
@@ -255,7 +258,7 @@ export class ReviewRemoteHosts {
 
 	private createHost(alias: string): ReviewRemoteHost {
 		return new ReviewRemoteHost({
-			session: reviewSshSession(alias, this.options.controlDirectory, this.options.instance),
+			session: this.session(alias),
 			spawn: this.options.spawn,
 			environment: async () => ({ ...(await this.options.environment()), ...(await this.askpass()).env(alias) }),
 			desktopVersion: () => this.options.desktopVersion(),
@@ -276,25 +279,32 @@ export class ReviewRemoteHosts {
 		if (serverId && ![...this.hosts.values()].some((other) => other.serverId === serverId)) this.options.removed?.(serverId);
 	}
 
+	private session(alias: string): ReviewSshSession {
+		const directory = this.options.controlDirectory;
+		return directory === undefined ? reviewSshDirectSession(alias) : reviewSshSession(alias, directory, this.options.instance);
+	}
+
 	private askpass(): Promise<ReviewSshAskpass> {
 		return (this.prepared ??= (async () => {
-			await prepareSshControlDirectory(this.options.controlDirectory);
-			await this.sweepOrphans();
+			const directory = this.options.controlDirectory;
+			if (directory === undefined) return { env: () => ({}), dispose() {} };
+			await prepareSshControlDirectory(directory);
+			await this.sweepOrphans(directory);
 			return this.options.createAskpass({
-				directory: this.options.controlDirectory,
+				directory,
 				prompt: (request) => this.prompt(request),
 				log: this.options.log,
 			});
 		})());
 	}
 
-	private async sweepOrphans(): Promise<void> {
+	private async sweepOrphans(directory: string): Promise<void> {
 		const prefix = reviewSshInstancePrefix(this.options.instance);
 		try {
 			const env = await this.options.environment();
-			for (const name of await readdir(this.options.controlDirectory)) {
+			for (const name of await readdir(directory)) {
 				if (!name.startsWith(prefix)) continue;
-				const controlPath = join(this.options.controlDirectory, name);
+				const controlPath = join(directory, name);
 				try {
 					// -O exit works on any socket; the alias after -- is only a placeholder.
 					const closed = await runSsh(this.options.spawn, env, sshCloseArgs({ alias: "orphan", controlPath }, env), REVIEW_REMOTE_TIMEOUTS.close);

apps/review-desktop/code-oss/src/vs/review/electron-main/remote/reviewSocksForward.test.ts

@@ -0,0 +1,88 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) dev.fast. All rights reserved.
+ *  Licensed under the MIT License. See LICENSE in the repository root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+import assert from "node:assert/strict";
+import { once } from "node:events";
+import { connect, createServer, type AddressInfo, type Server, type Socket } from "node:net";
+import test from "node:test";
+
+import { socksAnswers, socksForward } from "./reviewSocksForward.js";
+
+async function listen(server: Server): Promise<number> {
+	server.listen(0, "127.0.0.1");
+	await once(server, "listening");
+	return (server.address() as AddressInfo).port;
+}
+
+/** Like OpenSSH's -D: no auth, CONNECT only, and the reply before the far end answers. */
+function fakeSocks(): Server {
+	return createServer({ allowHalfOpen: true }, (client) => {
+		client.once("data", () => {
+			client.write(Buffer.from([5, 0]));
+			client.once("data", (request: Buffer) => {
+				const port = request.readUInt16BE(8);
+				client.write(Buffer.from([5, 0, 0, 1, 0, 0, 0, 0, 0, 0]));
+				const far = connect(port, "127.0.0.1");
+				far.on("error", () => client.destroy());
+				client.on("error", () => far.destroy());
+				client.pipe(far).pipe(client);
+			});
+		});
+	});
+}
+
+const echo = () => createServer((socket) => socket.pipe(socket));
+
+async function roundTrip(port: number, text: string): Promise<string> {
+	const socket = connect(port, "127.0.0.1");
+	socket.end(text);
+	let answer = "";
+	for await (const chunk of socket) answer += chunk;
+	return answer;
+}
+
+test("a forward reaches the far port through the SOCKS port, and close ends its connections", async (t) => {
+	const socks = fakeSocks();
+	const far = echo();
+	t.after(() => {
+		socks.close();
+		far.close();
+	});
+	const forward = await socksForward(await listen(socks), await listen(far));
+	assert.equal(await roundTrip(forward.port, "hello"), "hello");
+
+	const open: Socket = connect(forward.port, "127.0.0.1");
+	open.write("x");
+	await once(open, "data");
+	await forward.close();
+	await once(open, "close");
+	await assert.rejects(roundTrip(forward.port, "again"));
+});
+
+test("a far port that refuses closes the forwarded connection", async (t) => {
+	const socks = fakeSocks();
+	const closed = echo();
+	const closedPort = await listen(closed);
+	closed.close();
+	t.after(() => socks.close());
+	const forward = await socksForward(await listen(socks), closedPort);
+	t.after(() => forward.close());
+	assert.equal(await roundTrip(forward.port, "hello").catch(() => ""), "");
+});
+
+test("socksAnswers is true only for a SOCKS port", async (t) => {
+	const socks = fakeSocks();
+	const silent = createServer((socket) => socket.end());
+	t.after(() => {
+		socks.close();
+		silent.close();
+	});
+	assert.equal(await socksAnswers(await listen(socks)), true);
+	assert.equal(await socksAnswers(await listen(silent)), false);
+	const gone = echo();
+	const gonePort = await listen(gone);
+	gone.close();
+	assert.equal(await socksAnswers(gonePort), false);
+});