09 工具策略与安全原理（工具防火墙怎么落地）

这篇讲清楚两件事：

1. 哪些工具可以被模型看到
2. 哪些命令必须人工审批

核心源码入口

• src/agents/pi-tools.ts
• src/agents/tool-policy.ts
• src/agents/pi-tools.policy.ts
• src/agents/tool-policy-pipeline.ts
• src/agents/pi-tools.before-tool-call.ts
• src/agents/pi-tool-definition-adapter.ts
• src/infra/exec-approvals.ts
• src/gateway/exec-approval-manager.ts
• src/gateway/server-methods/exec-approval.ts

总体状态机

工具全集 -> owner 裁剪 -> 多层策略过滤 -> schema 规范化 -> before_tool_call 二次拦截 -> 执行 -> after_tool_call 审计 -> 高风险审批决策

模块一 工具构建与分层过滤（精确到代码顺序）

createOpenClawCodingTools(...) 的真实执行顺序（源码 pi-tools.ts）：

第 1 步：applyOwnerOnlyToolPolicy

// senderIsOwner 默认 false — 安全优先，opt-in，不是 opt-out
const senderIsOwner = options?.senderIsOwner === true;
const toolsByAuthorization = applyOwnerOnlyToolPolicy(tools, senderIsOwner);

非 owner 请求在这里先砍掉所有 owner-only 工具（高风险能力），后续步骤看不到它们。

第 2 步：applyToolPolicyPipeline（多步骤顺序固定）

buildDefaultToolPolicyPipelineSteps(...) 生成的步骤顺序（源码精确标签）：

[
  { policy: profilePolicy,         label: `tools.profile (${profile})`,         stripPluginOnlyAllowlist: true  },
  { policy: providerProfilePolicy, label: `tools.provider-profile (${profile})`, stripPluginOnlyAllowlist: true  },
  { policy: globalPolicy,          label: "tools.global"                                                         },
  { policy: globalProviderPolicy,  label: "tools.global-provider"                                                },
  { policy: agentPolicy,           label: `tools.agent (${agentId})` },           // 仅 agentId 非空时存在
  { policy: agentProviderPolicy,   label: `tools.agent-provider (${agentId})` },  // 仅 agentId 非空时存在
  { policy: groupPolicy,           label: "group tools.allow",                   stripPluginOnlyAllowlist: true  },
  // pi-tools.ts 在 pipeline 外追加：
  { policy: sandbox?.tools,        label: "sandbox tools.allow"                                                  },
  { policy: subagentPolicy,        label: "subagent tools.allow"                                                 },
]

顺序是逐层收紧，后面的步骤在前面步骤结果的基础上继续过滤，不会"后面覆盖前面"。

stripPluginOnlyAllowlist: true 的作用（关键防护）

profile / provider-profile / group 的 allowlist 如果只包含插件专有工具（但该插件未启用）， pipeline 会自动剥离该 allowlist，避免出现"allowlist 存在但无一匹配"导致核心工具全部被清空的问题。

// 源码 tool-policy-pipeline.ts
if (step.stripPluginOnlyAllowlist) {
  const resolved = stripPluginOnlyAllowlist(policy, pluginGroups, coreToolNames);
  if (resolved.unknownAllowlist.length > 0) {
    params.warn(`tools: ${step.label} allowlist contains unknown entries (${entries}). ${suffix}`);
  }
  policy = resolved.policy;
}

第 3 步：normalizeToolParameters（schema 规范化）

const normalized = subagentFiltered.map(normalizeToolParameters);

把工具的 JSON Schema 规范化后再交给 pi-agent/pi-ai。 原因：OpenAI 会拒绝根级 union schema（如 anyOf 在根节点），必须先展开。 没有这一步，部分工具在 OpenAI provider 下会被服务端拒绝，且报错不直观。

第 4 步：wrapToolWithBeforeToolCallHook（Hook 注入）

const withHooks = normalized.map((tool) =>
  wrapToolWithBeforeToolCallHook(tool, { agentId, sessionKey: options?.sessionKey }),
);

第 5 步：wrapToolWithAbortSignal（abort 信号，可选）

const withAbort = options?.abortSignal
  ? withHooks.map((tool) => wrapToolWithAbortSignal(tool, options.abortSignal))
  : withHooks;

模块二 before_tool_call Hook 精确行为

调用流程（源码 pi-tools.before-tool-call.ts）

const hookResult = await hookRunner.runBeforeToolCall({ toolName, params });

if (hookResult?.block) {
  // blocked: true → 直接 throw，工具不执行
  throw new Error(hookResult.blockReason || "Tool call blocked by plugin hook");
}

if (hookResult?.params && isPlainObject(hookResult.params)) {
  // 参数改写：合并方式是 { ...originalParams, ...hookResult.params }
  return { blocked: false, params: { ...params, ...hookResult.params } };
}

adjustedParamsByToolCallId（参数暂存机制）

改写后的参数不能直接用 hook 返回值，因为 pi-agent-core 会从 wrappedTool.execute 外层 拿到 toolCallId，再从内部暂存 Map 取回实际参数：

// wrapToolWithBeforeToolCallHook 内部
adjustedParamsByToolCallId.set(toolCallId, outcome.params);

// pi-tool-definition-adapter.ts 取回
const afterParams = beforeHookWrapped
  ? (consumeAdjustedParamsForToolCall(toolCallId) ?? executeParams)
  : executeParams;

Map 有上限（MAX_TRACKED_ADJUSTED_PARAMS），超出后自动淘汰最旧条目，防止内存泄漏。

多插件合并规则（runModifyingHook 的顺序执行合并）

// 每个插件的 before_tool_call 结果按注册顺序合并
mergeResults: (acc, next) => ({
  params:      next.params      ?? acc?.params,
  block:       next.block       ?? acc?.block,
  blockReason: next.blockReason ?? acc?.blockReason,
})
// 最后一个插件的 params 覆盖前面的（last-write-wins）

模块三 after_tool_call Hook 精确行为

触发条件（成功和失败路径都触发）

// pi-tool-definition-adapter.ts — 成功路径
hookRunner.runAfterToolCall({ toolName, params: afterParams, result }, { toolName });

// 失败路径（catch 分支）
hookRunner.runAfterToolCall({ toolName, params, error: described.message }, { toolName });

两条路径都是 fire-and-forget（不 await）。Hook 失败只记 debug log，不影响工具返回值。

durationMs 计算（源码 pi-embedded-subscribe.handlers.tools.ts）

订阅层里额外跟踪了工具执行耗时：

// tool_execution_start 时记录
toolStartData.set(toolCallId, { startTime: Date.now(), args });

// tool_execution_end 时触发 after_tool_call
const durationMs = startData?.startTime != null ? Date.now() - startData.startTime : undefined;
const hookEvent: PluginHookAfterToolCallEvent = {
  toolName, params, result: sanitizedResult,
  error: isToolError ? extractToolErrorMessage(sanitizedResult) : undefined,
  durationMs,
};

模块四 子智能体默认收敛策略

resolveSubagentToolPolicy(...) 默认 deny 一批工具：

• sessions_spawn/sessions_send/sessions_list/sessions_history
• gateway/agents_list/cron
• memory_search/memory_get

目的：防止子智能体获得"编排主控权"（不能再 spawn、不能发消息给其他 session、不能操作 cron）。

模块五 apply_patch 的 provider 限制（特殊门控）

apply_patch 默认仅对 OpenAI provider 开放：

// 源码测试用例证明了此行为
it("gates apply_patch behind tools.exec.applyPatch for OpenAI models", () => {
  const openAiTools  = createOpenClawCodingTools({ config, modelProvider: "openai",     modelId: "gpt-5.2"     });
  const anthropicTools = createOpenClawCodingTools({ config, modelProvider: "anthropic", modelId: "claude-opus-4-5" });

  expect(openAiTools.some((t) => t.name === "apply_patch")).toBe(true);
  expect(anthropicTools.some((t) => t.name === "apply_patch")).toBe(false);
});

可通过 tools.exec.applyPatch.allowModels 配置扩展允许的 model 列表。

模块六 exec 审批判定逻辑

ExecApprovalDecision 只有三种：

• allow-once
• allow-always
• deny

requiresExecApproval(...) 的核心条件：

1. ask=always：永远审批
2. ask=on-miss 且 security=allowlist：命令不满足白名单或分析失败时审批
3. 其他场景不审批

模块七 审批状态机（gateway）

ExecApprovalManager：

1. create(...)：生成 id/createdAtMs/expiresAtMs
2. register(...)：
   • 写入 pending map
   • 同 id 未决时幂等返回同一 promise
   • timeout 触发 resolve(null)
3. resolve(...)：
   • 写入 decision/resolvedAtMs/resolvedBy
   • 释放等待方
4. grace 保留：已决条目保留 15s，给 waitDecision 收尾

server-methods/exec-approval.ts 三个 RPC：

1. exec.approval.request
2. exec.approval.waitDecision
3. exec.approval.resolve

最小复刻骨架

function buildAllowedTools(all: Tool[], ctx: PolicyCtx) {
  // Step 1: owner-only 裁剪
  const ownerFiltered = applyOwnerOnlyToolPolicy(all, ctx.senderIsOwner);
  // Step 2: 多层 pipeline（顺序固定）
  const pipelineFiltered = applyToolPolicyPipeline(ownerFiltered, buildDefaultSteps(ctx));
  // Step 3: schema 规范化（防 OpenAI union schema 报错）
  const normalized = pipelineFiltered.map(normalizeToolParameters);
  // Step 4: Hook 注入
  const withHooks = normalized.map((t) => wrapWithBeforeToolCallHook(t, ctx));
  return withHooks;
}

async function executeWithApproval(cmd: string, ctx: PolicyCtx) {
  if (!requiresApproval(cmd, ctx)) {
    return runExec(cmd);
  }
  const record = approvalManager.create({ command: cmd }, 120_000);
  const pending = approvalManager.register(record, 120_000);
  notifyClientRequested(record);
  const decision = await pending;
  if (decision !== "allow-once" && decision !== "allow-always") {
    throw new Error("exec denied");
  }
  return runExec(cmd);
}

自检清单

1. 非 owner 是否拿不到 owner-only 工具。
2. pipeline 步骤顺序（profile → global → agent → group → sandbox → subagent）是否固定不变。
3. stripPluginOnlyAllowlist 是否在 profile/provider-profile/group 步骤开启。
4. schema 规范化步骤是否在 Hook 注入之前执行。
5. before_tool_call 修改的参数是否通过 adjustedParamsByToolCallId 传递，而不是直接覆盖。
6. after_tool_call 在错误分支是否仍触发（fire-and-forget）。
7. 审批 timeout 是否返回 null 而不是永远挂起。
8. apply_patch 是否对非 OpenAI provider 默认关闭。