Agent Governance Toolkit 实战在 kubernetes-sigs/agent-sandbox 命令进入 Pod 之前完成策略拦截【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit本文以仓库中的 k8s-agent-sandbox-governed 示例 为蓝本讲解如何用 Agent Governance Toolkit下称 AGT的govern()包装器对 Agent 向kubernetes-sigs/agent-sandbox沙箱 Pod 下发的每条命令做发送前策略检查。读完本文你将掌握一条可复用的治理模式自由文本命令如何被预分类为离散动作类型、policy.yaml如何以 fail-closed 语义裁决、被拒命令为何在申领任何沙箱 Pod 之前就被拦下以及这套方案的已知边界与扩展方向。背景agent-sandbox 提供了隔离但不提供语义策略kubernetes-sigs/agent-sandbox是 CNCF SIG 主导的 Kubernetes 原生执行隔离方案为 Agent 工作负载提供 Pod 级沙箱与按模板生成的NetworkPolicy。它的 CRD、控制器与 sandbox-router 均来自上游项目自身的发布版本本示例不做任何改动只在其 Python SDK 之上加了一层客户端侧治理包装。关键缺口在于agent-sandbox 的隔离能力集中在 L3/L4 网络层与进程/Pod 隔离它本身不裁决Agent 要求沙箱执行的内容是否安全。正如示例驱动脚本 run_agent.py 的 docstring 所写网络策略只覆盖 L3/L4不覆盖命令内容。rm -rf /这类命令一旦进入 Pod隔离层只能在事后兜底——把它关在容器里却无法阻止它发生。AGT 的govern()正好填补这一层它在驱动脚本自己的进程内、命令发往沙箱 Pod 执行 API之前完成评估被拒命令根本不会到达 Pod而不是到达后再被隔离手段拦截。这构成纵深防御defense-in-depth治理是隔离之前的检查不是隔离的替代品隔离仍是最后一道防线。整体架构一次调用的完整路径Agent driving script (run_agent.py) | v _classify_command() -- 预分类把命令 本地脚本正文归类为离散 action.type | 策略 DSL 只做字段比较不做自由文本匹配 v govern(...) -- 用 {type: ..., command: ...} 上下文评估 policy.yaml |-- allow -- create_sandbox() - files.write() - sandbox.commands.run(...) -- deny -- 抛出 GovernanceDenied任何沙箱 Pod 都未被申领两个关键设计点被拒命令永远不会甚至不会申领一个沙箱 Pod——决策发生在create_sandbox()之前避免了为治理本来就会拒绝的请求白白消耗 warm-pool 中的 Pod 名额。职责分离policy.yaml拥有 allow/deny 规则与审计线索run_agent.py负责把自由文本命令/脚本预分类成策略条件能够匹配的离散action.type值。快速开始五分钟跑通 allow 与 deny 两条路径1. 准备一个 agent-sandbox 集群按 agent-sandbox 上游的 quickstart 指南在任意集群默认 kind的agent-sandbox-demo命名空间里跑起一个python-warmpool。2. 安装依赖cd examples/k8s-agent-sandbox-governed pip install -r requirements.txtrequirements.txt 只有两个依赖agent-governance-toolkit[full]5.0.0—— 本仓库提供govern()/GovernanceDeniedk8s-agent-sandbox0.1.0—— 上游 Python SDK用于与 agent-sandbox 的 sandbox-router 通信。不需要任何其他第三方包。3. 运行良性脚本——放行python run_agent.py hello_world.py --warmpool python-warmpool --namespace agent-sandbox-demohello_world.py 只是一行print(Hello from a governed sandbox pod)被分类为shell_exec命中唯一一条 allow 规则输出Hello from a governed sandbox pod4. 运行破坏性脚本——下发前即被拒绝python run_agent.py destructive.sh --interpreter bash \ --warmpool python-warmpool --namespace agent-sandbox-demodestructive.sh 是一个刻意构造的夹具正文只有一行rm -rf /nonexistent-governance-test-path路径指向不存在的目录绝不会真正造成破坏注释中明确说明这是为了在不真正执行磁盘擦除命令的前提下演示 deny 路径。输出Command blocked by governance policy: Action denied by policy rule block-destructive-commands: Commands classified as destructive (rm -rf, mkfs, dd, disk wipes, fork bombs, pipe-to-shell installers) are blocked before dispatch.进程退出码为1rm -rf从未在 Pod 内执行——对比一下靠 Pod 自身隔离去事后兜住它的方案。深入 run_agent.py自由文本 → 离散 action.type 的预分类AGT 的策略条件 DSL 面向的是结构化字段比较而不是自由文本正则。因此所有模式匹配都被前置到_classify_command()见 run_agent.py它把命令与脚本正文归入三类destructive、credential_exfil、shell_exec。值得注意的实现细节命令与脚本正文分开匹配不拼接成一个字符串避免curl 出现在命令末尾、| sh出现在脚本开头这类跨边界伪命中。argv 层与原始文本层双层检测基于shlex的 token 级解析_line_segments/_is_destructive_segment解决引号/转义/flag 重排绕过例如r\m -rf /、rm -rf /、rm -r -f /、dd of/dev/sda if/dev/zero_strip_wrapper剥掉sudo/env/xargs包装含sudo -u root rm -rf /这类带参 flag让包装后的命令与裸rm -rf /归一到同一 argv原始文本层_DESTRUCTIVE_SYNTAX_PATTERNS兜住 argv 解析看不穿的形态fork bomb、管道安装器curl ... | sh、bash -c ...、子 shell$(...)、重定向前缀、以及脚本正文里os.system(rm -rf /)这类嵌入其他语言字符串的情况。无法解析的引号一律 fail-safe 视为破坏性_line_segments返回 None 时直接归类为 destructive。命令行参数解析上main() 自行以字面--切分脚本参数避免argparse.REMAINDER吞掉自家 flag。policy.yaml 详解fail-closed 的策略骨架policy.yaml 是治理的核心声明完整内容如下apiVersion: governance.toolkit/v1 name: agent-sandbox-command-policy description: Governs commands dispatched from an agents driving script to a kubernetes-sigs/agent-sandbox pods execution API. Evaluated in the agents own process, before the command ever reaches the pod — this complements (does not replace) the sandboxs NetworkPolicy/pod isolation, which is the last line of defense if a command does get through. # Fail closed: any action.type not explicitly allowed below is denied, # including types _classify_command() doesnt yet know how to detect. default_action: deny rules: - name: block-destructive-commands condition: action.type destructive action: deny description: Commands classified as destructive (rm -rf, mkfs, dd, disk wipes, fork bombs, pipe-to-shell installers) are blocked before dispatch. - name: block-credential-exfil condition: action.type credential_exfil action: deny description: Commands that read AWS/K8s credential material and forward it externally (curl/wget/nc combined with env var expansion of AWS_* or KUBECONFIG) are blocked before dispatch. - name: allow-shell-exec condition: action.type shell_exec action: allow description: Commands that didnt match any higher-risk classification are explicitly allowed. This is the only allow rule — anything else falls through to default_action: deny.三个规则之外最值得注意的是default_action: deny任何未显式放行的action.type都会被拒绝包括_classify_command()未来新增的分类、甚至它现在还没能力识别的类型。这符合底层 policy.py 中Policy.default_action的默认值语义缺省即deny。从源码层面看条件 DSL 由 PolicyRule._eval_expression 实现除之外还支持!、in [...]成员判断、contains、startswith、endswith、数值比较 、布尔属性真值以及and/or复合条件表达式超长2000 字符、嵌套过深20 层或语法无法识别时非 allow 规则按命中处理fail-closedallow 规则按未命中处理防止恶意输入通过触发异常绕过规则见 policy.py 的 fail-closed 注释。策略文档还支持apiVersion: governance.toolkit/v1的版本校验与extends父策略继承子策略不能削弱父级 deny 规则。govern() 集成模式包装 dispatch申领沙箱发生在治理之后示例的核心集成代码见 README.md 的 Integration pattern及 run_agent.py 的完整实现from agentmesh.governance import govern, GovernanceDenied def _dispatch(action): # Sandbox is claimed here, inside the governed call, so a denied # action never costs a pod claim. sandbox client.create_sandbox(warmpool..., namespace...) sandbox.files.write(remote_name, script_bytes) return sandbox.commands.run(action[command], timeout60) governed_run govern( _dispatch, policypolicy.yaml, agent_idrun_agent:agent-sandbox-demo, ) try: result governed_run(action{type: action_type, command: command}) except GovernanceDenied as e: print(fblocked: {e})底层原理见 govern.pygovern()工厂govern.py把任意可调用对象包装成GovernedCallable返回的包装对象与原函数 API 兼容functools.update_wrapper。GovernedCallable.__call__govern.py的执行顺序用 kwargs 构建评估上下文action键直接作为{type: ..., command: ...}字典传给策略→ 策略引擎evaluate默认冲突策略deny_overrides任何 deny 优先→ 审计policy_evaluation事件含命中的规则名、原因、评估耗时→ 若不允许则调用on_deny回调或抛GovernanceDenied→ 只有允许才真正执行被包装的_dispatch。GovernanceDeniedgovern.py的异常消息就是 README 里展示的那句Action denied by policy rule : 其中 rule 与 reason 来自PolicyDecision。govern()还支持audit_fileHash 链 HMAC 签名的 JSONL 审计落盘密钥需显式传入或设置AGT_AUDIT_SECRET_KEY环境变量、on_deny回调、ring执行环、require_approval审批流、rego_path/rego_content并行加载 OPA 策略等扩展能力详见 GovernanceConfig。三个容易被忽略的安全细节单次读取脚本内容run_agent.py分类与上传共用同一次read_bytes()避免 TOCTOU——文件在两次读取之间被替换以安全版本通过审查、以恶意版本执行。shlex.join()组装命令run_agent.pyk8s-agent-sandbox的commands.run()只接受单个 shell 命令字符串而非 argv 列表因此脚本参数用shlex.join()逐个加引号防止;、|、$()、反引号等元字符被沙箱 Pod 的 shell 当作额外语法解释。finally块保证sandbox.terminate()run_agent.py每次运行后单个沙箱都会被终止不留残留。测试验证test_run_agent.py 里的反绕过样本test_run_agent.py 通过参数化测试固化了分类器的行为值得借鉴的是它把绕过形态分成了三类链式与包装绕过echo ok rm -rf /、$(rm -rf /)、sudo -u root rm -rf /、env FOObar rm -rf /、echo hi | xargs rm -rf引号/转义/flag 重排绕过rm -r -f /、r\m -rf /、mkfs.ext4 /dev/sda1、dd of/dev/sda if/dev/zero仅原始文本层能抓到的形态nohup/timeout/exec/eval/busybox/doas/su -c/bash -c/子 shell/函数体/find -exec/重定向前缀/!前缀以及脚本内嵌os.system(rm -rf /)。同时用一组良性命令如transform -rf foo、rm file.txt、cat file.txt | grep foo防止误杀回归无法解析的引号echo unterminated必须 fail-safe 归类为 destructive。清理run_agent.py已在每次调用后用finally终止其创建的单体沙箱无需按次清理。共享资源按需拆除kubectl delete sandboxwarmpool python-warmpool -n agent-sandbox-demo kubectl delete sandboxtemplate python-sandbox-template -n agent-sandbox-demo若为示例专门创建了集群再按 agent-sandbox quickstart 的 Cleanup 步骤如kind delete cluster ...整体删除。已知局限与扩展建议原文明确声明了三条边界部署前必须知晓_classify_command()是演示级过滤器不是生产级检测器。匹配大小写不敏感、flag 顺序容忍但模式集本身很窄、可以被绕过例如把凭据管道给curl/wget/nc之外的命令、或使用模式集中未覆盖的破坏/外传手法。不要把这份 policy 原样当作生产风险模型——它是起点不是终点。分类只发生一次且只针对本地上传的脚本副本。它不覆盖沙箱运行时下载或动态生成的代码。commands.run()只接受单条 shell 命令字符串shlex.join()只保证参数不被 shell 二次解释不改变脚本运行后的行为——恶意脚本正文仍只受_classify_command()的模式匹配约束。扩展方向按你自己的风险模型扩展_classify_command()与策略的rules列表default_action: deny保证新增但未识别的分类默认被拒远程集群场景需把SandboxLocalTunnelConnectionConfig面向kubectl port-forward或本机 kind/minikube替换为 SDK 提供的直连/集群内连接配置。本示例的 README、policy 与驱动脚本虽标注由 Claude Code 起草但已经作者在真实集群上人工复核验证见 README.md 的 Disclaimer 说明。参考资料示例入口与完整说明examples/k8s-agent-sandbox-governed/README.md驱动脚本分类器 govern 包装examples/k8s-agent-sandbox-governed/run_agent.py策略文件examples/k8s-agent-sandbox-governed/policy.yaml反绕过测试样本examples/k8s-agent-sandbox-governed/test_run_agent.pygovern()底层实现agent-governance-python/agent-mesh/src/agentmesh/governance/govern.py策略引擎与条件 DSLagent-governance-python/agent-mesh/src/agentmesh/governance/policy.py【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考