Skip to content

Commit d90c9a1

Browse files
cxbAsDevsteipete
cxbAsDev
and
steipete
authored
fix(agents): handle stdout/stderr stream errors in ssh sandbox commands (#101031)
* fix(agents): handle stdout/stderr stream errors in ssh sandbox commands * fix(agents): also handle stdin stream errors in ssh sandbox command * fix(agents): handle tar/ssh pipeline stream errors in sandbox upload * fix(agents): kill ssh child on stdout/stderr/stdin stream errors in sandbox * fix(agents): move upload fail helper before use to avoid TDZ in ssh sandbox * docs(proof): note ssh-sandbox upload TDZ fix in proof header * fix(agents): harden SSH sandbox stream errors --------- Co-authored-by: Peter Steinberger <[email protected]>
1 parent 62e5d44 commit d90c9a1

2 files changed

Lines changed: 194 additions & 26 deletions

File tree

‎src/agents/sandbox/ssh.stream-errors.test.ts‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
import { spawn, type ChildProcess } from "node:child_process";
2+
import { EventEmitter } from "node:events";
3+
import fs from "node:fs/promises";
4+
import os from "node:os";
5+
import path from "node:path";
6+
import { PassThrough } from "node:stream";
7+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
8+
9+
const spawnMock = vi.hoisted(() => vi.fn());
10+
11+
type MockChildProcess = EventEmitter & {
12+
stdin: PassThrough;
13+
stdout: PassThrough;
14+
stderr: PassThrough;
15+
kill: ReturnType<typeof vi.fn>;
16+
};
17+
18+
function createMockChildProcess(): MockChildProcess {
19+
const child = new EventEmitter() as MockChildProcess;
20+
child.stdin = new PassThrough();
21+
child.stdout = new PassThrough();
22+
child.stderr = new PassThrough();
23+
child.kill = vi.fn(() => true);
24+
return child;
25+
}
26+
27+
vi.mock("node:child_process", async () => {
28+
const actual = await vi.importActual<typeof import("node:child_process")>("node:child_process");
29+
return {
30+
...actual,
31+
spawn: spawnMock,
32+
};
33+
});
34+
35+
const spawnMocked = vi.mocked(spawn);
36+
const tempDirs: string[] = [];
37+
38+
let runSshSandboxCommand: typeof import("./ssh.js").runSshSandboxCommand;
39+
let uploadDirectoryToSshTarget: typeof import("./ssh.js").uploadDirectoryToSshTarget;
40+
41+
beforeEach(async () => {
42+
vi.resetModules();
43+
vi.clearAllMocks();
44+
({ runSshSandboxCommand, uploadDirectoryToSshTarget } = await import("./ssh.js"));
45+
});
46+
47+
afterEach(async () => {
48+
await Promise.all(
49+
tempDirs.splice(0).map(async (dir) => {
50+
await fs.rm(dir, { recursive: true, force: true });
51+
}),
52+
);
53+
});
54+
55+
function fakeSession(): import("./ssh.js").SshSandboxSession {
56+
return {
57+
command: "ssh",
58+
configPath: "/tmp/ssh-config",
59+
host: "host",
60+
};
61+
}
62+
63+
describe("SSH sandbox stream errors", () => {
64+
it.each(["stdout", "stderr", "stdin"] as const)(
65+
"rejects and terminates once when command %s fails",
66+
async (streamName) => {
67+
const child = createMockChildProcess();
68+
spawnMocked.mockReturnValueOnce(child as unknown as ChildProcess);
69+
const expected = `${streamName} failed`;
70+
const result = runSshSandboxCommand({
71+
session: fakeSession(),
72+
remoteCommand: "echo hi",
73+
});
74+
75+
child[streamName].emit("error", new Error(expected));
76+
77+
await expect(result).rejects.toThrow(expected);
78+
expect(child.kill).toHaveBeenCalledExactlyOnceWith("SIGKILL");
79+
80+
child.emit("close", 0);
81+
child[streamName].emit("error", new Error("late stream error"));
82+
expect(child.kill).toHaveBeenCalledOnce();
83+
},
84+
);
85+
86+
it.each(["tar.stdout", "tar.stderr", "ssh.stdin", "ssh.stdout", "ssh.stderr"] as const)(
87+
"rejects and terminates both upload children once when %s fails",
88+
async (stream) => {
89+
const localDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-ssh-stream-test-"));
90+
tempDirs.push(localDir);
91+
const tar = createMockChildProcess();
92+
const ssh = createMockChildProcess();
93+
spawnMocked
94+
.mockReturnValueOnce(tar as unknown as ChildProcess)
95+
.mockReturnValueOnce(ssh as unknown as ChildProcess);
96+
const expected = `${stream} failed`;
97+
const result = uploadDirectoryToSshTarget({
98+
session: fakeSession(),
99+
localDir,
100+
remoteDir: "/remote/workspace",
101+
});
102+
const rejection = expect(result).rejects.toThrow(expected);
103+
await vi.waitFor(() => expect(spawnMocked).toHaveBeenCalledTimes(2));
104+
const [childName, streamName] = stream.split(".") as ["tar" | "ssh", keyof MockChildProcess];
105+
const failedStream = { tar, ssh }[childName][streamName] as PassThrough;
106+
107+
failedStream.emit("error", new Error(expected));
108+
109+
await rejection;
110+
expect(tar.kill).toHaveBeenCalledExactlyOnceWith("SIGKILL");
111+
expect(ssh.kill).toHaveBeenCalledExactlyOnceWith("SIGKILL");
112+
113+
tar.emit("close", 0);
114+
ssh.emit("close", 0);
115+
failedStream.emit("error", new Error("late stream error"));
116+
expect(tar.kill).toHaveBeenCalledOnce();
117+
expect(ssh.kill).toHaveBeenCalledOnce();
118+
},
119+
);
120+
});

‎src/agents/sandbox/ssh.ts‎

Lines changed: 74 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -681,32 +681,58 @@ export async function runSshSandboxCommand(
681681
});
682682
const stdoutChunks: Buffer[] = [];
683683
const stderrChunks: Buffer[] = [];
684+
let settled = false;
685+
686+
// Child and stdio errors can race with close. Settle once so an unusable
687+
// transport is terminated exactly once and later events stay harmless.
688+
const finish = (complete: () => void, terminate = false) => {
689+
if (settled) {
690+
return;
691+
}
692+
settled = true;
693+
if (terminate) {
694+
try {
695+
child.kill("SIGKILL");
696+
} catch {
697+
// Preserve the stream error that made the transport unusable.
698+
}
699+
}
700+
complete();
701+
};
702+
const fail = (error: unknown, terminate = false) => {
703+
finish(() => reject(toErrorObject(error, "Non-Error rejection")), terminate);
704+
};
684705

685706
child.stdout.on("data", (chunk) => stdoutChunks.push(Buffer.from(chunk)));
707+
child.stdout.on("error", (error) => fail(error, true));
686708
child.stderr.on("data", (chunk) => stderrChunks.push(Buffer.from(chunk)));
687-
child.on("error", reject);
709+
child.stderr.on("error", (error) => fail(error, true));
710+
child.on("error", fail);
688711
child.on("close", (code) => {
689-
const stdout = Buffer.concat(stdoutChunks);
690-
const stderr = Buffer.concat(stderrChunks);
691-
const exitCode = code ?? 0;
692-
if (exitCode !== 0 && !params.allowFailure) {
693-
reject(
694-
Object.assign(new Error(buildSshFailureMessage(stderr.toString("utf8"), exitCode)), {
695-
code: exitCode,
696-
stdout,
697-
stderr,
698-
}),
699-
);
700-
return;
701-
}
702-
resolve({ stdout, stderr, code: exitCode });
712+
finish(() => {
713+
const stdout = Buffer.concat(stdoutChunks);
714+
const stderr = Buffer.concat(stderrChunks);
715+
const exitCode = code ?? 0;
716+
if (exitCode !== 0 && !params.allowFailure) {
717+
reject(
718+
Object.assign(new Error(buildSshFailureMessage(stderr.toString("utf8"), exitCode)), {
719+
code: exitCode,
720+
stdout,
721+
stderr,
722+
}),
723+
);
724+
return;
725+
}
726+
resolve({ stdout, stderr, code: exitCode });
727+
});
703728
});
704729

705-
if (params.stdin !== undefined) {
730+
child.stdin?.on("error", (error) => fail(error, true));
731+
try {
706732
child.stdin.end(params.stdin);
707-
return;
733+
} catch (error) {
734+
fail(error, true);
708735
}
709-
child.stdin.end();
710736
});
711737
}
712738

@@ -790,20 +816,34 @@ export async function uploadDirectoryToSshTarget(params: {
790816
let sshClosed = false;
791817
let tarCode = 0;
792818
let sshCode = 0;
793-
794-
tar.stderr.on("data", (chunk) => tarStderr.push(Buffer.from(chunk)));
795-
ssh.stdout.on("data", (chunk) => sshStdout.push(Buffer.from(chunk)));
796-
ssh.stderr.on("data", (chunk) => sshStderr.push(Buffer.from(chunk)));
819+
let settled = false;
797820

798821
const fail = (error: unknown) => {
799-
tar.kill("SIGKILL");
800-
ssh.kill("SIGKILL");
822+
if (settled) {
823+
return;
824+
}
825+
settled = true;
826+
for (const child of [tar, ssh]) {
827+
try {
828+
child.kill("SIGKILL");
829+
} catch {
830+
// Preserve the pipeline error while still terminating the peer.
831+
}
832+
}
801833
reject(toErrorObject(error, "Non-Error rejection"));
802834
};
803835

836+
tar.stderr.on("data", (chunk) => tarStderr.push(Buffer.from(chunk)));
837+
tar.stderr.on("error", fail);
838+
tar.stdout.on("error", fail);
839+
ssh.stdout.on("data", (chunk) => sshStdout.push(Buffer.from(chunk)));
840+
ssh.stdout.on("error", fail);
841+
ssh.stderr.on("data", (chunk) => sshStderr.push(Buffer.from(chunk)));
842+
ssh.stderr.on("error", fail);
843+
ssh.stdin?.on("error", fail);
844+
804845
tar.on("error", fail);
805846
ssh.on("error", fail);
806-
tar.stdout.pipe(ssh.stdin);
807847

808848
tar.on("close", (code) => {
809849
tarClosed = true;
@@ -817,9 +857,10 @@ export async function uploadDirectoryToSshTarget(params: {
817857
});
818858

819859
function maybeResolve() {
820-
if (!tarClosed || !sshClosed) {
860+
if (settled || !tarClosed || !sshClosed) {
821861
return;
822862
}
863+
settled = true;
823864
if (tarCode !== 0) {
824865
reject(
825866
new Error(
@@ -838,6 +879,13 @@ export async function uploadDirectoryToSshTarget(params: {
838879
}
839880
resolve();
840881
}
882+
883+
try {
884+
// Readable pipe errors do not close the writable peer automatically.
885+
tar.stdout.pipe(ssh.stdin);
886+
} catch (error) {
887+
fail(error);
888+
}
841889
});
842890
}
843891

0 commit comments

Comments
 (0)