Merge pull request #5 from lhfer/claude/busy-noether-Rjbaz

security: harden credential handling, pipeline JS execution, and the HTTP/stream layer

packages/cli/src/commands/config/set.ts

@@ -1,6 +1,7 @@
 import {
   defineCommand,
   detectOutputFormat,
+  maskToken,
   readConfigFile,
   writeConfigFile,
   BailianError,
@@ -28,6 +29,11 @@ const VALID_KEYS = [
   "workspace_id",
 ];
 
+// Keys whose values are secrets. Their stored value must never be echoed back in
+// cleartext (CI logs, pipes, shared terminals); show a masked form instead — the
+// same policy `config show` and `auth status` already follow.
+const SECRET_KEYS = new Set(["api_key", "access_token", "access_key_id", "access_key_secret"]);
+
 // Allow hyphen-style keys (e.g. default-text-model → default_text_model)
 const KEY_ALIASES: Record<string, string> = {
   "base-url": "base_url",
@@ -120,7 +126,10 @@ export default defineCommand({
     await writeConfigFile(existing);
 
     if (!config.quiet) {
-      emitResult({ [resolvedKey]: existing[resolvedKey] }, format);
+      const shown = SECRET_KEYS.has(resolvedKey)
+        ? maskToken(String(existing[resolvedKey]))
+        : existing[resolvedKey];
+      emitResult({ [resolvedKey]: shown }, format);
     }
   },
 });

packages/cli/src/commands/knowledge/retrieve.ts

@@ -2,6 +2,7 @@ import {
   defineCommand,
   signRequest,
   detectOutputFormat,
+  maskToken,
   type Config,
   type GlobalFlags,
   type KnowledgeRetrieveRequest,
@@ -105,7 +106,7 @@ export default defineCommand({
 
     if (config.verbose) {
       process.stderr.write(`> POST ${url}\n`);
-      process.stderr.write(`> AK: ${accessKeyId.slice(0, 8)}...\n`);
+      process.stderr.write(`> AK: ${maskToken(accessKeyId)}\n`);
     }
 
     const timeoutMs = config.timeout * 1000;

packages/cli/src/pipeline/expressions.ts

@@ -289,25 +289,10 @@ function resolvePlannedExpression(
     return combineResolved(undefined, undefined, false, false);
   }
   if ("$js" in expression) {
-    const argsExpressions = (expression.args ?? {}) as Record<string, PipelineInputExpression>;
-    const hasFrom = Object.values(argsExpressions).some((v) => isRecord(v) && "$from" in v);
-    if (hasFrom) return combineResolved({ ...expression }, { ...expression }, false);
-    const code = expression.$js as string;
-    const resolvedArgs: Record<string, unknown> = {};
-    let sensitive = false;
-    for (const [key, argExpr] of Object.entries(argsExpressions)) {
-      const resolved = resolvePlannedExpression(argExpr, pipeline, runtimeInput);
-      resolvedArgs[key] = resolved.value;
-      sensitive = sensitive || resolved.sensitive;
-    }
-    try {
-      // eslint-disable-next-line @typescript-eslint/no-implied-eval
-      const fn = new Function("args", `return (${code})`);
-      const value = fn(resolvedArgs);
-      return combineResolved(value, sensitive ? REDACTED : value, sensitive);
-    } catch {
-      return combineResolved({ ...expression }, { ...expression }, false);
-    }
+    // Planning / dry-run must be a non-executing preview: never run user
+    // JavaScript here. Surface the expression as an unresolved placeholder so a
+    // `--dry-run` of an untrusted pipeline cannot trigger code execution.
+    return combineResolved({ ...expression }, { ...expression }, false);
   }
   return combineResolved(expression, expression, false);
 }

packages/cli/src/pipeline/scheduler.ts

@@ -70,6 +70,8 @@ export function orderReports(
   return [...reports].sort((a, b) => (index.get(a.id) ?? 0) - (index.get(b.id) ?? 0));
 }
 
+const MAX_CONCURRENCY = 64;
+
 export function normalizeConcurrency(value: number | undefined): number {
   if (value === undefined) return 1;
   if (!Number.isInteger(value) || value < 1) {
@@ -77,5 +79,7 @@ export function normalizeConcurrency(value: number | undefined): number {
       details: { issues: ["concurrency must be a positive integer"] },
     });
   }
-  return value;
+  // Cap fan-out so a single run cannot launch an unbounded number of concurrent
+  // API calls / downloads and exhaust sockets, file descriptors, or memory.
+  return Math.min(value, MAX_CONCURRENCY);
 }

packages/cli/src/pipeline/schema.ts

@@ -91,6 +91,14 @@ export function getByJsonPointer(value: unknown, pointer: string): unknown {
       continue;
     }
     if (isRecord(current)) {
+      // A JSON pointer over data must not reach object internals. Block
+      // prototype-polluting keys and only follow own properties so a crafted
+      // `$from`/`$input` path cannot pull out `constructor`/`__proto__` and feed
+      // it into downstream consumers.
+      if (segment === "__proto__" || segment === "constructor" || segment === "prototype") {
+        return undefined;
+      }
+      if (!Object.prototype.hasOwnProperty.call(current, segment)) return undefined;
       current = current[segment];
       continue;
     }

packages/cli/src/pipeline/validation.ts

@@ -125,6 +125,21 @@ function collectPipelineSemanticIssues(
         `semantic: step "${stepLabel}" timeout must be a positive number of seconds or duration string`,
       );
     }
+
+    // `script/js` executes its `code` as host JavaScript. Require it to be a
+    // literal string in the pipeline definition: code sourced from another step
+    // ($from) or any expression is rejected, so untrusted/model-generated text
+    // can never become the body of the executed function.
+    if (step.type === "script/js") {
+      const code = isRecord(step.input) ? step.input.code : undefined;
+      if (typeof code !== "string") {
+        issues.push(
+          `semantic: step "${stepLabel}" (script/js) requires a literal string "code"; ` +
+            `code sourced from another step ($from) or an expression is not allowed, ` +
+            `since it would execute untrusted text as host code`,
+        );
+      }
+    }
   }
 
   // Check dependency references

packages/cli/tests/index.test.ts

@@ -1,6 +1,9 @@
 import { expect, test } from "vite-plus/test";
 import { createStepDispatcher } from "../src/pipeline/dispatcher.ts";
 import { executePipeline } from "../src/pipeline/executor.ts";
+import { collectPipelineIssues } from "../src/pipeline/validation.ts";
+import { getByJsonPointer } from "../src/pipeline/schema.ts";
+import { normalizeConcurrency } from "../src/pipeline/scheduler.ts";
 import { WORKFLOW_VERSION, type PipelineDefinition } from "../src/pipeline/types.ts";
 
 test("cli package skeleton", () => {
@@ -34,3 +37,73 @@ test("pipeline execution can use an isolated step dispatcher", async () => {
     hasSignal: true,
   });
 });
+
+test("dry-run never executes $js expressions (preview must not run code)", async () => {
+  const dispatcher = createStepDispatcher();
+  dispatcher.registerStep("test/echo", (input) => ({ data: input }));
+  const flag = "__bailian_dryrun_should_not_run__";
+  delete (globalThis as Record<string, unknown>)[flag];
+
+  const pipeline: PipelineDefinition = {
+    version: WORKFLOW_VERSION,
+    steps: [
+      {
+        id: "s1",
+        type: "test/echo",
+        input: { probe: { $js: `(globalThis[${JSON.stringify(flag)}] = true), 1` } },
+      },
+    ],
+  };
+
+  const report = await executePipeline(pipeline, {}, { stepDispatcher: dispatcher, dryRun: true });
+  expect(report.status).toBe("planned");
+  expect((globalThis as Record<string, unknown>)[flag]).toBeUndefined();
+});
+
+test("script/js rejects non-literal code sourced from another step ($from)", () => {
+  const dispatcher = createStepDispatcher();
+  dispatcher.registerStep("test/echo", (input) => ({ data: input }));
+  dispatcher.registerStep("script/js", () => ({ data: {} }));
+
+  const pipeline: PipelineDefinition = {
+    version: WORKFLOW_VERSION,
+    steps: [
+      { id: "gen", type: "test/echo", input: { message: "x" } },
+      {
+        id: "run",
+        type: "script/js",
+        input: { code: { $from: "gen", path: "/data/message" } as never },
+      },
+    ],
+  };
+
+  const issues = collectPipelineIssues(pipeline, dispatcher);
+  expect(issues.some((issue) => issue.includes('literal string "code"'))).toBe(true);
+});
+
+test("script/js accepts a literal string code", () => {
+  const dispatcher = createStepDispatcher();
+  dispatcher.registerStep("script/js", () => ({ data: {} }));
+
+  const pipeline: PipelineDefinition = {
+    version: WORKFLOW_VERSION,
+    steps: [{ id: "run", type: "script/js", input: { code: "return 1" } }],
+  };
+
+  expect(collectPipelineIssues(pipeline, dispatcher)).toEqual([]);
+});
+
+test("getByJsonPointer refuses prototype keys and inherited properties", () => {
+  const obj = { a: { b: 1 } };
+  expect(getByJsonPointer(obj, "/a/b")).toBe(1);
+  expect(getByJsonPointer(obj, "/__proto__")).toBeUndefined();
+  expect(getByJsonPointer(obj, "/constructor")).toBeUndefined();
+  expect(getByJsonPointer(obj, "/a/constructor/constructor")).toBeUndefined();
+  expect(getByJsonPointer(obj, "/toString")).toBeUndefined();
+});
+
+test("normalizeConcurrency clamps to a safe maximum", () => {
+  expect(normalizeConcurrency(undefined)).toBe(1);
+  expect(normalizeConcurrency(4)).toBe(4);
+  expect(normalizeConcurrency(100000)).toBe(64);
+});

packages/core/src/client/endpoints.ts

@@ -24,13 +24,13 @@ export function videoGenerateEndpoint(baseUrl: string): string {
 // ---- Async Task Query ----
 
 export function taskEndpoint(baseUrl: string, taskId: string): string {
-  return `${baseUrl}/api/v1/tasks/${taskId}`;
+  return `${baseUrl}/api/v1/tasks/${encodeURIComponent(taskId)}`;
 }
 
 // ---- Application (Agent / Workflow) ----
 
 export function appCompletionEndpoint(baseUrl: string, appId: string): string {
-  return `${baseUrl}/api/v1/apps/${appId}/completion`;
+  return `${baseUrl}/api/v1/apps/${encodeURIComponent(appId)}/completion`;
 }
 
 // ---- Memory (DashScope v2) ----
@@ -48,7 +48,7 @@ export function memoryListEndpoint(baseUrl: string): string {
 }
 
 export function memoryNodeEndpoint(baseUrl: string, nodeId: string): string {
-  return `${baseUrl}/api/v2/apps/memory/memory_nodes/${nodeId}`;
+  return `${baseUrl}/api/v2/apps/memory/memory_nodes/${encodeURIComponent(nodeId)}`;
 }
 
 // ---- Speech Synthesis (TTS) ----
@@ -70,7 +70,7 @@ export function profileSchemaEndpoint(baseUrl: string): string {
 }
 
 export function userProfileEndpoint(baseUrl: string, schemaId: string): string {
-  return `${baseUrl}/api/v2/apps/memory/profile_schemas/${schemaId}/profiles`;
+  return `${baseUrl}/api/v2/apps/memory/profile_schemas/${encodeURIComponent(schemaId)}/profiles`;
 }
 
 // ---- MCP Services (Streamable HTTP) ----

packages/core/src/client/http.ts

@@ -4,6 +4,7 @@ import { BailianError } from "../errors/base.ts";
 import { ExitCode } from "../errors/codes.ts";
 import { resolveCredential } from "../auth/resolver.ts";
 import { mapApiError } from "../errors/api.ts";
+import { maskToken } from "../utils/token.ts";
 import { SOURCE_CONFIG, trackingHeaders } from "./headers.ts";
 
 export interface RequestOpts {
@@ -58,7 +59,7 @@ export async function request(config: Config, opts: RequestOpts): Promise<Respon
 
     if (config.verbose) {
       console.error(`> ${opts.method ?? "GET"} ${opts.url}`);
-      console.error(`> Auth: ${credential.token.slice(0, 8)}...`);
+      console.error(`> Auth: ${maskToken(credential.token)}`);
       console.error(`> x-dashscope-source-config: ${SOURCE_CONFIG}`);
     }
   }

packages/core/src/client/stream.ts

@@ -1,3 +1,6 @@
+import { BailianError } from "../errors/base.ts";
+import { ExitCode } from "../errors/codes.ts";
+
 export interface ServerSentEvent {
   event?: string;
   data: string;
@@ -11,12 +14,20 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
   const decoder = new TextDecoder();
   let buffer = "";
 
+  // Guard against a hostile or malfunctioning stream that never emits a newline
+  // (or builds a single absurdly large event): bound the in-memory buffer so the
+  // parser cannot be driven to exhaust process memory.
+  const MAX_SSE_BUFFER = 16 * 1024 * 1024; // 16 MiB
+
   try {
     while (true) {
       const { done, value } = await reader.read();
       if (done) break;
 
       buffer += decoder.decode(value, { stream: true });
+      if (buffer.length > MAX_SSE_BUFFER) {
+        throw new BailianError("SSE stream exceeded the maximum buffer size.", ExitCode.GENERAL);
+      }
 
       const lines = buffer.split("\n");
       buffer = lines.pop() || "";
@@ -43,6 +54,12 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
         switch (field) {
           case "data":
             event.data = event.data !== undefined ? `${event.data}\n${value}` : value;
+            if (event.data.length > MAX_SSE_BUFFER) {
+              throw new BailianError(
+                "SSE event exceeded the maximum buffer size.",
+                ExitCode.GENERAL,
+              );
+            }
             break;
           case "event":
             event.event = value;

packages/core/src/config/paths.ts

@@ -20,4 +20,13 @@ export async function ensureConfigDir(): Promise<void> {
   const dir = getConfigDir();
   const fs = await import("fs/promises");
   await fs.mkdir(dir, { recursive: true, mode: 0o700 });
+  // `mkdir`'s `mode` only applies to directories it creates (and is masked by
+  // umask). A config dir created by an older build or another tool may still be
+  // world/group-readable while holding cleartext credentials, so tighten it
+  // explicitly. Best-effort: never let a chmod failure break the command.
+  try {
+    await fs.chmod(dir, 0o700);
+  } catch {
+    /* best effort */
+  }
 }

packages/core/src/config/schema.ts

@@ -38,6 +38,21 @@ export interface ConfigFile {
 const VALID_REGIONS = new Set<string>(["cn", "us", "intl"]);
 const VALID_OUTPUTS = new Set<string>(["text", "json"]);
 
+/**
+ * A syntactically valid absolute http(s) URL. Used to validate `base_url` and
+ * `console_gateway_url` from the config file: the credential-bearing client
+ * sends the Bearer token to these origins, so a bare `startsWith("http")` check
+ * (which also accepts e.g. "httpfoo://…") is too loose.
+ */
+function isHttpUrl(value: string): boolean {
+  try {
+    const u = new URL(value);
+    return u.protocol === "http:" || u.protocol === "https:";
+  } catch {
+    return false;
+  }
+}
+
 export function parseConfigFile(raw: unknown): ConfigFile {
   if (!raw || typeof raw !== "object" || Array.isArray(raw)) return {};
   const obj = raw as Record<string, unknown>;
@@ -50,8 +65,7 @@ export function parseConfigFile(raw: unknown): ConfigFile {
     out.access_token = obj.accessToken;
   if (typeof obj.region === "string" && VALID_REGIONS.has(obj.region))
     out.region = obj.region as Region;
-  if (typeof obj.base_url === "string" && obj.base_url.startsWith("http"))
-    out.base_url = obj.base_url;
+  if (typeof obj.base_url === "string" && isHttpUrl(obj.base_url)) out.base_url = obj.base_url;
   if (typeof obj.output === "string" && VALID_OUTPUTS.has(obj.output))
     out.output = obj.output as ConfigFile["output"];
   if (typeof obj.output_dir === "string" && obj.output_dir.length > 0)
@@ -73,7 +87,7 @@ export function parseConfigFile(raw: unknown): ConfigFile {
     out.access_key_secret = obj.access_key_secret;
   if (typeof obj.workspace_id === "string" && obj.workspace_id.length > 0)
     out.workspace_id = obj.workspace_id;
-  if (typeof obj.console_gateway_url === "string" && obj.console_gateway_url.startsWith("http"))
+  if (typeof obj.console_gateway_url === "string" && isHttpUrl(obj.console_gateway_url))
     out.console_gateway_url = obj.console_gateway_url;
   if (typeof obj.telemetry === "boolean") out.telemetry = obj.telemetry;
 

packages/core/src/telemetry/sink.ts

@@ -90,7 +90,7 @@ export async function localSink(event: TrackingEvent): Promise<void> {
       // 文件还不存在，忽略
     }
 
-    appendFileSync(path, JSON.stringify(event) + "\n");
+    appendFileSync(path, JSON.stringify(event) + "\n", { mode: 0o600 });
   } catch {
     // 埋点逻辑任何异常都不能影响 CLI 主流程
   }

packages/core/tests/index.test.ts

@@ -1,6 +1,7 @@
 import { expect, test } from "vite-plus/test";
 import type { Config } from "../src/index.ts";
 import { BailianError, ExitCode, McpClient, mapApiError, request } from "../src/index.ts";
+import { parseConfigFile } from "../src/config/schema.ts";
 
 function testConfig(overrides: Partial<Config> = {}): Config {
   return {
@@ -173,3 +174,16 @@ test("McpClient uses injected client identity for initialize and User-Agent", as
     params: { clientInfo: { name: "test-client", version: "9.8.7" } },
   });
 });
+
+test("parseConfigFile accepts only well-formed http(s) base_url / console_gateway_url", () => {
+  expect(parseConfigFile({ base_url: "https://dashscope.aliyuncs.com" }).base_url).toBe(
+    "https://dashscope.aliyuncs.com",
+  );
+  expect(parseConfigFile({ base_url: "http://localhost:8080" }).base_url).toBe(
+    "http://localhost:8080",
+  );
+  // Previously accepted because the value merely "starts with http".
+  expect(parseConfigFile({ base_url: "httpfoo://evil" }).base_url).toBeUndefined();
+  expect(parseConfigFile({ base_url: "not a url" }).base_url).toBeUndefined();
+  expect(parseConfigFile({ console_gateway_url: "ftp://x" }).console_gateway_url).toBeUndefined();
+});