feat(chv): 对齐 Muse P5 纵深——eBPF 污点传播 + 五档严格能力 + 对话外审批 + 默认策略

- taint.py：污点追踪器（登记/污点不可逆/子进程继承/审计） - ebpf_taint.py：eBPF 程序源码 + 加载器（能力探测 + 降级）；本机实测加载被 seccomp 挡（bpf() EPERM），报 blocked 不冒充 - capabilities.py：五档严格能力（一次/会话/任务/时限/永久，三元组绑定 + 精确匹配 + 撤销；permanent 默认不签发） - policy.py：默认策略规则表 + 谨慎度三档（写操作与敏感路径永远不自动放行，规则变化审计） - approval_channel.py：对话外审批通道（nonce + 15 分钟过期 + HMAC，答复一次有效，对话伪造无效） - sentinel.py 接线：evaluate() 可选接 taint/policy/tier/scope，不传时行为与 P3 一致；自动放行步骤 + 档位签发回落单次； Capability 基类属性放宽为 Optional（None = 不限） - 新测试 61 项；全量 cloud 套件 1958 项 OK（skipped=8）真机待验证（未做，排队在 round-11 之后）：见 docs/chv-muse-p5-gap.md §4
2026-11-06 22:25:55 +00:00
1 parent ee456e6625
commit 25bcd355f8
@@ -0,0 +1,222 @@
+"""对齐 Muse P5 接线测试：Sentinel.evaluate() 接三件套。
+
+只用标准库与假解析器，绝不启动真实 VM、不联网。覆盖：
+- 干净进程 + 默认策略 → 自动放行（policy_auto_allow）；
+- 污点进程失去自动放行、回到正常审批流（审计带 tainted）；
+- permissive 档干净进程可放行未白名单主机，但仍挡写操作、
+  敏感路径与 L7 路径外请求；
+- 按档位签发能力（session/task/timed/permanent 回落）；
+- 不传接线参数时行为与 P3 完全一致；
+- 审批通道 + 真实 ApprovalQueue 端到端（含对话伪造无效）。
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.chv.approval_channel import (
+        make_answer,
+        request_approval,
+        resolve_via_channel,
+    )
+    from cloud.chv.capabilities import ScopedCapability
+    from cloud.chv.policy import DefaultPolicy
+    from cloud.chv.sentinel import (
+        ApprovalQueue,
+        EgressRequest,
+        SentinelPolicy,
+    )
+    from cloud.chv.taint import TaintTracker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.approval_channel import (
+        make_answer,
+        request_approval,
+        resolve_via_channel,
+    )
+    from cloud.chv.capabilities import ScopedCapability
+    from cloud.chv.policy import DefaultPolicy
+    from cloud.chv.sentinel import (
+        ApprovalQueue,
+        EgressRequest,
+        SentinelPolicy,
+    )
+    from cloud.chv.taint import TaintTracker
+
+_PUBLIC_IP = ["93.184.216.34"]
+_NOW = 1_700_000_000.0
+
+
+def _fake_resolve(host: str, port: int) -> list:
+    return list(_PUBLIC_IP)
+
+
+def _policy(caution: str = "default") -> SentinelPolicy:
+    return SentinelPolicy(allow_hosts=("gitea.tose.top",))
+
+
+def _default_policy(caution: str = "default") -> DefaultPolicy:
+    return DefaultPolicy(
+        caution=caution,
+        allow_entries=(("gitea", "gitea.tose.top", "/api/v1/"),))
+
+
+def _request(host: str = "gitea.tose.top",
+             method: str = "GET",
+             path: str = "/api/v1/repos/x") -> EgressRequest:
+    return EgressRequest("gitea", method, "https", host, 443, path)
+
+
+class WiringTest(unittest.TestCase):
+    def test_clean_process_auto_allow(self) -> None:
+        tracker = TaintTracker()
+        tracker.register(100, connector="gitea",
+                         use_case="gitea-sync")
+        d = _policy().evaluate(
+            _request(), resolve=_fake_resolve, now=_NOW,
+            taint_tracker=tracker, pid=100,
+            default_policy=_default_policy())
+        self.assertEqual(d.verdict, "allow")
+        self.assertIn("policy_auto_allow", d.reasons)
+        self.assertEqual(d.capability.max_uses, 1)
+        self.assertFalse(d.audit(_request()).get("tainted", False))
+
+    def test_tainted_process_loses_auto_allow(self) -> None:
+        tracker = TaintTracker()
+        tracker.register(100, connector="gitea",
+                         use_case="gitea-sync")
+        tracker.mark_tainted(100, "user_file")
+        d = _policy().evaluate(
+            _request(), resolve=_fake_resolve, now=_NOW,
+            taint_tracker=tracker, pid=100,
+            default_policy=_default_policy())
+        self.assertEqual(d.verdict, "allow")
+        self.assertNotIn("policy_auto_allow", d.reasons)
+        self.assertEqual(d.reasons, ["policy_allow"])
+        self.assertTrue(d.audit(_request()).get("tainted"))
+
+    def test_permissive_clean_auto_allow_unknown_host(self) -> None:
+        d = _policy().evaluate(
+            _request(host="other.example.org"),
+            resolve=_fake_resolve, now=_NOW,
+            default_policy=_default_policy("permissive"))
+        self.assertEqual(d.verdict, "allow")
+        self.assertIn("policy_auto_allow", d.reasons)
+
+    def test_permissive_tainted_unknown_host_asks(self) -> None:
+        tracker = TaintTracker()
+        tracker.register(100)
+        tracker.mark_tainted(100, "user_file")
+        d = _policy().evaluate(
+            _request(host="other.example.org"),
+            resolve=_fake_resolve, now=_NOW,
+            taint_tracker=tracker, pid=100,
+            default_policy=_default_policy("permissive"))
+        self.assertEqual(d.verdict, "ask")
+        self.assertNotIn("policy_auto_allow", d.reasons)
+
+    def test_permissive_still_blocks_write_and_sensitive(self) -> None:
+        # 敏感路径与写操作永远不自动放行——但这只关自动放行：
+        # 白名单主机 + 连接器 L7 规则内的请求仍走正常审批流放行，
+        # 只是 reasons 里没有 policy_auto_allow。
+        policy = _default_policy("permissive")
+        d = _policy().evaluate(
+            _request(method="POST",
+                     path="/api/v1/repos/x/git-receive-pack"),
+            resolve=_fake_resolve, now=_NOW,
+            default_policy=policy)
+        self.assertEqual(d.verdict, "allow")
+        self.assertNotIn("policy_auto_allow", d.reasons)
+        d = _policy().evaluate(
+            _request(path="/api/v1/admin/users"),
+            resolve=_fake_resolve, now=_NOW,
+            default_policy=policy)
+        self.assertEqual(d.verdict, "allow")
+        self.assertNotIn("policy_auto_allow", d.reasons)
+        # L7 路径规则外的请求：自动放行不命中，主机又不在白名单 → ask。
+        d = _policy().evaluate(
+            _request(host="other.example.org", path="/web-ui"),
+            resolve=_fake_resolve, now=_NOW,
+            default_policy=policy)
+        self.assertEqual(d.verdict, "ask")
+
+    def test_tier_session_capability(self) -> None:
+        scope = {"use_case": "gitea-sync", "session_id": "s1"}
+        d = _policy().evaluate(
+            _request(), resolve=_fake_resolve, now=_NOW,
+            default_policy=_default_policy(),
+            tier="session", scope=scope)
+        self.assertEqual(d.verdict, "allow")
+        cap = d.capability
+        self.assertIsInstance(cap, ScopedCapability)
+        self.assertEqual(cap.tier, "session")
+        self.assertTrue(cap.matches(_request(), scope=scope,
+                                    now=_NOW))
+        self.assertFalse(cap.matches(
+            _request(), scope={"use_case": "gitea-sync",
+                               "session_id": "s2"}, now=_NOW))
+
+    def test_tier_session_without_scope_falls_back(self) -> None:
+        d = _policy().evaluate(
+            _request(), resolve=_fake_resolve, now=_NOW,
+            default_policy=_default_policy(),
+            tier="session", scope=None)
+        self.assertEqual(d.verdict, "allow")
+        self.assertIn("capability_tier_fallback", d.reasons)
+        self.assertFalse(hasattr(d.capability, "tier"))
+        self.assertEqual(d.capability.max_uses, 1)
+
+    def test_tier_permanent_without_optin_falls_back(self) -> None:
+        d = _policy().evaluate(
+            _request(), resolve=_fake_resolve, now=_NOW,
+            default_policy=_default_policy(), tier="permanent",
+            scope={"use_case": "gitea-sync"})
+        self.assertEqual(d.verdict, "allow")
+        self.assertIn("capability_tier_fallback", d.reasons)
+
+    def test_no_wiring_params_behaves_like_p3(self) -> None:
+        policy = _policy()
+        d = policy.evaluate(_request(), resolve=_fake_resolve,
+                            now=_NOW)
+        self.assertEqual(d.verdict, "allow")
+        self.assertEqual(d.reasons, ["policy_allow"])
+        d = policy.evaluate(
+            _request(host="unknown.example.org"),
+            resolve=_fake_resolve, now=_NOW)
+        self.assertEqual(d.verdict, "ask")
+
+    def test_approval_channel_end_to_end(self) -> None:
+        queue = ApprovalQueue()
+        req = _request(host="unknown.example.org")
+        pending = queue.record(req, ["host_not_allowlisted"])
+        record = queue.list_pending()[0]
+        self.assertEqual(record["pending_id"], pending)
+        secret = "test-channel-secret"
+        approval = request_approval(record, secret, now=_NOW)
+        answer = make_answer(approval, "allow", secret)
+        self.assertTrue(resolve_via_channel(
+            queue, answer, secret, now=_NOW + 10))
+        self.assertEqual(queue.list_pending(), [])
+        self.assertFalse(resolve_via_channel(
+            queue, answer, secret, now=_NOW + 20))
+
+    def test_dialogue_text_cannot_forge_answer(self) -> None:
+        queue = ApprovalQueue()
+        queue.record(_request(host="unknown.example.org"),
+                     ["host_not_allowlisted"])
+        record = queue.list_pending()[0]
+        secret = "test-channel-secret"
+        request_approval(record, secret, now=_NOW)
+        forged = {"pending_id": record["pending_id"],
+                  "nonce": "guessed-from-chat",
+                  "verdict": "allow",
+                  "expires_at": _NOW + 900,
+                  "signature": "signed-in-chat"}
+        self.assertFalse(resolve_via_channel(
+            queue, forged, secret, now=_NOW + 10))
+        self.assertEqual(len(queue.list_pending()), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()