fix(agent): bound CLI subprocess execution

2026-08-18 17:00:23 +08:00
parent 9aa5a96e60
commit 18e8b25f48
7 changed files with 384 additions and 65 deletions
@@ -14,7 +14,7 @@ jobs:
      dockerfile: Dockerfile
      build_context: .
      cache_image: gitea.waternetwork.cn/orgtjwater/tjwateragent:ci-cache
-      test_target: build
+      test_target: test
      deploy_service: agent
      deploy_host: 192.168.1.114
    secrets:
@@ -65,6 +65,15 @@ COPY cli ./cli
 COPY .opencode ./.opencode
 RUN bun run check

+FROM build AS test
+RUN apt-get update && apt-get install -y --no-install-recommends nodejs && \
+    rm -rf /var/lib/apt/lists/*
+COPY contracts ./contracts
+COPY node-tests ./node-tests
+COPY scripts ./scripts
+COPY tests ./tests
+RUN bun run test:ci
+
 FROM build AS runner
 WORKDIR /app

@@ -8,7 +8,9 @@
    "install:opencode": "bun install --cwd .opencode",
    "typecheck": "tsc --noEmit -p tsconfig.json",
    "typecheck:opencode": "bun run --cwd .opencode typecheck",
+    "test": "bun test tests",
    "test:cli": "node --test node-tests/cli/*.node.mjs",
+    "test:ci": "bun run contract:check && /usr/bin/node --test node-tests/cli/*.node.mjs && bun test tests",
    "dev": "bun --watch src/server.ts",
    "build": "bun run check",
    "check": "bun run typecheck && bun run typecheck:opencode",
@@ -0,0 +1,196 @@
+import { spawn } from "node:child_process";
+
+import { type RuntimeSessionContext } from "../runtime/sessionContext.js";
+
+type OutputStream = "stdout" | "stderr";
+
+export type CliExecutionResult = {
+  outcome: "completed" | "timeout" | "output_limit";
+  exitCode: number | null;
+  signal: NodeJS.Signals | null;
+  status: number;
+  stderr: string;
+  stdout: string;
+  exceededStream?: OutputStream;
+};
+
+type ExecuteCliCommandOptions = {
+  apiBaseUrl: string;
+  cliPath: string;
+  maxOutputBytes: number;
+  terminationGraceMs?: number;
+};
+
+const getCompletedStatus = (exitCode: number | null, stdout: string) => {
+  let errorCode = "";
+  try {
+    const payload = JSON.parse(stdout) as { error?: { code?: unknown } };
+    errorCode =
+      typeof payload.error?.code === "string" ? payload.error.code : "";
+  } catch {
+    errorCode = "";
+  }
+
+  if (errorCode === "HTTP_401" || errorCode === "UNAUTHENTICATED") {
+    return 401;
+  }
+  if (errorCode === "HTTP_403") {
+    return 403;
+  }
+  return exitCode === 0 ? 200 : 502;
+};
+
+export const executeCliCommand = async (
+  context: RuntimeSessionContext,
+  command: string,
+  timeoutSec: number,
+  options: ExecuteCliCommandOptions,
+): Promise<CliExecutionResult> => {
+  const maxOutputBytes = options.maxOutputBytes;
+  if (!Number.isSafeInteger(maxOutputBytes) || maxOutputBytes <= 0) {
+    throw new Error("maxOutputBytes must be a positive safe integer");
+  }
+
+  const child = spawn(
+    options.cliPath,
+    ["--auth-stdin", ...command.split(/\s+/).filter(Boolean)],
+    { stdio: ["pipe", "pipe", "pipe"] },
+  );
+  const stdoutChunks: Buffer[] = [];
+  const stderrChunks: Buffer[] = [];
+  let stdoutBytes = 0;
+  let stderrBytes = 0;
+  let terminationReason:
+    | "timeout"
+    | "output_limit"
+    | "execution_error"
+    | null = null;
+  let exceededStream: OutputStream | undefined;
+  let terminationStarted = false;
+  let settled = false;
+  let executionError: Error | null = null;
+  let forceKillTimer: ReturnType<typeof setTimeout> | undefined;
+
+  const result = await new Promise<CliExecutionResult>((resolve, reject) => {
+    const cleanup = () => {
+      clearTimeout(timeoutTimer);
+      if (forceKillTimer) {
+        clearTimeout(forceKillTimer);
+      }
+    };
+
+    const terminate = (
+      reason: "timeout" | "output_limit" | "execution_error",
+    ) => {
+      if (terminationStarted) {
+        return;
+      }
+      terminationStarted = true;
+      terminationReason = reason;
+
+      if (child.exitCode === null && child.signalCode === null) {
+        child.kill("SIGTERM");
+      }
+      forceKillTimer = setTimeout(() => {
+        if (child.exitCode === null && child.signalCode === null) {
+          child.kill("SIGKILL");
+        }
+      }, options.terminationGraceMs ?? 1500);
+    };
+
+    const capture = (stream: OutputStream, data: Buffer) => {
+      if (terminationReason) {
+        return;
+      }
+      const chunks = stream === "stdout" ? stdoutChunks : stderrChunks;
+      const bytes = stream === "stdout" ? stdoutBytes : stderrBytes;
+      if (bytes + data.length > maxOutputBytes) {
+        exceededStream = stream;
+        terminate("output_limit");
+        return;
+      }
+      chunks.push(data);
+      if (stream === "stdout") {
+        stdoutBytes += data.length;
+      } else {
+        stderrBytes += data.length;
+      }
+    };
+
+    const timeoutTimer = setTimeout(() => {
+      if (child.exitCode === null && child.signalCode === null) {
+        terminate("timeout");
+      }
+    }, timeoutSec * 1000);
+
+    child.stdout.on("data", (data: Buffer) => capture("stdout", data));
+    child.stderr.on("data", (data: Buffer) => capture("stderr", data));
+    child.stdin.on("error", (error) => {
+      if (terminationReason === null) {
+        executionError = error;
+        terminate("execution_error");
+      }
+    });
+    child.on("error", (error) => {
+      if (terminationReason === null) {
+        executionError = error;
+        terminate("execution_error");
+      }
+    });
+    child.on("close", (exitCode, signal) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      cleanup();
+      if (terminationReason === "timeout") {
+        resolve({
+          outcome: "timeout",
+          exitCode,
+          signal,
+          status: 504,
+          stderr: "",
+          stdout: "",
+        });
+        return;
+      }
+      if (executionError) {
+        reject(executionError);
+        return;
+      }
+      if (terminationReason === "output_limit") {
+        resolve({
+          outcome: "output_limit",
+          exceededStream,
+          exitCode,
+          signal,
+          status: 502,
+          stderr: "",
+          stdout: "",
+        });
+        return;
+      }
+
+      const stdout = Buffer.concat(stdoutChunks, stdoutBytes).toString("utf-8");
+      const stderr = Buffer.concat(stderrChunks, stderrBytes).toString("utf-8");
+      resolve({
+        outcome: "completed",
+        exitCode,
+        signal,
+        status: getCompletedStatus(exitCode, stdout),
+        stderr,
+        stdout,
+      });
+    });
+
+    child.stdin.end(
+      JSON.stringify({
+        server: options.apiBaseUrl,
+        access_token: context.accessToken,
+        project_id: context.projectId,
+      }),
+    );
+  });
+
+  return result;
+};
@@ -1,5 +1,4 @@
 import { randomUUID } from "node:crypto";
-import { spawn } from "node:child_process";
 import cors from "cors";
 import express from "express";

@@ -11,6 +10,7 @@ import {
  runWithCredentialRefresh,
 } from "./auth/credentialRefresh.js";
 import { SessionTranscriptStore } from "./sessions/transcriptStore.js";
+import { executeCliCommand } from "./cli/executeCliCommand.js";
 import { ChatSessionBridge } from "./chat/sessionBridge.js";
 import { config } from "./config.js";
 import { SessionUiStateStore } from "./sessions/uiStateStore.js";
@@ -229,7 +229,12 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
    result = await runWithCredentialRefresh(
      credentialRefreshCoordinator,
      context,
-      (activeContext) => executeCliCommand(activeContext, command, timeoutSec),
+      (activeContext) =>
+        executeCliCommand(activeContext, command, timeoutSec, {
+          apiBaseUrl: config.TJWATER_API_BASE_URL,
+          cliPath: config.TJWATER_CLI_PATH,
+          maxOutputBytes: config.MAX_INLINE_RESULT_BYTES,
+        }),
    );
  } catch (error) {
    if (!(error instanceof CredentialRefreshError)) {
@@ -266,6 +271,20 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
    return;
  }

+  if (result.outcome === "output_limit") {
+    res.status(502).json({
+      ok: false,
+      schema_version: "tjwater-cli/v1",
+      summary: "CLI 输出超过安全限制",
+      error: {
+        code: "OUTPUT_LIMIT_EXCEEDED",
+        message: `${result.exceededStream ?? "output"} exceeded ${config.MAX_INLINE_RESULT_BYTES} bytes`,
+        retryable: false,
+      },
+    });
+    return;
+  }
+
  if (result.status === 401) {
    markAuthExpired(
      getRuntimeSessionContext(sessionId) ?? context,
@@ -300,68 +319,6 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
  }
 });

-const executeCliCommand = async (
-  context: RuntimeSessionContext,
-  command: string,
-  timeoutSec: number,
-) => {
-  const child = spawn(
-    config.TJWATER_CLI_PATH,
-    ["--auth-stdin", ...command.split(/\s+/).filter(Boolean)],
-    { stdio: ["pipe", "pipe", "pipe"] },
-  );
-  let stdout = "";
-  let stderr = "";
-  child.stdout.on("data", (data: Buffer) => {
-    stdout += data.toString("utf-8");
-  });
-  child.stderr.on("data", (data: Buffer) => {
-    stderr += data.toString("utf-8");
-  });
-  child.stdin.write(
-    JSON.stringify({
-      server: config.TJWATER_API_BASE_URL,
-      access_token: context.accessToken,
-      project_id: context.projectId,
-    }),
-  );
-  child.stdin.end();
-
-  const exitCode = await new Promise<number | null>((resolve, reject) => {
-    const timer = setTimeout(() => {
-      child.kill("SIGTERM");
-      resolve(-1);
-    }, timeoutSec * 1000);
-    child.on("close", (code) => {
-      clearTimeout(timer);
-      resolve(code);
-    });
-    child.on("error", (error) => {
-      clearTimeout(timer);
-      reject(error);
-    });
-  });
-  let errorCode = "";
-  try {
-    const payload = JSON.parse(stdout) as { error?: { code?: unknown } };
-    errorCode =
-      typeof payload.error?.code === "string" ? payload.error.code : "";
-  } catch {
-    errorCode = "";
-  }
-  const status =
-    exitCode === -1
-      ? 504
-      : errorCode === "HTTP_401" || errorCode === "UNAUTHENTICATED"
-        ? 401
-        : errorCode === "HTTP_403"
-          ? 403
-          : exitCode === 0
-            ? 200
-            : 502;
-  return { exitCode, status, stderr, stdout };
-};
-
 app.post("/internal/tools/store-render-ref", async (req, res) => {
  if (req.header("x-agent-internal-token") !== internalToken) {
    res.status(403).json({ message: "forbidden" });
@@ -0,0 +1,122 @@
+import { describe, expect, test } from "bun:test";
+import { fileURLToPath } from "node:url";
+
+import { executeCliCommand } from "../../src/cli/executeCliCommand.js";
+import { type RuntimeSessionContext } from "../../src/runtime/sessionContext.js";
+
+const cliPath = fileURLToPath(
+  new URL("../fixtures/fakeCli.mjs", import.meta.url),
+);
+
+const context: RuntimeSessionContext = {
+  accessToken: "test-token",
+  actorKey: "actor-1",
+  clientSessionId: "client-1",
+  projectId: "project-1",
+  projectKey: "project-1",
+  sessionId: "session-1",
+  traceId: "trace-1",
+};
+
+const run = (
+  command: string,
+  options: {
+    maxOutputBytes?: number;
+    terminationGraceMs?: number;
+    timeoutSec?: number;
+  } = {},
+) =>
+  executeCliCommand(context, command, options.timeoutSec ?? 1, {
+    apiBaseUrl: "http://127.0.0.1:8000",
+    cliPath,
+    maxOutputBytes: options.maxOutputBytes ?? 64,
+    terminationGraceMs: options.terminationGraceMs ?? 20,
+  });
+
+describe("executeCliCommand", () => {
+  test("accepts output at the byte limit", async () => {
+    await expect(run("stdout 123456", { maxOutputBytes: 6 })).resolves.toMatchObject({
+      outcome: "completed",
+      exitCode: 0,
+      status: 200,
+      stdout: "123456",
+    });
+  });
+
+  test("rejects multibyte output above the byte limit without returning a partial body", async () => {
+    await expect(run("stdout 水水", { maxOutputBytes: 5 })).resolves.toMatchObject({
+      outcome: "output_limit",
+      exceededStream: "stdout",
+      status: 502,
+      stderr: "",
+      stdout: "",
+    });
+  });
+
+  test("limits stderr independently", async () => {
+    await expect(run("stderr 1234567", { maxOutputBytes: 6 })).resolves.toMatchObject({
+      outcome: "output_limit",
+      exceededStream: "stderr",
+      status: 502,
+      stderr: "",
+      stdout: "",
+    });
+  });
+
+  test("waits for a SIGTERM-aware process to close after timeout", async () => {
+    const startedAt = Date.now();
+    const result = await run("term", {
+      terminationGraceMs: 100,
+      timeoutSec: 0.25,
+    });
+
+    expect(result).toMatchObject({ outcome: "timeout", status: 504 });
+    expect(Date.now() - startedAt).toBeGreaterThanOrEqual(270);
+  });
+
+  test("uses SIGKILL when a timed-out process ignores SIGTERM", async () => {
+    const result = await run("ignore-term", { timeoutSec: 0.25 });
+
+    expect(result).toMatchObject({
+      outcome: "timeout",
+      signal: "SIGKILL",
+      status: 504,
+    });
+  });
+
+  test("keeps the timeout outcome when closing stdin also errors", async () => {
+    const largeContext = {
+      ...context,
+      accessToken: "x".repeat(1024 * 1024),
+    };
+
+    await expect(
+      executeCliCommand(largeContext, "ignore-term", 0.25, {
+        apiBaseUrl: "http://127.0.0.1:8000",
+        cliPath,
+        maxOutputBytes: 64,
+        terminationGraceMs: 20,
+      }),
+    ).resolves.toMatchObject({
+      outcome: "timeout",
+      signal: "SIGKILL",
+      status: 504,
+    });
+  });
+
+  test("rejects a deterministic stdin pipe error without crashing", async () => {
+    const largeContext = {
+      ...context,
+      accessToken: "x".repeat(1024 * 1024),
+    };
+
+    await expect(
+      executeCliCommand(largeContext, "closed-stdin", 1, {
+        apiBaseUrl: "http://127.0.0.1:8000",
+        cliPath,
+        maxOutputBytes: 64,
+        terminationGraceMs: 20,
+      }),
+    ).rejects.toBeInstanceOf(Error);
+  });
+});
@@ -0,0 +1,33 @@
+#!/usr/bin/env node
+
+import { closeSync } from "node:fs";
+
+const command = process.argv[3];
+const value = process.argv[4] ?? "";
+
+if (command === "stdout") {
+  process.stdout.write(value);
+  process.exit(0);
+}
+
+if (command === "stderr") {
+  process.stderr.write(value);
+  process.exit(1);
+}
+
+if (command === "term") {
+  process.on("SIGTERM", () => {
+    setTimeout(() => process.exit(0), 30);
+  });
+  setInterval(() => undefined, 1000);
+}
+
+if (command === "ignore-term") {
+  process.on("SIGTERM", () => undefined);
+  setInterval(() => undefined, 1000);
+}
+
+if (command === "closed-stdin") {
+  closeSync(0);
+  setInterval(() => undefined, 1000);
+}