Agent Governance Toolkit 教程用 Attribute Ratchets 构建会话级 DLP 数据防泄漏系统【免费下载链接】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本文基于 docs/tutorials/39-dlp-attribute-ratchets.md 展开。你将学习如何利用 Agent Governance ToolkitPythonagentmesh包中的SessionState/SessionAttribute让 Agent 的权限在触碰敏感数据后自动收紧、且在整个会话内只能单向升级monotonic ratchet从而在策略引擎中实现真正具备会话记忆的 Data Loss PreventionDLP。读完本文你将掌握会话属性的定义、DLP 策略编写、会话模拟与重置并理解底层棘轮机制的源码实现。传统的策略引擎无状态每一次工具调用都是独立评估的Agent 先读了机密文档、再尝试把内容发邮件两次调用分别看都是允许的——策略根本不知道前一次读取发生过。Attribute Ratchets属性棘轮通过会话级单调递增属性解决这个问题一旦 Agent 接触了 confidential 数据会话的data_sensitivity就永久锁定在 confidential之后的邮件外发、文件导出等动作会被策略引擎直接拒绝且 Agent 无法在会话内忘记或重置这一状态。1. 核心问题无会话状态的 DLP 漏洞在没有会话状态时策略对每次工具调用独立求值形成典型的 DLP 绕过路径Agent reads confidential document → ✅ allowed Agent sends email with that content → ✅ allowed (policy doesnt know about the read!)引入属性棘轮之后同一场景被彻底逆转Agent reads confidential document → ✅ allowed, sensitivity ratchets to confidential Agent tries to send email externally → ❌ blocked (session.data_sensitivity confidential) Agent tries to reset sensitivity → ❌ ignored (monotonic — can only go up)这正是读到机密 → 立刻进入戒备状态的会话内数据防泄漏语义。该能力由 session_state.py 中的SessionState与SessionAttribute实现并通过init.py 作为agentmesh.governance的公开 API 导出。2. Step 1定义会话属性SessionAttribute描述一个会话级属性的四个要素字段类型说明namestr属性名策略条件中以session.name引用orderinglist[str]有序取值列表低 → 高如[public, internal, confidential, restricted]monotonicbool为True时取值只能沿ordering向前移动默认Falseinitialstr \| None初始值不传时默认取ordering的第一个元素从源码看session_state.pySessionAttribute是 dataclass__post_init__会在initial为空且ordering非空时自动把ordering[0]设为初始值——这保证了棘轮语义有明确的起点。例如data_jurisdiction的domestic → cross_border映射正好对应 GDPR / 数据驻留合规场景。from agentmesh.governance import SessionState, SessionAttribute state SessionState([ SessionAttribute( namedata_sensitivity, ordering[public, internal, confidential, restricted], monotonicTrue, initialpublic, ), SessionAttribute( namedata_jurisdiction, ordering[domestic, eu, cross_border], monotonicTrue, initialdomestic, ), ]) print(fInitial sensitivity: {state.get(data_sensitivity)}) # → publicSessionState构造时会把每个定义了initial的属性写入内部_values字典get(name)返回当前值get_all()返回全部快照。注意测试 test_session_state.py 还验证了一个细节不传ordering的属性initial为None即未初始化的属性不会出现在状态中。3. Step 2创建 DLP 策略会话属性只是事实真正的拦截动作由声明式策略文件驱动。以下dlp-policy.yaml完全继承自原教程策略 Schema 版本为governance.toolkit/v1当前版本见 policy.py# dlp-policy.yaml apiVersion: governance.toolkit/v1 name: dlp-policy agents: [*] default_action: allow rules: # Block email when handling sensitive data - name: block-email-sensitive stage: pre_tool condition: action.type send_email and session.data_sensitivity in [confidential, restricted] action: deny description: Cannot send emails after accessing confidential data priority: 900 # Block file export for restricted data - name: block-export-restricted stage: pre_tool condition: action.type export and session.data_sensitivity restricted action: deny description: Restricted data cannot be exported priority: 1000 # Require approval for cross-border transfers - name: approve-cross-border stage: pre_tool condition: action.type transfer and session.data_jurisdiction cross_border action: require_approval approvers: [data-protection-officer] priority: 800 # Allow read operations (but they may ratchet sensitivity) - name: allow-read stage: pre_tool condition: action.type read action: allow priority: 100逐条解读结合 policy.py 的PolicyRule模型stage: pre_tool规则在工具调用前这一生命周期阶段执行。PolicyRule.stage取值包括pre_input/pre_tool默认/post_tool/pre_outputcondition支持、!、in [...]、contains、startswith、endswith、数值比较以及and/or复合表达式session.data_sensitivity即注入的会话属性actionallow/deny/warn/require_approval/log。require_approval配合approvers字段指定审批人本策略中为data-protection-officer对应仓库中的审批工作流参见 approval.pypriority数值越大越先求值。block-export-restricted1000block-email-sensitive900approve-cross-border800allow-read100default_action: allow无规则匹配时的兜底动作可选allow/deny。需要说明的是当引擎未加载任何策略时policy.py 会按 fail-closed 原则默认deny。4. Step 3模拟一个完整的 Agent 会话把状态与策略引擎接起来的关键是SessionState.inject_context(context)它把当前会话属性以session键注入求值上下文源码见 session_state.py随后PolicyEngine.evaluate()中的每条规则都能读到session.data_sensitivity。from agentmesh.governance import PolicyEngine, SessionState, SessionAttribute engine PolicyEngine(conflict_strategydeny_overrides) engine.load_yaml_file(dlp-policy.yaml) state SessionState([ SessionAttribute( namedata_sensitivity, ordering[public, internal, confidential, restricted], monotonicTrue, ), ]) # ── Turn 1: Agent reads a public document ────────────────── ctx1 {action: {type: read}, resource: {type: document, classification: public}} state.inject_context(ctx1) result1 engine.evaluate(*, ctx1) print(f1. Read public doc: {result1.action}) # → allow # Simulate: tool returns classification public (no ratchet) # ── Turn 2: Agent reads a confidential report ────────────── ctx2 {action: {type: read}, resource: {type: document, classification: confidential}} state.inject_context(ctx2) result2 engine.evaluate(*, ctx2) print(f2. Read confidential report: {result2.action}) # → allow # Simulate: tool reports this document is confidential state.set(data_sensitivity, confidential) print(f → Sensitivity ratcheted to: {state.get(data_sensitivity)}) # ── Turn 3: Agent tries to email the content ─────────────── ctx3 {action: {type: send_email}} state.inject_context(ctx3) result3 engine.evaluate(*, ctx3) print(f3. Send email: {result3.action}) # → DENY! print(f Rule: {result3.matched_rule}) # → block-email-sensitive # ── Turn 4: Agent tries to forget the sensitivity ──────── reset_ok state.set(data_sensitivity, public) print(f4. Reset sensitivity: {reset_ok}) # → False (monotonic!) print(f Still: {state.get(data_sensitivity)}) # → confidential # ── Turn 5: Sensitivity can still go UP ──────────────────── state.set(data_sensitivity, restricted) print(f5. Ratcheted to: {state.get(data_sensitivity)}) # → restricted运行输出1. Read public doc: allow 2. Read confidential report: allow → Sensitivity ratcheted to: confidential 3. Send email: deny Rule: block-email-sensitive 4. Reset sensitivity: False Still: confidential 5. Ratcheted to: restricted值得注意的细节conflict_strategydeny_overrides当多条规则同时匹配时任何 deny 都会胜出。引擎支持priority_first_match默认保持 v1.0 行为、deny_overrides、allow_overrides、most_specific_wins四种策略policy.pyevaluate(*, ctx)第一个参数是 agent DID*通配符命中策略中的agents: [*]第二个参数是被注入会话状态的上下文默认在pre_tool阶段求值result.matched_rulePolicyDecision携带命中规则名block-email-sensitive便于审计与调试。完整字段还包括allowed、action、policy_name、reason、approvers、rate_limited、evaluation_ms等policy.py。5. Step 4从策略 YAML 解析会话属性属性定义不必与策略分离——可以直接把session_attributes写进策略 YAML通过SessionState.from_policy_yaml()一次解析。该工厂方法在源码 session_state.py 中实现yaml.safe_load后逐个构建SessionAttribute并兼容缺省字段ordering缺省为空列表、monotonic缺省为False。state SessionState.from_policy_yaml( session_attributes: - name: data_sensitivity ordering: [public, internal, confidential, restricted] monotonic: true initial: public - name: user_verified ordering: [unverified, email_verified, mfa_verified] monotonic: true initial: unverified )user_verified是渐进式认证的典型用例Agent 尚未验证时处于unverified完成 MFA 后升到mfa_verified且永远不会降级——与 DLP 灵敏度棘轮同理。对应测试 test_session_state.py 还覆盖了空 YAML 与缺少session_attributes键的容错场景均返回空状态。6. Step 5会话之间重置棘轮是会话内的承诺不是永久性的。会话结束如换一个用户时调用reset()所有属性回到各自的initial值# End of session — reset for next user state.reset() print(state.get(data_sensitivity)) # → public (back to initial)源码实现session_state.py会遍历属性定义有initial的恢复初值无initial的从状态中移除。测试 test_session_state.py 验证了 high → reset → low 的完整闭环。7. 源码级原理monotonic 棘轮如何工作SessionState.set()是棘轮语义的核心session_state.py其逻辑可概括为若属性定义了monotonicTrue且ordering非空则先在ordering中定位当前值与新值的下标新值不在ordering中记录 warning 并返回False拒绝新值下标 ≤ 当前值下标含写回同一值返回False状态不变只有新值下标严格大于当前值时才更新_values并返回True。测试 test_session_state.py 明确覆盖了同值拒绝与未知值拒绝两个容易被忽略的边界set(level, med)两次中的第二次返回Falseset(level, unknown)返回False且状态保持low。非单调属性monotonicFalse则可任意双向移动见test_non_monotonic_allows_any_direction。从策略引擎侧看规则条件求值还具备fail-closed语义policy.py当条件表达式抛异常时非allow规则按匹配处理即 deny 生效allow规则按不匹配处理——防止攻击者通过构造畸形输入绕过拦截。这一设计与棘轮的只能收紧、不能放松哲学一脉相承。端到端联动已被仓库测试验证TestIntegrationWithPolicyContext.test_ratchet_blocks_export_after_sensitive_readtest_session_state.py模拟读取机密 → 导出被拒 → 尝试重置灵敏度失败 → 导出依然被拒的完整链路与本文 Step 3 的会话模拟完全一致。8. 与多阶段流水线结合Tutorial 37 的延伸棘轮通常与 多阶段策略流水线 配合使用形成闭环post_tool阶段工具返回后根据结果上调会话灵敏度state.set(data_sensitivity, confidential)pre_tool阶段下一次工具调用前依据session.data_sensitivity执行拦截。即原教程所述的post_tool sets sensitivity, pre_tool enforces it。流水线四个阶段pre_input/pre_tool/post_tool/pre_output分别负责注入检测、动作授权、输出分类与响应泄密检查属性棘轮让后三个阶段之间具备了时间上的因果记忆。策略组合方面父策略与子策略可通过extends做 additive-only 合并——父策略的 deny 规则不可被子策略削弱policy.py这意味着 CISO 定义的 DLP 红线不会因为应用团队扩展策略而失效。9. 真实世界的 DLP 模式原教程总结的通用属性模式可直接复用AttributeOrderingUse Casedata_sensitivitypublic → restrictedDocument classification ratchetdata_jurisdictiondomestic → cross_borderGDPR/data residencyauth_levelanonymous → mfa_verifiedProgressive authenticationrisk_scorelow → criticalCumulative risk escalationcompliance_statusclean → flagged → blockedCompliance state machine这些模式覆盖了文档分级收紧数据驻留合规渐进式认证累积风险升级合规状态机五类典型治理诉求。如需更细粒度的数据分类判定还可参考 41-advisory-defense-in-depth.md 中的 advisory 层advisory.py 提供PatternAdvisory/CompositeAdvisory/CallbackAdvisory等检查器用 ML 或规则分类结果驱动棘轮上调。10. 下一步Tutorial 37 — 多阶段策略流水线37-multi-stage-pipeline.md掌握post_tool设置灵敏度、pre_tool强制执行的四阶段闭环Tutorial 41 — Advisory 纵深防御41-advisory-defense-in-depth.md将棘轮与基于 ML 的分类 advisory 层结合。核心参考实现与测试路径会话状态实现session_state.py会话状态测试test_session_state.py策略引擎实现规则模型、条件求值、冲突策略、fail-closedpolicy.py公共 API 导出init.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），仅供参考