diff --git a/package.json b/package.json index 3d77244..da5fb50 100644 --- a/package.json +++ b/package.json @@ -188,9 +188,10 @@ "scripts": { "compile": "tsc --noEmit && eslint src", "build": "npm run build:dev", - "build:dev": "node esbuild.js && cp src/remote/installServer.sh dist/ && cp src/engine/Dockerfile dist/", + "copy_templates": "cp src/remote/installServer.sh dist/ && cp src/engine/Dockerfile dist/ && cp src/remote/exec-pipe/remote.ts dist/", + "build:dev": "node esbuild.js && npm run copy_templates", + "build:prod": "node esbuild.js --production && npm run copy_templates", "build:clean": "rm -rf dist out", - "build:prod": "node esbuild.js --production && cp src/remote/installServer.sh dist/ && cp src/engine/Dockerfile dist/", "test": "npm run test:noengine", "test:noengine": "SKIP_ENGINE_TESTS=1 vitest run", "test:engine": "vitest run", diff --git a/src/common/ctx/execCtx.ts b/src/common/ctx/execCtx.ts index 05a1274..52e5684 100644 --- a/src/common/ctx/execCtx.ts +++ b/src/common/ctx/execCtx.ts @@ -16,10 +16,20 @@ export interface FsCtx { write(filePath: vscode.Uri, content: string): Promise, }; +export interface SpawnedProcess { + readonly stdin: { write(data: Buffer | Uint8Array): void, end(): void }, + onStdout(cb: (data: Buffer) => void): void, + onStderr(cb: (data: Buffer) => void): void, + onError(cb: (err: Error) => void): void, + onExit(cb: (code: number | null) => void): void, + onClose(cb: () => void): void, +} + export interface ExecCtx { kind: ExecCtxKind, fs: FsCtx, run(cmdArgs: string[], opts: RunOpts): Promise, + spawn(cmdArgs: string[], opts: RunOpts): Promise, getHostUserInfo(): Promise, env(): Promise, getSettings(): Promise, diff --git a/src/common/ctx/localCtx.ts b/src/common/ctx/localCtx.ts index 19bf671..eb2a138 100644 --- a/src/common/ctx/localCtx.ts +++ b/src/common/ctx/localCtx.ts @@ -1,8 +1,9 @@ import * as vscode from "vscode"; +import * as chproc from "node:child_process"; -import { ExecCtx, ExecCtxKind, FsCtx } from "./execCtx"; +import { ExecCtx, ExecCtxKind, FsCtx, SpawnedProcess } from "./execCtx"; import { HostUserInfo } from "../utils"; -import { formatCmdErr, spawn } from "../spawn"; +import { collectOutput, formatCmdErr } from "../spawn"; import { getLogSink } from "../../extension/log"; import { SpawnError } from "../../extension/error"; import { CmdResult, RunOpts } from "../opts"; @@ -39,14 +40,37 @@ export class LocalExecCtx implements ExecCtx { } async run(cmdArgs: string[], opts: RunOpts): Promise { + const proc = await this.spawn(cmdArgs, opts); + return collectOutput(proc, opts); + } + + async spawn(cmdArgs: string[], opts: RunOpts): Promise { if (this.execServer) { - return await spawnRemote(this.execServer, cmdArgs, opts); + return spawnRemote(this.execServer, cmdArgs, opts); } - return await spawn(cmdArgs[0], cmdArgs.slice(1), { ...opts, log: getLogSink() }); + + const [cmd, ...args] = cmdArgs; + const proc = chproc.spawn(cmd, args, { + cwd: opts.cwd, + env: { ...process.env, BUILDKIT_PROGRESS: "plain", ...opts.env }, + stdio: ["pipe", "pipe", "pipe"], + }); + + return { + stdin: { + write(data: Buffer | Uint8Array) { proc.stdin.write(data); }, + end() { proc.stdin.end(); }, + }, + onStdout(cb) { proc.stdout.on("data", cb); }, + onStderr(cb) { proc.stderr.on("data", cb); }, + onError(cb) { proc.on("error", cb); }, + onExit(cb) { proc.on("exit", cb); }, + onClose(cb) { proc.on("close", cb); }, + }; } async getHostUserInfo(): Promise { - const userName = await spawn("id", ["-nu"], { log: getLogSink() }); + const userName = await this.run(["id", "-nu"], {}); /* eslint-disable @typescript-eslint/no-non-null-assertion */ if (userName.exit === 0) { return { diff --git a/src/common/ctx/remoteCtx.ts b/src/common/ctx/remoteCtx.ts index 35bdc5b..6ecf4e3 100644 --- a/src/common/ctx/remoteCtx.ts +++ b/src/common/ctx/remoteCtx.ts @@ -1,8 +1,8 @@ import * as vscode from "vscode"; -import { ExecCtx, ExecCtxKind, FsCtx } from "./execCtx"; +import { ExecCtx, ExecCtxKind, FsCtx, SpawnedProcess } from "./execCtx"; import { HostUserInfo } from "../utils"; -import { formatCmdErr } from "../spawn"; +import { collectOutput, formatCmdErr } from "../spawn"; import { getLogSink } from "../../extension/log"; import { SpawnError } from "../../extension/error"; import { CmdResult, RunOpts } from "../opts"; @@ -54,6 +54,11 @@ export class RemoteExecCtx implements ExecCtx { } public async run(cmdArgs: string[], opts: RunOpts): Promise { + const proc = await this.spawn(cmdArgs, opts); + return collectOutput(proc, opts); + } + + public async spawn(cmdArgs: string[], opts: RunOpts): Promise { return spawnRemote(this.execServer, cmdArgs, opts); } @@ -95,7 +100,7 @@ export class RemoteExecCtx implements ExecCtx { } } -export async function spawnRemote(execServer: vscode.ExecServer, cmdArgs: string[], opts: RunOpts) { +export async function spawnRemote(execServer: vscode.ExecServer, cmdArgs: string[], opts: RunOpts): Promise { cmdCount += 1; const cmdId = cmdCount; @@ -117,41 +122,28 @@ export async function spawnRemote(execServer: vscode.ExecServer, cmdArgs: string env: { ...env, BUILDKIT_PROGRESS: "plain" }, }); - if (opts.stdin !== undefined) { - spawned.stdin.write(new TextEncoder().encode(opts.stdin)); - spawned.stdin.end(); - } - - let stdout = ""; - let stderr = ""; - const stdoutDecoder = new TextDecoder("utf-8"); - const stderrDecoder = new TextDecoder("utf-8"); - - spawned.stdout.onDidReceiveMessage((data: Uint8Array) => { - const chunk = stdoutDecoder.decode(data, { stream: true }); - stdout += chunk; - log.debug(`RemoteExecCtx.stdout.onDidReceiveMessage[${cmdStr()}]: len ${chunk.length}, chunk: '${chunk.trim()}'`); - }); - - spawned.stderr.onDidReceiveMessage((data: Uint8Array) => { - const chunk = stderrDecoder.decode(data, { stream: true }); - stderr += chunk; - log.debug(`RemoteExecCtx.stderr.onDidReceiveMessage[${cmdStr()}]: len ${chunk.length}, chunk: '${chunk.trim()}'`); - }); - - const exit = await spawned.onExit; - const info: CmdResult = { - exit: exit.status, - stdout: stdout, - stderr: stderr, + return { + stdin: { + write(data) { spawned.stdin.write(data instanceof Buffer ? new Uint8Array(data) : data); }, + end() { spawned.stdin.end(); }, + }, + onStdout(cb) { + spawned.stdout.onDidReceiveMessage((d) => { + cb(Buffer.from(d)); + }); + }, + onStderr(cb) { + spawned.stderr.onDidReceiveMessage((d) => { + cb(Buffer.from(d)); + }); + }, + onError(cb) { void spawned.onExit.then(() => { }, (err: unknown) => { cb(err instanceof Error ? err : new Error(String(err))); }); }, + onExit(cb) { void spawned.onExit.then((e) => { cb(e.status); }); }, + onClose(cb) { + void spawned.onExit.then( + () => { cb(); }, + () => { cb(); }, + ); + }, }; - - if (exit.status === 0) { - log.info(`RemoteExecCtx[${cmdStr()}]: stdout: ${info.stdout.trim()}`); - } - else { - log.error(`RemoteExecCtx[${cmdStr()}]: ${formatCmdErr(info)}`); - } - - return { exit: exit.status, stdout, stderr }; } diff --git a/src/common/opts.ts b/src/common/opts.ts index 0fc3444..614cf31 100644 --- a/src/common/opts.ts +++ b/src/common/opts.ts @@ -1,14 +1,5 @@ -import { LogOutputChannel } from "vscode"; - export type Envs = Record; -export interface SpawnOpts { - env?: Envs, - cwd?: string, - stdin?: string | undefined, - log: LogOutputChannel, -} - export interface RunOpts { cwd?: string, env?: Envs, diff --git a/src/common/spawn.ts b/src/common/spawn.ts index 2b90165..4236bac 100644 --- a/src/common/spawn.ts +++ b/src/common/spawn.ts @@ -1,91 +1,60 @@ -import * as chproc from "node:child_process"; -import { CmdResult, SpawnOpts } from "./opts"; +import { StringDecoder } from "node:string_decoder"; +import { CmdResult, RunOpts } from "./opts"; +import { getLogSink } from "../extension/log"; +import type { SpawnedProcess } from "./ctx/execCtx"; let cmdCount: number = 0; -/* eslint-disable @typescript-eslint/no-confusing-void-expression */ -export function spawn( - cmd: string, - args: string[], - opts: SpawnOpts, -): Promise { - return new Promise((resolve, _) => { +export function collectOutput(proc: SpawnedProcess, opts?: RunOpts): Promise { + return new Promise((resolve) => { cmdCount += 1; const cmdId = cmdCount; + const log = getLogSink(); + const tag = `CMD${String(cmdId).padStart(4, "0")}`; - const cmdStr = () => `[CMD${String(cmdId).padStart(4, "0")}]:`; + let stdout = ""; + let stderr = ""; + const stdoutDecoder = new StringDecoder("utf-8"); + const stderrDecoder = new StringDecoder("utf-8"); - let stdout: string = ""; - let stderr: string = ""; - - // TODO: do we need env without inheriting parent's env? - const finalEnv = { - ...process.env, - ...(opts.env ?? {}), - BUILDKIT_PROGRESS: "plain", - }; - - const proc = chproc.spawn(cmd, args, { - cwd: opts.cwd, - env: finalEnv, - stdio: "pipe", - }); - proc.stdout.setEncoding("utf-8"); - proc.stderr.setEncoding("utf-8"); - - if (opts.stdin !== undefined) { - proc.stdin.write(opts.stdin); + if (opts?.stdin !== undefined) { + proc.stdin.write(Buffer.from(opts.stdin)); proc.stdin.end(); } - proc.on("spawn", () => { - const strz_args = args.map(e => `'${e}'`).join(", "); - // TODO: log env values as well - opts.log.info(`${cmdStr()} Running (spawn) ['${cmd}', ${strz_args}]`); + proc.onStdout((d) => { + const chunk = stdoutDecoder.write(d); + stdout += chunk; + log.debug(`${tag} stdout: ${chunk.trimEnd()}`); }); - proc.on("error", (err: NodeJS.ErrnoException) => { - const msg = `${err.code} :: ${err.message} :: :: ${err.syscall}`; - const res: CmdResult = { - exit: err.errno ?? 255, - stdout: "", - stderr: err.message, - }; - - opts.log.error(cmdStr(), msg); - // TODO: reject? - return resolve(res); + proc.onStderr((d) => { + const chunk = stderrDecoder.write(d); + stderr += chunk; + log.debug(`${tag} stderr: ${chunk.trimEnd()}`); }); - proc.stdout.on("data", (data: string) => { - stdout += data; - opts.log.info(cmdStr(), data); + proc.onError((err) => { + log.error(`${tag} error: ${err.message}`); + resolve({ exit: 255, stdout: "", stderr: err.message }); }); - proc.stderr.on("data", (data: string) => { - stderr += data; - opts.log.error(cmdStr(), data); - }); - - proc.on("exit", (code, signal) => { - const res: CmdResult = { - exit: code ?? (signal ?? 256), - stdout: stdout, - stderr: stderr, - }; + let exitCode: number = 256; - if (code !== 0) { - opts.log.error(`${cmdStr()} Command failed with {code / signal ${res.exit}}`); - } - else { - opts.log.info(`${cmdStr()} command exit: ${res.exit}`); + proc.onExit((code) => { + exitCode = code ?? 256; + if (exitCode !== 0) { + log.error(`${tag} exited ${exitCode}`); } + }); - return resolve(res); + proc.onClose(() => { + stdout += stdoutDecoder.end(); + stderr += stderrDecoder.end(); + resolve({ exit: exitCode, stdout, stderr }); }); }); } -/* eslint-enable @typescript-eslint/no-confusing-void-expression */ export function formatCmdErr(res: CmdResult): string { return `Error: ${res.exit}: stdout: [${res.stdout.trim()}] stderr: [${res.stderr.trim()}]`; diff --git a/src/common/tests/spawn.test.ts b/src/common/tests/spawn.test.ts index 02909b3..c495adc 100644 --- a/src/common/tests/spawn.test.ts +++ b/src/common/tests/spawn.test.ts @@ -1,19 +1,11 @@ import { describe, expect, test } from "vitest"; -import { spawn } from "../spawn"; +import { run } from "../cmd"; import { getLogSink } from "../../extension/log"; import { ContainerInspectResult } from "../../engine/lifecycle"; import { initMocks, jsonFormat, getMockSettings, getTestTimeout } from "../../tests/common"; -function getcwd() { - return __dirname; -} - -function opts() { - return { cwd: getcwd(), env: {}, log: getLogSink() }; -} - const bashSleepCmd = ["bash", "-c", "trap 'exit 0' SIGINT SIGTERM; while true; do sleep 1; done"]; const SETTINGS = getMockSettings(); @@ -27,34 +19,40 @@ describe.skipIf(!SETTINGS.dockerPath)("cmd spawn tests", () => { }); test("env buildkit", async () => { - const out = await spawn("env", [], opts()); + const out = await run(["env"], {}); expect(out.stdout).contains("BUILDKIT_PROGRESS=plain"); expect(out.exit).eq(0); }); + test("stdin/stdout sanity", async () => { + const out = await run(["cat"], { stdin: "HELLO\nWORLD" }); + expect(out.stdout).eq("HELLO\nWORLD"); + expect(out.exit).eq(0); + }); + test("get engine version", async () => { - const out = await spawn(engine, ["version", ...jsonFormat], opts()); + const out = await run([engine, "version", ...jsonFormat], {}); expect(out.exit).eq(0); }); test("create and remove container", async () => { - const create = await spawn(engine, ["create", "hello-world"], opts()); + const create = await run([engine, "create", "hello-world"], {}); - const remove = await spawn(engine, ["rm", create.stdout.trim()], opts()); + const remove = await run([engine, "rm", create.stdout.trim()], {}); expect(create.exit).eq(0); expect(remove.exit).eq(0); }, getTestTimeout(10)); test("run and exec command", async () => { - const create = await spawn(engine, ["run", "--name", "turtles", "-d", "ubuntu:24.04", ...bashSleepCmd], opts()); + const create = await run([engine, "run", "--name", "turtles", "-d", "ubuntu:24.04", ...bashSleepCmd], {}); const containerId = create.stdout.trim(); try { - const exec = await spawn(engine, ["exec", create.stdout.trim(), "cat", "/etc/os-release"], opts()); + const exec = await run([engine, "exec", create.stdout.trim(), "cat", "/etc/os-release"], {}); expect(exec.stdout.trim()).toContain("Ubuntu 24.04"); expect(exec.exit).eq(0); - const inspect = await spawn(engine, ["inspect", containerId, ...jsonFormat], opts()); + const inspect = await run([engine, "inspect", containerId, ...jsonFormat], {}); expect(inspect.exit).eq(0); const inspectData = JSON.parse(inspect.stdout.trim()) as ContainerInspectResult; @@ -63,10 +61,10 @@ describe.skipIf(!SETTINGS.dockerPath)("cmd spawn tests", () => { expect(inspectData.State.Running).eq(true); } finally { - const stop = await spawn(engine, ["stop", "turtles"], opts()); + const stop = await run([engine, "stop", "turtles"], {}); expect(stop.exit).eq(0); - const rm = await spawn(engine, ["rm", "turtles"], opts()); + const rm = await run([engine, "rm", "turtles"], {}); expect(rm.exit).eq(0); } }, getTestTimeout(30)); diff --git a/src/engine/lifecycle.ts b/src/engine/lifecycle.ts index c15980b..3286079 100644 --- a/src/engine/lifecycle.ts +++ b/src/engine/lifecycle.ts @@ -1,19 +1,21 @@ import path from "node:path"; import * as fs from "node:fs/promises"; - import { run } from "../common/cmd"; +import { getExecCtx } from "../common/ctx/ctx"; import { getLogSink } from "../extension/log"; import { formatCmdErr } from "../common/spawn"; -import { parseEnv, getHostUserInfo, getProductJson, getEngineCmd } from "../common/utils"; +import { parseEnv, getHostUserInfo, getProductJson, getEngineCmd, fmtErr } from "../common/utils"; import { ContainerConfig, ContainerEngine, LifecycleCmd } from "./container"; import { EngineError, InstallError, InternalError } from "../extension/error"; import { NotificationLevel, showNotification } from "../extension/workspace"; import * as settings from "../common/settings"; import * as server from "../remote/installServer"; +import * as sshAgent from "../remote/exec-pipe/local"; import { EXTENSION_ID, DEVCONTAINER_SERVER_LISTEN_PORT, CAT_PIPE_STDIN_WORKAROUND } from "../common/constants"; export const STAGE2_DOCKERFILE_LOCATION: string = path.join(__dirname, "Dockerfile"); +export const SSH_RELAY_SCRIPT_LOCATION: string = path.join(__dirname, "remote.ts"); const UUID_TOKEN_LEN = 36; const jsonFormat = ["--format", "{{json .}}"]; @@ -618,6 +620,59 @@ export class ContainerState { } } + public async startSshAgentRelay(): Promise { + const sshAuthSock = process.env.SSH_AUTH_SOCK; + if (!sshAuthSock) { + getLogSink().info("ssh-agent-relay: SSH_AUTH_SOCK env not set, skipping"); + return undefined; + } + + const remoteEnv = this.remoteEnvProbe.HOME; + + if (!remoteEnv) { + throw new InternalError("ssh-agent-relay: HOME envvar not present in remoteEnv"); + } + + const nodeBin = path.posix.join(remoteEnv, ".vscode-oss-devcontainer", "node"); + const containerName = this.getContainerName(); + + try { + const sockPath = sshAgent.sshAgentSockPath(); + const bridgeScript = await fs.readFile(SSH_RELAY_SCRIPT_LOCATION, { encoding: "utf-8" }) + + "\nmain();"; + + getLogSink().debug(`ssh-agent-relay: bridge script contents:\n${bridgeScript}\n`); + + const execArgs = this.cc.getExecArgs(containerName, this.remoteEnvProbe, { interactive: true }); + const proc = await getExecCtx().spawn([ + ...this.cmd, + ...execArgs, + nodeBin, + "--input-type=module-typescript", + "-e", bridgeScript, + "--", + `--sock-path=${sockPath}`, + ], { cwd: this.workspaceFolder }); + + const relay = new sshAgent.SshAgentRelay(proc, sshAuthSock, sockPath); + + let timer: ReturnType; + await Promise.race([ + relay.ready.then(() => { clearTimeout(timer); }), + new Promise((_, reject) => { + timer = setTimeout(() => { reject(new Error("bridge startup timeout")); }, 10_000); + }), + ]); + + getLogSink().info("ssh-agent-relay: started"); + return relay; + } + catch (e: unknown) { + getLogSink().error(`ssh-agent-relay: ${fmtErr(e)}`); + return undefined; + } + } + public async dispose() { // TODO: handle shutdown in installServer.sh } diff --git a/src/remote/exec-pipe/local.ts b/src/remote/exec-pipe/local.ts new file mode 100644 index 0000000..27da14a --- /dev/null +++ b/src/remote/exec-pipe/local.ts @@ -0,0 +1,195 @@ +import * as net from "net"; +import path from "node:path"; + +import { randomUUID } from "node:crypto"; +import { getLogSink } from "../../extension/log"; +import { EXTENSION_ID } from "../../common/constants"; +import type { SpawnedProcess } from "../../common/ctx/execCtx"; + +import { + FrameType, + HEADER_SIZE, + TYPE_OFFSET, + CHANNEL_OFFSET, + LENGTH_OFFSET, +} from "./remote"; + +// Location of socket in container. +export const SSH_AGENT_SOCK_DIR = `/tmp/codium-${EXTENSION_ID}-ipc`; + +export function sshAgentSockPath(): string { + return path.posix.join(SSH_AGENT_SOCK_DIR, `ssh-agent-fwd-${randomUUID()}.sock`); +} + +function getMsg(msg: string) { + return `[ssh-agent-relay/host]: ${msg}`; +} + +/** + * We create a relay per-window, so stale sockets might accumulate + * but that's more reliable as we won't have to maintain state + * that's not tied to window lifetime. Using a socket that's tied + * to the container lifecycle is a lil more painful, especially for + * devcontainers over ssh. + * + * For a remote over SSH, we exec over the ssh connection via the + * SSH-extension's execServer provider, so the connection lifetime + * is necessarily bound to the window, meaning we can't exec a + * forwarded "outside" the window's lifetime, unlike in local case. + * So just use window-bound lifetime for all sockets for consistency. + * Also prevents buggy cases with unstable connections. + * + * local.ts is the workspace-local relay-end that connects to the + * socket. "workspace-local" on a remote-ssh machine = on the remote + * ssh machine. Uses docker exec and shuttles data over stdin/stdout + * to the socket in the container. The container-side listener is + * in remote.ts. + * +**/ +export class SshAgentRelay { + private readonly agentConns = new Map(); + private inBuf = Buffer.alloc(0); + private disposed = false; + private readyResolve: (() => void) | undefined; + private readyReject: ((err: Error) => void) | undefined; + readonly ready: Promise; + + constructor( + private readonly proc: SpawnedProcess, + private readonly sshAuthSock: string, + readonly containerSockPath: string, + ) { + this.ready = new Promise((resolve, reject) => { + this.readyResolve = resolve; + this.readyReject = reject; + }); + + proc.onStdout((d) => { this.onData(d); }); + proc.onStderr((d) => { + getLogSink().info(getMsg(d.toString().trim())); + }); + proc.onExit((code) => { + if (!this.disposed) { + getLogSink().warn(getMsg(`bridge exited (code ${code})`)); + } + this.disposed = true; + this.readyReject?.(new Error(`bridge exited (code ${code}) before Ready`)); + this.readyReject = undefined; + this.cleanup(); + }); + proc.onError((err) => { + getLogSink().error(getMsg(`bridge error: ${err.message}`)); + }); + + getLogSink().info(getMsg(`started, agent=${this.sshAuthSock}`)); + } + + private onData(d: Buffer): void { + this.inBuf = Buffer.concat([this.inBuf, d]); + + while (this.inBuf.length >= HEADER_SIZE) { + const type = this.inBuf.readUInt16BE(TYPE_OFFSET) as FrameType; + if (type >= FrameType.LAST) { + getLogSink().warn(getMsg(`invalid frame type ${type}, dropping buffer`)); + this.inBuf = Buffer.alloc(0); + break; + } + + const chan = this.inBuf.readUInt32BE(CHANNEL_OFFSET); + const len = this.inBuf.readUInt32BE(LENGTH_OFFSET); + + if (this.inBuf.length < HEADER_SIZE + len) { break; } + + const payload = len > 0 + ? Buffer.from(this.inBuf.subarray(HEADER_SIZE, HEADER_SIZE + len)) + : null; + + this.inBuf = this.inBuf.subarray(HEADER_SIZE + len); + this.handleFrame(type, chan, payload); + } + } + + private handleFrame(type: FrameType, chan: number, payload: Buffer | null): void { + if (type === FrameType.Ready) { + getLogSink().info(getMsg("bridge ready")); + this.readyResolve?.(); + this.readyResolve = undefined; + this.readyReject = undefined; + return; + } + if (type === FrameType.Open) { + getLogSink().debug(getMsg(`ch${chan} Open -> connecting to agent`)); + this.openAgentConn(chan); + } + else if (type === FrameType.Data && payload) { + getLogSink().debug(getMsg(`ch${chan} Data ${payload.length}B -> agent`)); + this.forwardToAgent(chan, payload); + } + else if (type === FrameType.Close) { + getLogSink().debug(getMsg(`ch${chan} Close`)); + this.closeAgentConn(chan); + } + } + + private openAgentConn(chan: number): void { + const conn = net.createConnection(this.sshAuthSock); + this.agentConns.set(chan, conn); + + conn.on("connect", () => { + getLogSink().debug(getMsg(`ch${chan} agent connected`)); + }); + conn.on("data", (d: Buffer) => { + getLogSink().debug(getMsg(`ch${chan} agent->bridge ${d.length}B`)); + this.writeFrame(FrameType.Data, chan, d); + }); + conn.on("error", (err) => { + getLogSink().warn(getMsg(`ch${chan} agent error: ${err.message}`)); + this.writeFrame(FrameType.Close, chan, null); + this.agentConns.delete(chan); + }); + conn.on("close", () => { + getLogSink().debug(getMsg(`ch${chan} agent closed`)); + if (this.agentConns.delete(chan)) { + this.writeFrame(FrameType.Close, chan, null); + } + }); + } + + private forwardToAgent(chan: number, data: Buffer): void { + const conn = this.agentConns.get(chan); + if (conn) { conn.write(data); } + } + + private closeAgentConn(chan: number): void { + const conn = this.agentConns.get(chan); + if (conn) { + conn.destroy(); + this.agentConns.delete(chan); + } + } + + private writeFrame(type: FrameType, chan: number, data: Buffer | null): void { + if (this.disposed) { return; } + + const hdr = Buffer.alloc(HEADER_SIZE); + hdr.writeUInt16BE(type, TYPE_OFFSET); + hdr.writeUInt32BE(chan, CHANNEL_OFFSET); + hdr.writeUInt32BE(data ? data.length : 0, LENGTH_OFFSET); + this.proc.stdin.write(hdr); + + if (data && data.length > 0) { this.proc.stdin.write(data); } + } + + private cleanup(): void { + for (const [, conn] of this.agentConns) { conn.destroy(); } + this.agentConns.clear(); + } + + dispose(): void { + if (this.disposed) { return; } + this.disposed = true; + getLogSink().info(getMsg(`disposing (${this.agentConns.size} active agent conns)`)); + this.proc.stdin.end(); + this.cleanup(); + } +} diff --git a/src/remote/exec-pipe/remote.ts b/src/remote/exec-pipe/remote.ts new file mode 100644 index 0000000..28deca2 --- /dev/null +++ b/src/remote/exec-pipe/remote.ts @@ -0,0 +1,194 @@ +import net from "net"; +import fs from "fs"; + +export const FrameType = { + Ready: 0, + Open: 1, + Data: 2, + Close: 3, + LAST: 4, + MAX: 15, +} as const; +export type FrameType = (typeof FrameType)[keyof typeof FrameType]; + +/** + * remote.ts: container-side socket, shuttles data to the + * local.ts on the host via docker exec stdin/stdout + * + * All types are exported here and used in local.ts to make + * remote.ts standalone enough that it can be passed to node -e + * without extra setup. We could compile this, embed into the + * extension and send it in, but for transparency and quick + * debugging, it's easier to keep the .ts file. Also makes + * it easier to monkey-patch in the wild for quick-fixes. + * + * Format is simple: [ type(2B), channel(4B), length(4B) ] + */ + +export const TYPE_SIZE = 2; +export const CHAN_SIZE = 4; +export const LEN_SIZE = 4; +export const HEADER_SIZE = TYPE_SIZE + CHAN_SIZE + LEN_SIZE; + +export const TYPE_OFFSET = 0; +export const CHANNEL_OFFSET = TYPE_OFFSET + TYPE_SIZE; +export const LENGTH_OFFSET = CHANNEL_OFFSET + CHAN_SIZE; + +function parseArgs(argv: string[]): { sock: string, disableNodeSafety: boolean } { + let sock = ""; + let disableNodeSafety = false; + for (const arg of argv) { + if (arg.startsWith("--sock-path=")) { sock = arg.slice("--sock-path=".length); } + else if (arg === "--disable-node-safety") { disableNodeSafety = true; } + } + return { sock, disableNodeSafety }; +} + +export function main() { + const args = parseArgs(process.argv.slice(1)); + if (!args.sock) { + console.error("[ssh-agent-relay/container]: --sock-path= is required"); + process.exit(1); + } + const SOCK = args.sock; + + // stderr for all logs because stdio/stdout are for the data + const log = (msg: string) => { console.error(`[ssh-agent-relay/container]: ${msg}`); }; + + function cleanupSocket() { + try { + if (SOCK.startsWith("/tmp/")) { + fs.unlinkSync(SOCK); + } + else { + log(`[ERROR]: SOCK not in /tmp/, refusing to unlink ${SOCK}`); + } + } + catch { } + } + + function acquireSocket() { + const dir = SOCK.substring(0, SOCK.lastIndexOf("/")); + try { fs.mkdirSync(dir, { recursive: true }); } + catch { } + + try { + if (SOCK.startsWith("/tmp/")) { + fs.unlinkSync(SOCK); + } + else { + log(`[ERROR]: SOCK not in /tmp/, refusing to unlink ${SOCK}`); + } + } + catch { } + } + + let nextChan = 0; + const clients = new Map(); + + function writeFrame(type: FrameType, chan: number, data: Buffer | null) { + const header = Buffer.alloc(HEADER_SIZE); + header.writeUInt16BE(type, TYPE_OFFSET); + header.writeUInt32BE(chan, CHANNEL_OFFSET); + header.writeUInt32BE(data ? data.length : 0, LENGTH_OFFSET); + process.stdout.write(header); + if (data && data.length > 0) { process.stdout.write(data); } + } + + function closeClient(chan: number) { + const sock = clients.get(chan); + if (!sock) { return; } + + sock.destroy(); + clients.delete(chan); + writeFrame(FrameType.Close, chan, null); + } + + let buf = Buffer.alloc(0); + + process.stdin.on("data", (d: Buffer) => { + buf = Buffer.concat([buf, d]); + + while (buf.length >= HEADER_SIZE) { + const type: FrameType = buf.readUInt16BE(TYPE_OFFSET) as FrameType; + if (type >= FrameType.LAST) { + log(`[ERROR]: invalid frame type ${type}, dropping buffer`); + buf = Buffer.alloc(0); + break; + } + + const chan = buf.readUInt32BE(CHANNEL_OFFSET); + const len = buf.readUInt32BE(LENGTH_OFFSET); + if (buf.length < HEADER_SIZE + len) { break; } + + const payload = len > 0 ? Buffer.from(buf.subarray(HEADER_SIZE, HEADER_SIZE + len)) : null; + buf = buf.subarray(HEADER_SIZE + len); + + const sock = clients.get(chan); + if (!sock) { + log(`[ERROR]: stdin: ch${chan} type=${type}: unknown channel, skipping frame`); + continue; + } + + if (type === FrameType.Data && payload) { + log(`stdin->sock: ch${chan} ${payload.length}B`); + sock.write(payload); + } + else if (type === FrameType.Close) { + log(`stdin: ch${chan} Close`); + sock.destroy(); + clients.delete(chan); + } + } + }); + + const serverOpts = args.disableNodeSafety ? { allowHalfOpen: true } : {}; + const server = net.createServer(serverOpts, (sock) => { + const chan = nextChan++; + + clients.set(chan, sock); + log(`client connected: ch${chan} (active: ${clients.size})`); + + writeFrame(FrameType.Open, chan, null); + + sock.on("data", (d) => { + log(`sock->stdout: ch${chan} ${d.length}B`); + writeFrame(FrameType.Data, chan, d); + }); + sock.on("end", () => { + log(`client disconnected: ch${chan} (active: ${clients.size - 1})`); + closeClient(chan); + }); + sock.on("error", (err) => { + log(`[ERROR]: client [error]: ch${chan} ${(err as NodeJS.ErrnoException).code ?? err}`); + closeClient(chan); + }); + }); + + process.on("SIGTERM", () => { + log("SIGTERM received, shutting down"); + server.close(); + for (const [, s] of clients) { s.destroy(); } + cleanupSocket(); + process.exit(0); + }); + + process.stdin.on("end", () => { + log(`stdin closed, shutting down (active clients: ${clients.size})`); + server.close(); + for (const [, s] of clients) { s.destroy(); } + cleanupSocket(); + process.exit(0); + }); + + acquireSocket(); + server.listen(SOCK, () => { + try { fs.chmodSync(SOCK, 0o600); } + catch { } + + // TODO: symlink agent.sock to the latest working socket? + + log(`listening on ${SOCK}`); + writeFrame(FrameType.Ready, 0, null); + }); +} diff --git a/src/remote/exec-pipe/tests/relay-engine.test.ts b/src/remote/exec-pipe/tests/relay-engine.test.ts new file mode 100644 index 0000000..d63f914 --- /dev/null +++ b/src/remote/exec-pipe/tests/relay-engine.test.ts @@ -0,0 +1,179 @@ +import * as net from "net"; +import * as fs from "fs"; +import * as childProcess from "child_process"; +import path from "node:path"; + +import { describe, test, expect, afterAll, afterEach } from "vitest"; + +import { run } from "../../../common/cmd"; +import { getMockSettings, getTestTimeout, initMocks } from "../../../tests/common"; +import type { SpawnedProcess } from "../../../common/ctx/execCtx"; + +import { SshAgentRelay, SSH_AGENT_SOCK_DIR } from "../local"; + +initMocks(); + +const SETTINGS = getMockSettings(); +const CONTAINER_IMAGE = "node:24-slim"; + +const CONTAINER_NAME = `ord-relay-test-${process.pid}`; +const AGENT_SOCK_PATH = `/tmp/relay-engine-agent-${process.pid}.sock`; +const CONTAINER_SOCK_PATH = `${SSH_AGENT_SOCK_DIR}/test-${process.pid}.sock`; + +function readBridgeScript(): string { + const SSH_RELAY_SCRIPT_LOCATION = path.join(__dirname, "../remote.ts"); + return fs.readFileSync(SSH_RELAY_SCRIPT_LOCATION, "utf-8") + "\nmain();"; +} + +function createFakeAgent(): Promise { + return new Promise((resolve) => { + try { fs.unlinkSync(AGENT_SOCK_PATH); } + catch { } + const server = net.createServer((sock) => { + sock.on("data", (d) => { + sock.write(Buffer.concat([Buffer.from("AGENT:"), d])); + }); + }); + server.listen(AGENT_SOCK_PATH, () => { resolve(server); }); + }); +} + +function wrapChildProcess(proc: childProcess.ChildProcess): SpawnedProcess { + return { + stdin: { + write(data) { proc.stdin?.write(data); }, + end() { proc.stdin?.end(); }, + }, + onStdout(cb) { proc.stdout?.on("data", cb); }, + onStderr(cb) { proc.stderr?.on("data", cb); }, + onError(cb) { proc.on("error", cb); }, + onExit(cb) { proc.on("exit", cb); }, + onClose(cb) { proc.on("close", cb); }, + }; +} + +function startBridge(engine: string): SpawnedProcess { + const bridgeScript = readBridgeScript(); + const proc = childProcess.spawn(engine, [ + "exec", "-i", CONTAINER_NAME, + "node", + "--input-type=module-typescript", + "-e", bridgeScript, + "--", + `--sock-path=${CONTAINER_SOCK_PATH}`, + ], { + stdio: ["pipe", "pipe", "pipe"], + }); + return wrapChildProcess(proc); +} + +async function waitForRelay(r: SshAgentRelay, timeoutMs = 15_000): Promise { + let timer: ReturnType; + await Promise.race([ + r.ready.then(() => { clearTimeout(timer); }), + new Promise((_, reject) => { + timer = setTimeout(() => { reject(new Error("bridge startup timeout")); }, timeoutMs); + }), + ]); +} + +// client script that runs inside the container, connects to the bridge socket, +// sends a message, waits for one response, prints it to stdout +function containerSocketClientScript(id: number): string { + return [ + "import net from \"net\";", + "const timer = setTimeout(() => { process.stderr.write(\"timeout\"); process.exit(1); }, 10000);", + `const sock = net.createConnection("${CONTAINER_SOCK_PATH}", () => {`, + ` sock.write("ping-${id}");`, + "});", + "sock.once(\"data\", (d) => {", + " clearTimeout(timer);", + " process.stdout.write(d.toString());", + " sock.destroy();", + "});", + "sock.on(\"error\", (e) => {", + " clearTimeout(timer);", + " process.stderr.write(e.message);", + " process.exit(1);", + "});", + ].join("\n"); +} + +let agentServer: net.Server | undefined; +let relay: SshAgentRelay | undefined; + +describe.skipIf(!SETTINGS.dockerPath)("exec-pipe engine integration", () => { + const engine = SETTINGS.dockerPath; + + afterAll(async () => { + relay?.dispose(); + relay = undefined; + agentServer?.close(); + agentServer = undefined; + try { fs.unlinkSync(AGENT_SOCK_PATH); } + catch { } + await run([engine, "container", "stop", "-t", "2", CONTAINER_NAME], {}); + await run([engine, "container", "rm", "--force", CONTAINER_NAME], {}); + }); + + afterEach(async () => { + relay?.dispose(); + relay = undefined; + agentServer?.close(); + agentServer = undefined; + await run([engine, "exec", CONTAINER_NAME, "rm", "-f", CONTAINER_SOCK_PATH], {}); + }); + + test("start container", async () => { + const result = await run([ + engine, "run", "-d", + "--name", CONTAINER_NAME, + "--entrypoint", "/bin/sh", + CONTAINER_IMAGE, + "-c", 'trap "exit 0" SIGINT SIGTERM; while sleep 60 & wait $! ; do : ; done', + ], {}); + expect(result.exit).toBe(0); + }, getTestTimeout(60)); + + test("100 concurrent clients through container relay", async () => { + agentServer = await createFakeAgent(); + const proc = startBridge(engine); + relay = new SshAgentRelay(proc, AGENT_SOCK_PATH, CONTAINER_SOCK_PATH); + await waitForRelay(relay); + + const NUM_CLIENTS = 100; + + const results = await Promise.all( + Array.from({ length: NUM_CLIENTS }, async (_, i) => { + const r = await run([ + engine, "exec", CONTAINER_NAME, + "node", "--input-type=module-typescript", + "-e", containerSocketClientScript(i), + ], {}); + return { i, exit: r.exit, stdout: r.stdout, stderr: r.stderr }; + }), + ); + + for (const r of results) { + expect(r.exit, `client ${r.i} failed: ${r.stderr}`).toBe(0); + expect(r.stdout).toBe(`AGENT:ping-${r.i}`); + } + }, getTestTimeout(120)); + + test("sequential rapid connect/disconnect", async () => { + agentServer = await createFakeAgent(); + const proc = startBridge(engine); + relay = new SshAgentRelay(proc, AGENT_SOCK_PATH, CONTAINER_SOCK_PATH); + await waitForRelay(relay); + + for (let i = 0; i < 50; i++) { + const r = await run([ + engine, "exec", CONTAINER_NAME, + "node", "--input-type=module-typescript", + "-e", containerSocketClientScript(i), + ], {}); + expect(r.exit, `client ${i} failed: ${r.stderr}`).toBe(0); + expect(r.stdout).toBe(`AGENT:ping-${i}`); + } + }, getTestTimeout(120)); +}); diff --git a/src/remote/exec-pipe/tests/relay.test.ts b/src/remote/exec-pipe/tests/relay.test.ts new file mode 100644 index 0000000..4d6584a --- /dev/null +++ b/src/remote/exec-pipe/tests/relay.test.ts @@ -0,0 +1,395 @@ +import * as net from "net"; +import * as fs from "fs"; +import * as path from "path"; +import * as childProcess from "child_process"; + +import { describe, test, expect, afterEach } from "vitest"; + +import { SshAgentRelay } from "../local"; +import { getTestTimeout, initMocks } from "../../../tests/common"; +import type { SpawnedProcess } from "../../../common/ctx/execCtx"; +import { + CHANNEL_OFFSET, + FrameType, + HEADER_SIZE, + LENGTH_OFFSET, + TYPE_OFFSET, +} from "../remote"; + +initMocks(); + +const SOCK_PATH = `/tmp/relay-test-${process.pid}.sock`; +const AGENT_SOCK_PATH = `/tmp/relay-test-agent-${process.pid}.sock`; + +interface Frame { + type: FrameType, + chan: number, + payload: Buffer | null, +} + +function localReadFrames(buf: Buffer): { frames: Frame[], remainder: Buffer } { + const frames: Frame[] = []; + let offset = 0; + while (offset + HEADER_SIZE <= buf.length) { + const type = buf.readUInt16BE(offset + TYPE_OFFSET) as FrameType; + const chan = buf.readUInt32BE(offset + CHANNEL_OFFSET); + const len = buf.readUInt32BE(offset + LENGTH_OFFSET); + if (offset + HEADER_SIZE + len > buf.length) { break; } + + const payload = len > 0 + ? Buffer.from(buf.subarray(offset + HEADER_SIZE, offset + HEADER_SIZE + len)) + : null; + + frames.push({ type, chan, payload }); + offset += HEADER_SIZE + len; + } + return { frames, remainder: Buffer.from(buf.subarray(offset)) }; +} + +function localWriteFrame(stream: NodeJS.WritableStream, type: FrameType, chan: number, data: Buffer | null) { + const hdr = Buffer.alloc(HEADER_SIZE); + hdr.writeUInt16BE(type, TYPE_OFFSET); + hdr.writeUInt32BE(chan, CHANNEL_OFFSET); + hdr.writeUInt32BE(data ? data.length : 0, LENGTH_OFFSET); + stream.write(hdr); + if (data && data.length > 0) { stream.write(data); } +} + +function waitForFrames(proc: childProcess.ChildProcess, count: number, timeoutMs = 2000): Promise { + return new Promise((resolve, reject) => { + const frames: Frame[] = []; + let buf: Buffer = Buffer.alloc(0); + const timer = setTimeout(() => { + reject(new Error(`timeout waiting for ${count} frames, got ${frames.length}: ${JSON.stringify(frames)}`)); + }, timeoutMs); + + if (!proc.stdout) { throw new Error("proc.stdout is null"); } + proc.stdout.on("data", (d: Buffer) => { + buf = Buffer.concat([buf, d]); + const result = localReadFrames(buf); + frames.push(...result.frames); + buf = result.remainder; + if (frames.length >= count) { + clearTimeout(timer); + resolve(frames.slice(0, count)); + } + }); + }); +} + +function spawnBridge(opts?: { disableNodeSafety?: boolean }): childProcess.ChildProcess { + const REMOTE_SCRIPT = path.resolve(__dirname, "..", "remote.ts"); + const script = fs.readFileSync(REMOTE_SCRIPT, "utf-8") + "\nmain();"; + + const nodeArgs = [ + "--input-type=module-typescript", + "-e", script, + "--", + `--sock-path=${SOCK_PATH}`, + (opts?.disableNodeSafety ? "--disable-node-safety" : ""), + ]; + + return childProcess.spawn("node", nodeArgs, { + stdio: ["pipe", "pipe", "pipe"], + }); +} + +function createFakeAgent(): Promise { + return new Promise((resolve) => { + try { fs.unlinkSync(AGENT_SOCK_PATH); } + catch { } + const server = net.createServer((sock) => { + sock.on("data", (d) => { + // echo back with a prefix + sock.write(Buffer.concat([Buffer.from("AGENT:"), d])); + }); + }); + server.listen(AGENT_SOCK_PATH, () => { resolve(server); }); + }); +} + +let bridge: childProcess.ChildProcess | undefined; +let agentServer: net.Server | undefined; + +afterEach(() => { + bridge?.kill(); + bridge = undefined; + agentServer?.close(); + agentServer = undefined; + try { fs.unlinkSync(SOCK_PATH); } + catch { } + try { fs.unlinkSync(AGENT_SOCK_PATH); } + catch { } +}); + +describe("exec-pipe relay", () => { + test("sanity", () => { + // if this changes, then ensure correct handling of [0, 3] + expect(FrameType.LAST).eq(4); + + // if this changes, then type field needs to be 3 + expect(FrameType.MAX).eq(15); + }, getTestTimeout(10)); + + test("bridge sends Ready frame on startup", async () => { + bridge = spawnBridge(); + const frames = await waitForFrames(bridge, 1); + expect(frames[0].type).toBe(FrameType.Ready); + expect(frames[0].chan).toBe(0); + }, getTestTimeout(10)); + + test("bridge sends Open frame when client connects to socket", async () => { + bridge = spawnBridge(); + const [ready] = await waitForFrames(bridge, 1); + expect(ready.type).toBe(FrameType.Ready); + + // connect to the bridge's socket + const client = net.createConnection(SOCK_PATH); + await new Promise((resolve) => { client.on("connect", resolve); }); + + const frames = await waitForFrames(bridge, 1); + expect(frames[0].type).toBe(FrameType.Open); + + client.destroy(); + }, getTestTimeout(10)); + + test("bridge forwards data from client to stdout", async () => { + bridge = spawnBridge(); + await waitForFrames(bridge, 1); // Ready + + const client = net.createConnection(SOCK_PATH); + await new Promise((resolve) => { client.on("connect", resolve); }); + + const [open] = await waitForFrames(bridge, 1); + expect(open.type).toBe(FrameType.Open); + const chan = open.chan; + + client.write(Buffer.from("hello")); + + const [data] = await waitForFrames(bridge, 1); + expect(data.type).toBe(FrameType.Data); + expect(data.chan).toBe(chan); + expect(data.payload).not.toBeNull(); + expect(data.payload?.toString()).toBe("hello"); + + client.destroy(); + }, getTestTimeout(10)); + + test("bridge forwards data from stdin to client", async () => { + bridge = spawnBridge(); + await waitForFrames(bridge, 1); // Ready + + const client = net.createConnection(SOCK_PATH); + await new Promise((resolve) => { client.on("connect", resolve); }); + + const [open] = await waitForFrames(bridge, 1); + const chan = open.chan; + + const clientData = new Promise((resolve, reject) => { + const timer = setTimeout(() => { reject(new Error("timeout waiting for client data")); }, 2000); + client.once("data", (d) => { + clearTimeout(timer); + resolve(d); + }); + }); + + if (!bridge.stdin) { throw new Error("bridge.stdin is null"); } + localWriteFrame(bridge.stdin, FrameType.Data, chan, Buffer.from("world")); + + const received = await clientData; + expect(received.toString()).toBe("world"); + + client.destroy(); + }, getTestTimeout(10)); + + test("full roundtrip through fake agent", async () => { + agentServer = await createFakeAgent(); + bridge = spawnBridge(); + await waitForFrames(bridge, 1); // Ready + + // client connects to bridge socket (simulates ssh-add inside container) + const client = net.createConnection(SOCK_PATH); + await new Promise((resolve) => { client.on("connect", resolve); }); + + const [open] = await waitForFrames(bridge, 1); + const chan = open.chan; + + // client sends data -> bridge stdout -> we read Open+Data frames + client.write(Buffer.from("list-keys")); + + const [dataFrame] = await waitForFrames(bridge, 1); + expect(dataFrame.type).toBe(FrameType.Data); + expect(dataFrame.payload?.toString()).toBe("list-keys"); + + // simulate host side: forward to real agent, get response, send back + const agentConn = net.createConnection(AGENT_SOCK_PATH); + await new Promise((resolve) => { agentConn.on("connect", resolve); }); + + const agentResponse = new Promise((resolve, reject) => { + const timer = setTimeout(() => { reject(new Error("timeout waiting for agent response")); }, 2000); + agentConn.once("data", (d) => { + clearTimeout(timer); + resolve(d); + }); + }); + + agentConn.write(dataFrame.payload ?? Buffer.alloc(0)); + const response = await agentResponse; + expect(response.toString()).toBe("AGENT:list-keys"); + + // send response back through bridge stdin -> bridge socket -> client + const clientResponse = new Promise((resolve, reject) => { + const timer = setTimeout(() => { reject(new Error("timeout waiting for client response")); }, 2000); + client.once("data", (d) => { + clearTimeout(timer); + resolve(d); + }); + }); + + if (!bridge.stdin) { throw new Error("bridge.stdin is null"); } + localWriteFrame(bridge.stdin, FrameType.Data, chan, response); + + const finalResponse = await clientResponse; + expect(finalResponse.toString()).toBe("AGENT:list-keys"); + + agentConn.destroy(); + client.destroy(); + }, getTestTimeout(10)); + + test("bridge sends Close frame when client disconnects", async () => { + bridge = spawnBridge(); + await waitForFrames(bridge, 1); // Ready + + const client = net.createConnection(SOCK_PATH); + await new Promise((resolve) => { client.on("connect", resolve); }); + + await waitForFrames(bridge, 1); // Open + + client.destroy(); + + const [close] = await waitForFrames(bridge, 1); + expect(close.type).toBe(FrameType.Close); + }, getTestTimeout(10)); +}); + +describe("stability edge-cases", () => { + test("ready rejects when bridge exits before 'FrameType.Ready'", async () => { + let exitCb: ((code: number | null) => void) | undefined; + + const mockProc: SpawnedProcess = { + stdin: { write() { }, end() { } }, + onStdout() { }, + onStderr() { }, + onError() { }, + onExit(cb) { exitCb = cb; }, + onClose() { }, + }; + + const relay = new SshAgentRelay(mockProc, "/tmp/fake-agent.sock", "/tmp/fake.sock"); + + // bridge dies before sending Ready + exitCb?.(1); + + const result = await Promise.race([ + relay.ready.then(() => "resolved").catch(() => "rejected"), + new Promise(r => setTimeout(() => { r("hung"); }, 500)), + ]); + + // ready promise never rejects = it hangs forever + expect(result).toBe("rejected"); + + relay.dispose(); + }); + + test("closeClient sends exactly one 'FrameType.Close' per disconnect", async () => { + bridge = spawnBridge({ disableNodeSafety: true }); + + const allFrames: Frame[] = []; + let parseBuf: Buffer = Buffer.alloc(0); + + const onData = (d: Buffer) => { + parseBuf = Buffer.concat([parseBuf, d]); + const result = localReadFrames(parseBuf); + allFrames.push(...result.frames); + parseBuf = result.remainder; + }; + + if (!bridge.stdout) { throw new Error("bridge.stdout is null"); } + bridge.stdout.on("data", onData); + + // wait for Ready + while (!allFrames.some(f => f.type === FrameType.Ready)) { + await new Promise(r => setTimeout(r, 10)); + } + + const client = net.createConnection(SOCK_PATH); + await new Promise(r => client.on("connect", r)); + + // wait for Open + while (!allFrames.some(f => f.type === FrameType.Open)) { + await new Promise(r => setTimeout(r, 10)); + } + + const openFrame = allFrames.find(f => f.type === FrameType.Open); + if (!openFrame) { throw new Error("no Open frame"); } + const chan = openFrame.chan; + + // flood data from host->bridge->client socket to fill the socket buffer + // bridge will queue sock.write() calls and when client is destroyed, + // pending writes fail (EPIPE) = error event = closeClient called + if (!bridge.stdin) { throw new Error("bridge.stdin is null"); } + const payload = Buffer.alloc(65536, 0x42); + for (let i = 0; i < 100; i++) { + localWriteFrame(bridge.stdin, FrameType.Data, chan, payload); + } + + // don't read on client side — let bridge-side socket buffer fill + await new Promise(r => setTimeout(r, 100)); + + // half-close first -> bridge gets 'end' -> closeClient -> Close #1 + client.end(); + + // wait for bridge to process the FIN + await new Promise(r => setTimeout(r, 50)); + + // full-close -> pending writes fail -> ECONNRESET -> 'error' -> closeClient -> Close #2 + client.destroy(); + + // collect frames for 1s + await new Promise(r => setTimeout(r, 1000)); + bridge.stdout.removeListener("data", onData); + + // potential bug: both 'end' and 'error' fire on bridge-side socket = two Close frames + const closeFrames = allFrames.filter(f => f.type === FrameType.Close && f.chan === chan); + expect(closeFrames).toHaveLength(1); + }); + + test("bridge destroys server-side socket on client disconnect", async () => { + bridge = spawnBridge({ disableNodeSafety: true }); + await waitForFrames(bridge, 1); // Ready + + const client = net.createConnection(SOCK_PATH); + await new Promise(r => client.on("connect", r)); + await waitForFrames(bridge, 1); // Open + + // register listener before triggering events to avoid race + const clientClosed = new Promise((r) => { + const timer = setTimeout(() => { r(false); }, 1000); + client.on("close", () => { + clearTimeout(timer); + r(true); + }); + }); + + // client half-closes (sends FIN to bridge) + client.end(); + + // bridge receives 'end' -> closeClient -> sends Close frame, deletes from map + // potential bug: closeClient does not call sock.destroy(), so bridge-side socket + // stays open (writable). Client never receives FIN from bridge. + await waitForFrames(bridge, 1); // Close + + // if bridge properly destroyed its socket, client receives close event + expect(await clientClosed).toBe(true); + }); +}); diff --git a/src/remote/resolver.ts b/src/remote/resolver.ts index 15a37a4..df35ae7 100644 --- a/src/remote/resolver.ts +++ b/src/remote/resolver.ts @@ -12,6 +12,7 @@ import { DEVCONTAINER_SERVER_LISTEN_PORT } from "../common/constants"; import { setExecCtx, getExecCtx } from "../common/ctx/ctx"; import { RemoteExecCtx } from "../common/ctx/remoteCtx"; import { SSHDestination } from "../extension/ssh"; +import { SshAgentRelay } from "./exec-pipe/local"; import * as utils from "../common/utils"; export const AUTHORITY_BASE: string = "devcontainer-remote"; @@ -85,13 +86,14 @@ export class DevContainerResolver implements vscode.RemoteAuthorityResolver, vsc private serverHostPort: number | undefined; private statusItemFormatter: vscode.Disposable | undefined; + private sshAgentRelay: SshAgentRelay | undefined; constructor(context: vscode.ExtensionContext) { this.extensionCtx = context; void this.extensionCtx; // TODO } - resolve(_authority: string, context: vscode.RemoteAuthorityResolverContext): Thenable { + async resolve(_authority: string, context: vscode.RemoteAuthorityResolverContext): Promise { const fullAuthority = vscode.env.remoteAuthority!; // eslint-disable-line @typescript-eslint/no-non-null-assertion if (context.execServer) { @@ -205,6 +207,10 @@ export class DevContainerResolver implements vscode.RemoteAuthorityResolver, vsc ...parsedConfig.customizations?.vscode?.extensions ?? [], ]); this.serverHostPort = port; + + this.sshAgentRelay?.dispose(); + this.sshAgentRelay = await this.containerState.startSshAgentRelay(); + progress.report({ message: "Connecting...", increment: 40 }); const ctkn = await this.containerState.getConnectionToken(); @@ -214,14 +220,29 @@ export class DevContainerResolver implements vscode.RemoteAuthorityResolver, vsc vscode.commands.executeCommand("setContext", "forwardedPortsFeaturesEnabled", true); // if exec server, host and port are on the remote machine. forward it out to the local machine - if (context.execServer) { - return new vscode.ManagedResolvedAuthority(() => { - return connectToRemote(context.execServer!, host, port); // eslint-disable-line @typescript-eslint/no-non-null-assertion - }, ctkn); + const authority = (() => { + if (context.execServer) { + return new vscode.ManagedResolvedAuthority(() => { + return connectToRemote(context.execServer!, host, port); // eslint-disable-line @typescript-eslint/no-non-null-assertion + }, ctkn); + } + else { + return new vscode.ResolvedAuthority(host, port, ctkn); + } + })(); + + if (this.sshAgentRelay) { + const sockPath = this.sshAgentRelay.containerSockPath; + getLogSink().info(`Injecting SSH_AUTH_SOCK=${sockPath} into resolver env`); + return Object.assign(authority, { + extensionHostEnv: { SSH_AUTH_SOCK: sockPath }, + } satisfies vscode.ResolvedOptions); } else { - return new vscode.ResolvedAuthority(host, port, ctkn); + getLogSink().info("Could not create SshAgentRelay, not injecting env SSH_AUTH_SOCK into resolver env"); } + + return authority; } // getCanonicalURI?(uri: vscode.Uri): vscode.ProviderResult { @@ -242,6 +263,7 @@ export class DevContainerResolver implements vscode.RemoteAuthorityResolver, vsc public async dispose() { this.statusItemFormatter?.dispose(); + this.sshAgentRelay?.dispose(); await this.containerState?.dispose(); } }; diff --git a/src/tests/lint.test.ts b/src/tests/lint.test.ts new file mode 100644 index 0000000..42b6466 --- /dev/null +++ b/src/tests/lint.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, test } from "vitest"; +import { readFileSync, readdirSync, statSync } from "node:fs"; +import { resolve } from "node:path"; + +const SRC_DIR = resolve(__dirname, ".."); +const OBJECT_DEFAULT_PATTERN = /\w+:\s*(\{[^}]+\}|[\w<>.\[\]]+)\s*=\s*\{[^}]+\}/; + +function getSourceFiles(dir: string = SRC_DIR): string[] { + const entries: string[] = []; + for (const entry of readdirSync(dir)) { + const full = resolve(dir, entry); + if (statSync(full).isDirectory() && entry !== "node_modules") { + entries.push(...getSourceFiles(full)); + } + else if (entry.endsWith(".ts") && !entry.endsWith(".test.ts") && !entry.endsWith(".d.ts")) { + entries.push(full); + } + } + return entries; +} + +describe("lint", () => { + test("no object literal defaults in function parameters", () => { + const violations: string[] = []; + + for (const file of getSourceFiles()) { + const lines = readFileSync(file, "utf-8").split("\n"); + for (let i = 0; i < lines.length; i++) { + if (OBJECT_DEFAULT_PATTERN.test(lines[i])) { + const rel = file.replace(SRC_DIR + "/", ""); + violations.push(`${rel}:${i + 1}: ${lines[i].trim()}`); + } + } + } + + expect(violations, [ + "Use destructured defaults instead of object literal defaults:", + " Bad: opts: T = { key: val }", + " Good: { key = val }: T = {}", + "", + ...violations, + ].join("\n")).toHaveLength(0); + }); +});