fix(pi-workflow): bound host command execution
This commit is contained in:
@@ -16,6 +16,7 @@ import { promisify } from "node:util";
|
||||
import { deflateSync } from "node:zlib";
|
||||
import {
|
||||
ApprovedHostCommandExecutor,
|
||||
ExecFileCommandRunner,
|
||||
FilesystemPrototypeValidator,
|
||||
GitStateProbe,
|
||||
GitWorkspaceManager,
|
||||
@@ -283,6 +284,27 @@ test("Git probe and approved host commands enforce identity and ownership", asyn
|
||||
'process.exit(process.env.PI_OFFLINE === "1" ? 0 : 1)',
|
||||
"utf8",
|
||||
);
|
||||
await writeFile(
|
||||
path.join(directory, "spawn-hanging-child.cjs"),
|
||||
[
|
||||
'const { spawn } = require("node:child_process");',
|
||||
'const { writeFileSync } = require("node:fs");',
|
||||
'const child = spawn(process.execPath, ["delayed-write.cjs"], { stdio: "ignore" });',
|
||||
'writeFileSync("src/timeout-pids.json", JSON.stringify([process.pid, child.pid]));',
|
||||
"setInterval(() => {}, 1_000);",
|
||||
].join("\n"),
|
||||
"utf8",
|
||||
);
|
||||
await writeFile(
|
||||
path.join(directory, "delayed-write.cjs"),
|
||||
[
|
||||
'const { writeFileSync } = require("node:fs");',
|
||||
'process.on("SIGTERM", () => {});',
|
||||
'setTimeout(() => writeFileSync("src/delayed.txt", "late\\n"), 1_200);',
|
||||
"setInterval(() => {}, 1_000);",
|
||||
].join("\n"),
|
||||
"utf8",
|
||||
);
|
||||
await execFileAsync("git", ["-C", directory, "add", "."]);
|
||||
await execFileAsync("git", ["-C", directory, "commit", "-qm", "initial"]);
|
||||
const probe = new GitStateProbe();
|
||||
@@ -315,6 +337,35 @@ test("Git probe and approved host commands enforce identity and ownership", asyn
|
||||
args: ["PI_OFFLINE=1", process.execPath, "check-offline.cjs"],
|
||||
});
|
||||
assert.equal(offline.result.code, 0);
|
||||
const boundedExecutor = new ApprovedHostCommandExecutor(
|
||||
new ExecFileCommandRunner(),
|
||||
undefined,
|
||||
250,
|
||||
);
|
||||
const startedAt = Date.now();
|
||||
const timedOut = await boundedExecutor.run({
|
||||
worktreeRoot: directory,
|
||||
ownedPaths: ["src"],
|
||||
executable: process.execPath,
|
||||
args: ["spawn-hanging-child.cjs"],
|
||||
});
|
||||
assert.equal(timedOut.result.code, 124);
|
||||
assert.match(timedOut.result.stderr, /timed out after 250ms/);
|
||||
assert.ok(Date.now() - startedAt < 2_500);
|
||||
const timedOutProcessIds = JSON.parse(
|
||||
await readFile(path.join(directory, "src", "timeout-pids.json"), "utf8"),
|
||||
) as number[];
|
||||
for (const processId of timedOutProcessIds) {
|
||||
assert.throws(
|
||||
() => process.kill(processId, 0),
|
||||
(error: unknown) => (error as NodeJS.ErrnoException).code === "ESRCH",
|
||||
);
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 1_000));
|
||||
await assert.rejects(
|
||||
access(path.join(directory, "src", "delayed.txt")),
|
||||
/ENOENT/,
|
||||
);
|
||||
await assert.rejects(
|
||||
executor.run({
|
||||
worktreeRoot: directory,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { execFile, spawn } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import { constants } from "node:fs";
|
||||
import {
|
||||
@@ -25,6 +25,7 @@ export interface CommandResult {
|
||||
|
||||
export interface CommandRunOptions {
|
||||
env?: NodeJS.ProcessEnv;
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
export interface CommandRunner {
|
||||
@@ -504,16 +505,21 @@ const PACKAGE_EXEC_SUBCOMMANDS = new Map<string, Set<string>>([
|
||||
["bun", new Set(["x"])],
|
||||
]);
|
||||
|
||||
const DEFAULT_HOST_COMMAND_TIMEOUT_MS = 3 * 60 * 1_000;
|
||||
|
||||
export class ApprovedHostCommandExecutor {
|
||||
private readonly runner: CommandRunner;
|
||||
private readonly probe: GitStateProbe;
|
||||
private readonly timeoutMs: number;
|
||||
|
||||
constructor(
|
||||
runner: CommandRunner = new ExecFileCommandRunner(),
|
||||
probe: GitStateProbe = new GitStateProbe(runner),
|
||||
timeoutMs = DEFAULT_HOST_COMMAND_TIMEOUT_MS,
|
||||
) {
|
||||
this.runner = runner;
|
||||
this.probe = probe;
|
||||
this.timeoutMs = timeoutMs;
|
||||
}
|
||||
|
||||
async run(options: {
|
||||
@@ -570,6 +576,7 @@ export class ApprovedHostCommandExecutor {
|
||||
const before = beforeRepository.get(root) ?? (await this.probe.capture(root));
|
||||
const result = await this.runner.run(root, executable, args, {
|
||||
env: commandEnv,
|
||||
timeoutMs: this.timeoutMs,
|
||||
});
|
||||
try {
|
||||
const afterRepository = await this.probe.captureRepository(root);
|
||||
@@ -813,6 +820,50 @@ export class WorkspaceValidationError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
const PROCESS_TERMINATION_GRACE_MS = 100;
|
||||
const PROCESS_TERMINATION_POLL_MS = 10;
|
||||
const PROCESS_TERMINATION_TIMEOUT_MS = 1_000;
|
||||
|
||||
function wait(milliseconds: number) {
|
||||
return new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
|
||||
}
|
||||
|
||||
function processGroupExists(processId: number) {
|
||||
try {
|
||||
process.kill(-processId, 0);
|
||||
return true;
|
||||
} catch (error) {
|
||||
return (error as NodeJS.ErrnoException).code !== "ESRCH";
|
||||
}
|
||||
}
|
||||
|
||||
function signalProcessGroup(processId: number, signal: NodeJS.Signals) {
|
||||
try {
|
||||
process.kill(-processId, signal);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function terminateProcessTree(processId: number) {
|
||||
if (process.platform === "win32") {
|
||||
await execFileAsync("taskkill", ["/pid", String(processId), "/T", "/F"]);
|
||||
return;
|
||||
}
|
||||
signalProcessGroup(processId, "SIGTERM");
|
||||
await wait(PROCESS_TERMINATION_GRACE_MS);
|
||||
signalProcessGroup(processId, "SIGKILL");
|
||||
const deadline = Date.now() + PROCESS_TERMINATION_TIMEOUT_MS;
|
||||
while (processGroupExists(processId)) {
|
||||
if (Date.now() >= deadline) {
|
||||
throw new Error(`process group ${processId} remained alive after SIGKILL`);
|
||||
}
|
||||
await wait(PROCESS_TERMINATION_POLL_MS);
|
||||
}
|
||||
}
|
||||
|
||||
const COMMAND_OUTPUT_LIMIT_BYTES = 4 * 1024 * 1024;
|
||||
|
||||
export class ExecFileCommandRunner implements CommandRunner {
|
||||
async run(
|
||||
cwd: string,
|
||||
@@ -820,26 +871,83 @@ export class ExecFileCommandRunner implements CommandRunner {
|
||||
args: string[],
|
||||
options: CommandRunOptions = {},
|
||||
) {
|
||||
try {
|
||||
const result = await execFileAsync(executable, args, {
|
||||
return new Promise<CommandResult>((resolve) => {
|
||||
const stdoutChunks: Buffer[] = [];
|
||||
const stderrChunks: Buffer[] = [];
|
||||
let stdoutBytes = 0;
|
||||
let stderrBytes = 0;
|
||||
let spawnError: Error | undefined;
|
||||
let timedOut = false;
|
||||
let outputExceeded = false;
|
||||
let termination: Promise<void> | undefined;
|
||||
let timeout: NodeJS.Timeout | undefined;
|
||||
const child = spawn(executable, args, {
|
||||
cwd,
|
||||
encoding: "utf8",
|
||||
maxBuffer: 4 * 1024 * 1024,
|
||||
detached: process.platform !== "win32",
|
||||
env: options.env,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
return { code: 0, stdout: result.stdout, stderr: result.stderr };
|
||||
} catch (error) {
|
||||
const failed = error as Error & {
|
||||
code?: number | string;
|
||||
stdout?: string;
|
||||
stderr?: string;
|
||||
const terminate = () => {
|
||||
if (!termination && child.pid !== undefined) {
|
||||
termination = terminateProcessTree(child.pid);
|
||||
}
|
||||
};
|
||||
return {
|
||||
code: typeof failed.code === "number" ? failed.code : 1,
|
||||
stdout: failed.stdout ?? "",
|
||||
stderr: failed.stderr ?? failed.message,
|
||||
const capture = (chunks: Buffer[], chunk: Buffer, stream: "stdout" | "stderr") => {
|
||||
const capturedBytes = stream === "stdout" ? stdoutBytes : stderrBytes;
|
||||
const remaining = COMMAND_OUTPUT_LIMIT_BYTES - capturedBytes;
|
||||
if (remaining > 0) chunks.push(chunk.subarray(0, remaining));
|
||||
if (stream === "stdout") stdoutBytes += chunk.length;
|
||||
else stderrBytes += chunk.length;
|
||||
if (chunk.length > remaining && !outputExceeded) {
|
||||
outputExceeded = true;
|
||||
terminate();
|
||||
}
|
||||
};
|
||||
}
|
||||
child.stdout.on("data", (chunk: Buffer) => capture(stdoutChunks, chunk, "stdout"));
|
||||
child.stderr.on("data", (chunk: Buffer) => capture(stderrChunks, chunk, "stderr"));
|
||||
child.on("error", (error) => {
|
||||
spawnError = error;
|
||||
});
|
||||
child.on("close", async (code) => {
|
||||
if (timeout !== undefined) clearTimeout(timeout);
|
||||
let terminationError: string | undefined;
|
||||
try {
|
||||
await termination;
|
||||
} catch (failure) {
|
||||
terminationError = failure instanceof Error ? failure.message : String(failure);
|
||||
}
|
||||
const stdout = Buffer.concat(stdoutChunks).toString("utf8");
|
||||
const stderr = Buffer.concat(stderrChunks).toString("utf8");
|
||||
if (timedOut) {
|
||||
const detail = [stderr, terminationError].filter(Boolean).join("\n");
|
||||
resolve({
|
||||
code: terminationError ? 125 : 124,
|
||||
stdout,
|
||||
stderr: `Host command timed out after ${options.timeoutMs}ms${detail ? `\n${detail}` : ""}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (outputExceeded) {
|
||||
resolve({
|
||||
code: terminationError ? 125 : 1,
|
||||
stdout,
|
||||
stderr: `Host command output exceeded ${COMMAND_OUTPUT_LIMIT_BYTES} bytes${terminationError ? `\n${terminationError}` : ""}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
resolve({
|
||||
code: code ?? 1,
|
||||
stdout,
|
||||
stderr: stderr || spawnError?.message || "",
|
||||
});
|
||||
});
|
||||
if (options.timeoutMs !== undefined) {
|
||||
timeout = setTimeout(() => {
|
||||
timedOut = true;
|
||||
terminate();
|
||||
}, options.timeoutMs);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user