"""对齐 Muse P12 测试：审计轨迹与活动视图。 只用标准库，不启动真实 VM，不联网，不碰生产文件： 审计轨迹、计划登记、权限登记一律落在临时目录。 覆盖：清洗白名单、哈希链验签、项目隔离、旧行迁移、 导出、计划流转、授权有效期/撤销/次数、审批卡风险与 通道一致性——含否定测试。 """ import json import os import sys import tempfile import unittest from pathlib import Path try: from cloud.chv.activity import ( PermissionRegistry, PlanStore, activity_summary, ) from cloud.chv.approval_card import ( build_approval_card, card_matches_channel, risk_level, ) from cloud.chv.audit_trail import ( AUDIT_CATEGORIES, AuditTrail, from_legacy_row, sanitize_detail, ) except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.chv.activity import ( PermissionRegistry, PlanStore, activity_summary, ) from cloud.chv.approval_card import ( build_approval_card, card_matches_channel, risk_level, ) from cloud.chv.audit_trail import ( AUDIT_CATEGORIES, AuditTrail, from_legacy_row, sanitize_detail, ) PROJECT = "proj-p12" OTHER = "proj-other" def _capability(cap_id="cap-1", tier="task", uses_left=3, expires_at=None, revoked=False): """P5 ScopedCapability.describe() 形状的 dict。""" return { "capability_id": cap_id, "tier": tier, "connector": "gitea", "host": "gitea.tose.top", "use_case": "sync", "session_id": "s1", "task_id": "t1", "methods": ["GET", "POST"], "max_uses": 3, "uses_left": uses_left, "expires_at": expires_at, "revoked": revoked, } def _queue_record(**overrides): """P3 ApprovalQueue.record() 形状的 dict。""" record = { "pending_id": "pend-abc123", "connector": "gitea", "method": "GET", "scheme": "https", "host": "gitea.tose.top", "port": 443, "path": "/api/v1/repos", "reasons": ["host_not_allowlisted"], "ts": 1000.0, } record.update(overrides) return record class SanitizeTest(unittest.TestCase): def test_secret_keys_dropped(self): cleaned = sanitize_detail({ "action": "sync", "gitea_token": "gto_abcdef", "api_key": "sk-123", "password": "hunter2", "nested": {"secret": "x", "repo": "r"}, }) self.assertEqual(cleaned, {"action": "sync", "nested": {"repo": "r"}}) def test_secret_value_in_string_redacted(self): cleaned = sanitize_detail( {"detail": "failed with token=abc123xyz and ok"}) self.assertEqual(cleaned["detail"], "failed with token=*** and ok") def test_benign_fields_kept(self): cleaned = sanitize_detail( {"repo": "admin/chv-sync-test", "count": 3, "tags": ["a", "b"]}) self.assertEqual(cleaned, {"repo": "admin/chv-sync-test", "count": 3, "tags": ["a", "b"]}) def test_empty_and_none_detail(self): self.assertEqual(sanitize_detail({}), {}) self.assertEqual(sanitize_detail(None), {}) class AuditTrailTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = os.path.join(self.tmp.name, "audit.jsonl") self.trail = AuditTrail(self.path) def tearDown(self): self.tmp.cleanup() def test_append_and_list_roundtrip(self): rec = self.trail.append( PROJECT, "worker", "vm_lifecycle", "vm_boot", object_type="vm", object_id="chv-1", ts=1000.0, detail={"image": "lite10"}) self.assertEqual(rec["seq"], 1) self.assertEqual(rec["prev_hash"], "0" * 64) self.assertEqual(len(rec["hash"]), 64) rows = self.trail.list_records(PROJECT) self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["action"], "vm_boot") def test_append_sanitizes_detail(self): self.trail.append( PROJECT, "worker", "approval", "ask", detail={"token": "gto_secret", "host": "example.invalid"}, ts=1000.0) rows = self.trail.list_records(PROJECT) self.assertNotIn("token", rows[0]["detail"]) self.assertEqual(rows[0]["detail"]["host"], "example.invalid") def test_unknown_category_rejected(self): with self.assertRaises(ValueError): self.trail.append(PROJECT, "worker", "nope", "x") def test_empty_project_rejected(self): with self.assertRaises(ValueError): self.trail.append("", "worker", "approval", "x") def test_project_isolation(self): self.trail.append(PROJECT, "worker", "approval", "ask", ts=1000.0) self.trail.append(OTHER, "worker", "approval", "ask", ts=1001.0) self.assertEqual(len(self.trail.list_records(PROJECT)), 1) self.assertEqual(len(self.trail.list_records(OTHER)), 1) self.assertEqual(self.trail.list_records(""), []) self.assertEqual(self.trail.list_records("proj-missing"), []) def test_list_filters_and_pagination(self): for index in range(5): self.trail.append( PROJECT, "worker", "vm_lifecycle", "vm_boot", task_id="t-%d" % (index % 2), ts=1000.0 + index) self.assertEqual(len(self.trail.list_records(PROJECT, task_id="t-0")), 3) self.assertEqual( len(self.trail.list_records(PROJECT, since=1003.0)), 2) self.assertEqual( len(self.trail.list_records(PROJECT, limit=2)), 2) self.assertEqual( len(self.trail.list_records(PROJECT, limit=2, offset=2)), 2) rows = self.trail.list_records(PROJECT) self.assertGreater(rows[0]["ts"], rows[-1]["ts"]) def test_verify_passes_then_detects_tamper(self): for index in range(3): self.trail.append(PROJECT, "worker", "approval", "ask", ts=1000.0 + index) ok, bad = self.trail.verify() self.assertTrue(ok) self.assertEqual(bad, -1) with open(self.path, "r", encoding="utf-8") as f: lines = f.readlines() tampered = json.loads(lines[1]) tampered["action"] = "allow" lines[1] = json.dumps(tampered) + "\n" with open(self.path, "w", encoding="utf-8") as f: f.writelines(lines) ok, bad = self.trail.verify() self.assertFalse(ok) self.assertEqual(bad, 1) def test_verify_detects_deletion(self): for index in range(3): self.trail.append(PROJECT, "worker", "approval", "ask", ts=1000.0 + index) with open(self.path, "r", encoding="utf-8") as f: lines = f.readlines() with open(self.path, "w", encoding="utf-8") as f: f.writelines([lines[0], lines[2]]) ok, bad = self.trail.verify() self.assertFalse(ok) self.assertEqual(bad, 1) def test_reload_restores_chain(self): self.trail.append(PROJECT, "worker", "approval", "ask", ts=1000.0) second = AuditTrail(self.path) rec = second.append(PROJECT, "worker", "approval", "allow", ts=1001.0) self.assertEqual(rec["seq"], 2) ok, _ = second.verify() self.assertTrue(ok) def test_export_json(self): self.trail.append(PROJECT, "worker", "approval", "ask", detail={"token": "nope"}, ts=1000.0) dest = os.path.join(self.tmp.name, "export.json") count = self.trail.export_json(PROJECT, dest) self.assertEqual(count, 1) with open(dest, "r", encoding="utf-8") as f: rows = json.load(f) self.assertEqual(len(rows), 1) self.assertNotIn("token", rows[0]["detail"]) self.assertEqual(os.stat(dest).st_mode & 0o777, 0o600) self.assertEqual(self.trail.export_json(OTHER, dest), 0) def test_all_categories_accepted(self): for category in sorted(AUDIT_CATEGORIES): self.trail.append(PROJECT, "worker", category, "probe", ts=1000.0) self.assertEqual(len(self.trail.list_records(PROJECT, limit=0)), len(AUDIT_CATEGORIES)) def test_legacy_row_conversion(self): row = {"id": 7, "user_id": "system", "project_id": PROJECT, "action": "worker_exec_start", "detail": {"token": "gto_x", "task": "t1"}, "created_at": 1000.0} converted = from_legacy_row(row) self.assertTrue(converted["legacy"]) self.assertEqual(converted["action"], "worker_exec_start") self.assertEqual(converted["object_id"], "7") self.assertNotIn("token", converted["detail"]) self.assertEqual(converted["detail"]["task"], "t1") class PlanStoreTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = os.path.join(self.tmp.name, "plans.json") self.store = PlanStore(self.path) def tearDown(self): self.tmp.cleanup() def test_add_and_list(self): plan = self.store.add_plan(PROJECT, "建 round-11 镜像", task_id="t1", ts=1000.0) self.assertEqual(plan["status"], "pending") self.assertEqual(len(self.store.list_plans(PROJECT)), 1) def test_valid_transitions(self): plan = self.store.add_plan(PROJECT, "同步仓库", ts=1000.0) moved = self.store.set_status( PROJECT, plan["plan_id"], "running", ts=1001.0) self.assertEqual(moved["status"], "running") done = self.store.set_status( PROJECT, plan["plan_id"], "done", ts=1002.0) self.assertEqual(done["status"], "done") def test_illegal_transition_rejected(self): plan = self.store.add_plan(PROJECT, "跑测试", ts=1000.0) with self.assertRaises(ValueError): self.store.set_status(PROJECT, plan["plan_id"], "done") self.store.set_status(PROJECT, plan["plan_id"], "cancelled") with self.assertRaises(ValueError): self.store.set_status(PROJECT, plan["plan_id"], "running") def test_unknown_plan_rejected(self): with self.assertRaises(ValueError): self.store.set_status(PROJECT, "nope", "running") def test_empty_title_rejected(self): with self.assertRaises(ValueError): self.store.add_plan(PROJECT, "") def test_list_filter_by_status(self): first = self.store.add_plan(PROJECT, "甲", ts=1000.0) self.store.add_plan(PROJECT, "乙", ts=1001.0) self.store.set_status(PROJECT, first["plan_id"], "running") self.assertEqual( len(self.store.list_plans(PROJECT, "running")), 1) self.assertEqual( len(self.store.list_plans(PROJECT, "pending")), 1) def test_persistence_roundtrip(self): self.store.add_plan(PROJECT, "甲", ts=1000.0) second = PlanStore(self.path) self.assertEqual(len(second.list_plans(PROJECT)), 1) class PermissionRegistryTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = os.path.join(self.tmp.name, "grants.json") self.registry = PermissionRegistry(self.path) def tearDown(self): self.tmp.cleanup() def test_grant_and_list_active(self): self.registry.record_grant(PROJECT, _capability(), ts=1000.0) active = self.registry.list_active(PROJECT, now=1001.0) self.assertEqual(len(active), 1) self.assertEqual(active[0]["connector"], "gitea") self.assertEqual(active[0]["uses_left"], 3) def test_missing_field_rejected(self): bad = _capability() del bad["host"] with self.assertRaises(ValueError): self.registry.record_grant(PROJECT, bad) def test_expired_grant_not_active(self): self.registry.record_grant( PROJECT, _capability(expires_at=1000.0), ts=900.0) self.assertEqual(self.registry.list_active(PROJECT, now=1001.0), []) def test_revoked_grant_not_active(self): self.registry.record_grant(PROJECT, _capability(), ts=1000.0) self.assertTrue(self.registry.revoke(PROJECT, "cap-1")) self.assertFalse(self.registry.revoke(PROJECT, "cap-1")) self.assertEqual(self.registry.list_active(PROJECT, now=1001.0), []) self.assertEqual(len(self.registry.list_all(PROJECT)), 1) def test_uses_exhausted_not_active(self): self.registry.record_grant(PROJECT, _capability(uses_left=1), ts=1000.0) self.assertTrue(self.registry.note_use(PROJECT, "cap-1")) self.assertFalse(self.registry.note_use(PROJECT, "cap-1")) self.assertEqual(self.registry.list_active(PROJECT, now=1001.0), []) def test_unlimited_uses_never_exhaust(self): self.registry.record_grant( PROJECT, _capability(uses_left=None), ts=1000.0) for _ in range(5): self.assertTrue(self.registry.note_use(PROJECT, "cap-1")) self.assertEqual(len(self.registry.list_active(PROJECT)), 1) def test_unknown_capability_use_revoke_false(self): self.assertFalse(self.registry.note_use(PROJECT, "nope")) self.assertFalse(self.registry.revoke(PROJECT, "nope")) def test_regrant_updates(self): self.registry.record_grant(PROJECT, _capability(), ts=1000.0) self.registry.record_grant( PROJECT, _capability(uses_left=9), ts=1001.0) active = self.registry.list_active(PROJECT, now=1002.0) self.assertEqual(len(active), 1) self.assertEqual(active[0]["uses_left"], 9) class ActivitySummaryTest(unittest.TestCase): def test_summary_shape_and_counts(self): plans = [ {"title": "跑测试", "status": "running"}, {"title": "建镜像", "status": "pending"}, {"title": "同步", "status": "done"}, ] recent = [{"ts": 1000.0, "action": "vm_boot", "category": "vm_lifecycle", "result": "ok", "object_type": "vm", "object_id": "chv-1", "detail": {"token": "must-not-leak"}}] grants = [_capability()] summary = activity_summary(plans, recent, grants) self.assertEqual(summary["current"], "跑测试") self.assertEqual(summary["counts"], {"running": 1, "planned": 1, "done": 1, "granted": 1}) self.assertEqual(len(summary["recent"]), 1) self.assertNotIn("detail", summary["recent"][0]) def test_no_running_current_empty(self): summary = activity_summary([], [], []) self.assertEqual(summary["current"], "") self.assertEqual(summary["counts"]["granted"], 0) class ApprovalCardTest(unittest.TestCase): def test_read_get_low_risk(self): card = build_approval_card(_queue_record()) self.assertEqual(card["risk"], "low") self.assertEqual(card["action"], "gitea GET") self.assertEqual(card["expires_at"], 1900.0) self.assertEqual([a["id"] for a in card["actions"]], ["accept", "deny"]) self.assertEqual( [o["tier"] for o in card["grant_options"]], ["one-shot", "session", "task", "timed", "permanent"]) def test_permanent_always_opt_in(self): card = build_approval_card(_queue_record()) permanent = [o for o in card["grant_options"] if o["tier"] == "permanent"][0] self.assertTrue(permanent["requires_explicit_opt_in"]) others = [o for o in card["grant_options"] if o["tier"] != "permanent"] self.assertTrue(all( not o["requires_explicit_opt_in"] for o in others)) def test_write_to_admin_high_risk(self): card = build_approval_card(_queue_record( method="POST", path="/admin/settings")) self.assertEqual(card["risk"], "high") def test_write_to_private_host_high_risk(self): card = build_approval_card(_queue_record( method="DELETE", host="192.168.1.5", path="/api/v1/x")) self.assertEqual(card["risk"], "high") def test_write_public_medium_risk(self): card = build_approval_card(_queue_record( method="POST", path="/api/v1/repos")) self.assertEqual(card["risk"], "medium") def test_missing_pending_id_rejected(self): record = _queue_record() del record["pending_id"] with self.assertRaises(ValueError): build_approval_card(record) def test_tiers_filtered_to_sentinel_choice(self): card = build_approval_card(_queue_record(), tiers=["one-shot", "task", "bogus"]) self.assertEqual([o["tier"] for o in card["grant_options"]], ["one-shot", "task"]) def test_risk_level_function(self): self.assertEqual(risk_level(_queue_record()), "low") self.assertEqual(risk_level(_queue_record( method="PUT", path="/api/v1/x")), "medium") def test_card_matches_channel(self): card = build_approval_card(_queue_record()) payload = { "pending_id": "pend-abc123", "nonce": "deadbeef", "target": {"connector": "gitea", "method": "GET", "scheme": "https", "host": "gitea.tose.top", "port": 443, "path": "/api/v1/repos"}, "reasons": ["host_not_allowlisted"], "expires_at": 1900.0, } self.assertTrue(card_matches_channel(card, payload)) def test_card_mismatch_channel(self): card = build_approval_card(_queue_record()) payload = {"pending_id": "pend-other", "target": {}} self.assertFalse(card_matches_channel(card, payload)) payload = {"pending_id": "pend-abc123", "target": {"host": "evil.example.invalid"}} self.assertFalse(card_matches_channel(card, payload)) if __name__ == "__main__": unittest.main()