Agent Governance Toolkit 协议面（Protocol Facets）：面向线级语义的 SQL 与 Kubernetes 策略评估指南

Agent Governance Toolkit 协议面（Protocol Facets）：面向线级语义的 SQL 与 Kubernetes 策略评估指南

【免费下载链接】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 Agent 在执行数据库查询、Kubernetes API 调用等高危操作时，传统策略引擎只能看到工具名或 HTTP 元数据，无法识别一条 SQL 到底是SELECT还是DROP、一个 K8s 请求是否指向production命名空间。Agent Governance Toolkit 的Protocol Facets（协议面）机制，在PolicyEngine.evaluate()进入规则匹配之前，从原始协议上下文（SQL 语句、Kubernetes API 路径）中结构化提取sql.*与k8s.*字段并合并进评估上下文，使 YAML 规则可以用点号（dot-notation）条件直接引用线级语义。阅读本文后，你将掌握 SQL/K8s 面字段的提取规则、HTTP 方法到 Kubernetes 动词的映射、无侵入的 MCP 透明代理集成方式、自定义协议解析器的注册方法，以及该模型在 Python、Rust、TypeScript、.NET 等多语言 SDK 中的一致性实现。

工作机制：规则求值前的结构化提取

PolicyEngine.evaluate()的入口逻辑位于 agent-governance-python/agent-mesh/src/agentmesh/governance/policy.py：

# Populate sql.* and k8s.* fields before rules run from agentmesh.governance.protocol_facets import extract_protocol_facets extract_protocol_facets(context)

也就是说，每次调用evaluate()时，系统都会先运行extract_protocol_facets(context)再评估规则。只要上下文（context）中包含sql或k8s子字典，对应的解析器就会填充结构化字段，随后 YAML 规则即可用{field, operator, value}条件引用这些字段。面提取发生在规则匹配之前，因此规则作者完全不需要关心原始 SQL 解析或 K8s 路径正则的细节——只需按字段名书写条件。

从源码结构看，面提取模块 protocol_facets.py 由三部分组成：

• FacetRegistry类：按上下文键名持有面提取器，提取器收到对应键的子字典并返回要合并的字段；提取器内部异常会被捕获并记录日志，单个解析器故障不会阻塞策略评估；
• SQL 解析器_extract_sql_facets与 K8s 解析器_extract_k8s_facets；
• 模块级default_registry：预注册了sql与k8s两个提取器，并对外暴露extract_protocol_facets(context, registry=None)便捷函数（未传 registry 时使用default_registry）。

SQL 面（sql.*）

字段定义

调用方在上下文中放入context["sql"]["query"]（SQL 语句字符串），解析器提取以下字段供策略条件引用：

依赖说明：SQL 解析依赖sqlglot（pip install sqlglot）。未安装或解析失败时，sql.verb会被置为UNKNOWN——这是一种fail-closed（失败即拒绝倾向）行为，避免因解析器缺失而放过危险语句。

底层解析原理

从 protocol_facets.py 的源码可以看到_extract_sql_facets的实现思路：

• 通过sqlglot.parse(query)将语句解析为 AST，再依据 AST 节点类型（exp.Select、exp.Insert、exp.Update、exp.Delete、exp.Drop、exp.Create、exp.AlterTable、exp.Grant、exp.Merge等）判定动词；
• 对于无法归类的Command节点，通过_SQL_VERB_MAP按命令名映射（覆盖TRUNCATE、REVOKE、CALL、EXECUTE、EXPLAIN、WITH等）；
• tables通过stmt.find_all(exp.Table)收集，functions通过stmt.find_all(exp.Func)收集并转为大写；
• target取语句涉及的第一张表（例如SELECT * FROM orders JOIN users ...的 target 是orders）。

空查询或纯空白查询返回空字符串字段；缺少sqlglot时返回UNKNOWN。这些边界行为在 test_protocol_facets.py 中有完整的测试覆盖，包括test_no_sqlglot_returns_unknown（mock 掉sqlglot导入后断言 verb 为UNKNOWN）。

示例规则

每条规则通过{field, operator, value}匹配单个提取字段。复合检查（例如动词 + 目标组合）应拆分为多条带合适优先级的规则：

rules: - name: deny-destructive-sql condition: {field: "sql.verb", operator: in, value: ["DROP", "TRUNCATE", "DELETE"]} action: deny priority: 100 - name: deny-schema-changes condition: {field: "sql.verb", operator: in, value: ["ALTER", "GRANT", "REVOKE"]} action: deny priority: 100 - name: allow-read-only-sql condition: {field: "sql.verb", operator: eq, value: "SELECT"} action: allow priority: 5

示例评估上下文

engine.evaluate( agent_did="did:example:agent1", context={"sql": {"query": "DROP TABLE production"}}, )

上面的上下文经extract_protocol_facets处理后，sql.verb为DROP、sql.target为production，从而命中deny-destructive-sql规则。集成测试 TestPolicyEngineIntegration.test_sql_drop_denied_by_rule 验证了DROP TABLE users会被拒绝且matched_rule == "deny-drop"；同文件还验证了DROP TABLE staging（非 protected 目标）在带default_action: allow的策略下可以放行，说明规则可以精确到“目标表”粒度。

Kubernetes 面（k8s.*）

字段定义

调用方在上下文中放入context["k8s"]["method"]（HTTP 方法）与context["k8s"]["path"]（API Server 路径），解析器提取以下字段：

HTTP 方法到 Kubernetes 动词的映射

源码中还额外处理了HEAD（具名资源映射为get，集合映射为list）。从 protocol_facets.py 可以看到，判定“具名”还是“集合”的关键在于路径是否包含资源名称：_K8S_PATH_PATTERNS中按“最具体优先”排列了 10 条正则，覆盖/api/<version>/...与/apis/<group>/<version>/...两类前缀下的具名/集合/命名空间/子资源组合；若路径匹配到name组，则用_METHOD_TO_VERB_NAMED映射，否则用_METHOD_TO_VERB_COLLECTION映射。例如GET /api/v1/namespaces/default/pods/mypod→ verb 为get，而GET /api/v1/namespaces/default/pods→ verb 为list（见 test_get_named_resource 与 test_get_collection）。

示例规则

rules: - name: deny-k8s-production-namespace condition: {field: "k8s.namespace", operator: eq, value: "production"} action: deny priority: 110 - name: deny-k8s-exec condition: {field: "k8s.subresource", operator: eq, value: "exec"} action: deny priority: 100 - name: deny-k8s-deletecollection condition: {field: "k8s.verb", operator: eq, value: "deletecollection"} action: deny priority: 100 - name: allow-k8s-readonly condition: {field: "k8s.verb", operator: in, value: ["get", "list", "watch"]} action: allow priority: 5

示例评估上下文

engine.evaluate( agent_did="did:example:agent1", context={ "k8s": { "method": "DELETE", "path": "/api/v1/namespaces/production/pods/mypod", } }, )

该上下文解析后k8s.verb = delete、k8s.namespace = production、k8s.resource = pods、k8s.name = mypod，命中deny-k8s-production-namespace。对应集成测试 test_k8s_delete_prod_denied 验证了 production 下的删除被拒绝，而 test_k8s_delete_staging_allowed 验证了 staging 下的同形态删除在default_action: allow下放行——这是“按命名空间隔离风险”的典型用法。

透明代理集成：零应用改造

MCP 代理（agentmesh proxy）会自动从工具调用参数中填充线级协议上下文，无需修改任何应用代码：

• 名为query或sql的工具参数 → 填充context["sql"]["query"]
• 名为method/http_method配合path/api_path且路径以/api/或/apis/开头 → 填充context["k8s"]

因此，只要在策略文件中定义 SQL 或 K8s 规则，这些规则就会自动作用于所有经过代理的工具调用。这对于治理 Agent 通过 MCP 工具触发的数据库操作和集群管理操作非常实用：治理规则与业务代码解耦，新增协议面规则不需要任何应用侧改动。

扩展自定义协议解析器

Python：注册到 default_registry

面提取模块的文档字符串与测试用例（test_register_and_extract_custom_protocol）均演示了自定义协议的注册方式：

from agentmesh.governance.protocol_facets import default_registry def extract_redis_facets(redis_ctx: dict) -> dict: cmd = (redis_ctx.get("command") or "").upper() return {"verb": cmd, "key": redis_ctx.get("key", "")} default_registry.register("redis", extract_redis_facets)

之后在评估上下文中传入{"redis": {"command": "FLUSHALL"}}，即可编写如下规则：

- name: deny-redis-flush condition: {field: "redis.verb", operator: in, value: ["FLUSHALL", "FLUSHDB"]} action: deny

值得注意的几个设计细节（均有测试佐证）：

• 提取器接收的是子字典（redis_ctx），返回的字段会update回该子字典（见 FacetRegistry.extract）；
• 若上下文键的值不是字典（例如"sql": "not-a-dict"），提取器会被跳过（test_non_dict_context_key_skipped）；
• 提取器抛出的异常会被捕获并记录日志，不会阻断策略评估（test_extractor_exception_is_swallowed）；
• 多个提取器按注册顺序依次执行（test_multiple_extractors_all_run_in_order）；
• 也可以构造独立FacetRegistry并通过extract_protocol_facets(ctx, registry=...)传入，实现不同调用方使用不同解析集合（test_custom_registry_passed_to_extract_protocol_facets）。

完整的规则示例参见 examples/policy-templates/wire-protocol-rules.yaml（Python SDK 的agent_control_specification_version: 0.4.0-alpha.1格式策略模板）。

多语言一致性：同一套面模型跨 SDK 可用

同一面模型在多个语言 SDK 中保持一致：每个 SDK 都暴露FacetRegistry、默认注册表以及extract_protocol_facets等价辅助函数，内置sql.*与k8s.*提取器且字段名一致，并在策略求值内部自动运行提取器。

说明：原文档以 GitHub issue 编号（#2553、#2587 等）标注各语言状态，本仓库中 TypeScript 的实现在 agent-governance-typescript/src/protocol-facets.ts 及其测试 tests/protocol-facets.test.ts，并通过 src/index.ts 对外导出；Go 与 .NET 的规则示例文件分别位于 agent-governance-golang/examples/wire-protocol-rules.yaml 与 agent-governance-dotnet/examples/Quickstart/wire-protocol-rules.yaml。

Rust 用法

Rust SDK 在agentmesh::protocol_facets模块下提供同样的面模型：FacetRegistry、default_registry()、extract_protocol_facets、extract_sql_facets、extract_k8s_facets，并暴露相同的sql.*与k8s.*字段。PolicyEngine::evaluate会在调用方上下文的防御性副本上运行注册表，再匹配规则，因此不会修改调用方数据。

use agentmesh::{PolicyEngine, default_registry}; use serde_yaml::Value; use std::collections::HashMap; let engine = PolicyEngine::new(); engine.load_from_yaml(r#" version: "1" agent: "did:example:agent1" policies: - name: deny-destructive-sql type: capability denied_actions: ["*"] conditions: sql.verb: [DROP, TRUNCATE, DELETE] "#).unwrap(); let mut sub = serde_yaml::Mapping::new(); sub.insert(Value::String("query".into()), Value::String("DROP TABLE production".into())); let mut ctx = HashMap::new(); ctx.insert("sql".to_string(), Value::Mapping(sub)); let decision = engine.evaluate("db.exec", Some(&ctx)); // decision == PolicyDecision::Deny(...) // Register a custom protocol extractor: default_registry().register("redis", |sub| { let mut m = std::collections::HashMap::new(); if let Some(cmd) = sub.get(Value::String("command".into())).and_then(|v| v.as_str()) { m.insert("verb".to_string(), Value::String(cmd.to_uppercase())); } m });

Rust SDK 的规则条件沿用现有 YAML 映射形态（key: value或key: [v1, v2]表示in式成员判断），字段名与决策结果与 Python 实现完全一致。完整的 Rust 规则示例见 agent-governance-rust/agentmesh/examples/wire-protocol-rules.yaml。

SQL 解析器说明：Rust 提取器使用内置的正则分词器，覆盖策略规则常用的动词/目标/函数场景。对于复杂的方言级 SQL，请通过default_registry().register("sql", ...)注册自定义提取器。

.NET 用法

.NET SDK 在AgentGovernance.Policy命名空间下暴露同样的面模型：FacetRegistry、ProtocolFacets.DefaultRegistry、ProtocolFacets.ExtractProtocolFacets、ProtocolFacets.ExtractSqlFacets、ProtocolFacets.ExtractK8sFacets。PolicyEngine.Evaluate在内部上下文副本上运行注册表，调用方只需填充原始sql/k8s子字典——调用方自己的字典不会被修改。

using AgentGovernance.Policy; var engine = new PolicyEngine(); engine.LoadYaml(@" apiVersion: governance.toolkit/v1 name: sql-guard scope: global default_action: allow rules: - name: deny-destructive-sql condition: ""sql.verb == 'DROP'"" action: deny priority: 100 "); var decision = engine.Evaluate("did:mesh:agent1", new Dictionary<string, object> { ["sql"] = new Dictionary<string, object> { ["query"] = "DROP TABLE production" }, }); // decision.Allowed == false, decision.MatchedRule == "deny-destructive-sql" // Register a custom protocol extractor: ProtocolFacets.DefaultRegistry.Register("redis", sub => { var cmd = sub.TryGetValue("command", out var v) ? v?.ToString() ?? "" : ""; return new Dictionary<string, object> { ["verb"] = cmd.ToUpperInvariant() }; });

.NET 的规则条件使用表达式字符串格式（"sql.verb == 'DROP'"、点路径字段引用、and/or组合），字段名与高层决策结果与 Python 实现一致；存在少量规则语法差异——尤其是 .NET 的in运算符引用的是列表值上下文字段而非 YAML 字面量列表，因此多动词规则建议拆分为单独的==检查（参见 agent-governance-dotnet/examples/Quickstart/wire-protocol-rules.yaml 中的拆分写法，如deny-destructive-sql-drop、deny-destructive-sql-truncate、deny-destructive-sql-delete各自独立成条）。

SQL 解析器说明：.NET 提取器同样使用内置正则分词器，并非完整的 SQL 解析器，无法覆盖所有方言构造；在高保障环境中，建议通过ProtocolFacets.DefaultRegistry.Register("sql", ...)注册基于真正 SQL 解析器的自定义提取器。

实战建议与注意事项

1. fail-closed 语义：SQL 解析在缺少sqlglot或解析失败时返回UNKNOWN动词。治理生产环境的 Agent 时，应把UNKNOWN纳入 deny 规则或依赖default_action: deny，避免“解析不出就放行”的漏洞。
2. 复合条件用多条规则表达：Python 的{field, operator, value}每条规则匹配单个字段，动词 + 目标的复合检查应拆分为多条规则并用优先级排序；Rust 用conditions映射、.NET 用表达式字符串，语法形态各有差异，跨语言迁移时留意 各语言规则示例 的差异。
3. 面提取不会污染调用方数据：Python 端extract_protocol_facets就地更新上下文并返回同一字典对象（test_returns_same_dict），Rust 与 .NET 则在内部副本上运行——调用方无需担心原始上下文被修改或重复提取产生副作用。
4. 代理层自动生效：通过agentmesh proxy透传的 MCP 工具调用，只要参数命名符合query/sql与method/path（/api/、/apis/前缀）约定，规则即可零改造生效；自定义协议可遵循同样的参数命名约定并注册对应提取器。

总结

Wire-Protocol-Aware Policy Evaluation 通过协议面机制，把“Agent 到底要做什么”从 HTTP 元数据下沉到线级语义：SQL 侧覆盖SELECT到DROP的完整动词谱系与表/函数维度，K8s 侧覆盖具名/集合、命名空间、子资源与动词映射。配合 MCP 透明代理的自动填充与default_registry的可扩展注册，治理团队可以用一套 YAML 规则，在 Python、Rust、TypeScript、.NET、Go 各 SDK 间保持一致的策略表达，将破坏性 SQL、生产命名空间写入、exec子资源等高危操作在规则求值层直接阻断。

【免费下载链接】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