"""对齐 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()