From 3978dbbbf2abc9663355836c1dad017476fe6d20 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 15:30:36 +0000 Subject: [PATCH] v7 mech-6: production-evidence fail-closed live wiring seam (no real reader, no endpoint wiring yet) - LiveEvidenceReader protocol: get_checks(repo, ref) / get_branch_protection(repo, branch), results carry status plus raw evidence; production code has no HTTP call, no token read, no real GitHub API call (AST-asserted). - verify_live_evidence: no reader / no repo-ref / reader failure / mismatch all verified=False; comparison shared with verify_gate_evidence via _compare_claimed_vs_live. - live_gate_decision: verified=False -> allowed=False, http_status=409, status=blocked; verified=True -> 200. - Endpoints _merge_task/_release_task NOT wired yet (no production reader exists; wiring None would block all merges/releases); real live verification still pending B6 real App install and B1 user-side items. (cherry picked from commit 7e8bfdee1f22c28329602314f3e166cde34eb6cf) --- cloud/gate_verification.py | 361 +++++++++++++++++++++--- cloud/tests/test_scorecard_mech_live.py | 311 ++++++++++++++++++++ 2 files changed, 639 insertions(+), 33 deletions(-) create mode 100644 cloud/tests/test_scorecard_mech_live.py diff --git a/cloud/gate_verification.py b/cloud/gate_verification.py index 30a4779..41062d1 100644 --- a/cloud/gate_verification.py +++ b/cloud/gate_verification.py @@ -56,8 +56,95 @@ mismatch）当作阻断，核对器本身不用改。 零第三方依赖，只用标准库。本模块不 import cloud 其它模块（reader 由调用方注入），不发网络请求。 + +v7 机制第 6 项（生产证据 fail-closed live 接线，本批新增， +只动本模块，不动其它 7 项）： +- 新增可注入 reader 接口 LiveEvidenceReader（协议）： + get_checks(repo, ref) 与 get_branch_protection(repo, + branch)，返回结构含状态与原始证据（LiveChecksResult / + LiveProtectionResult，或等价 dict）。本模块的生产代码 + 里绝不实现真实 HTTP 调用、绝不读真实 token、绝不调 + 用真实 GitHub API——接口是未来真实 reader 的插入缝， + 不是真实 reader 本身。 +- 新增 verify_live_evidence(claimed, reader, repo, ref, + branch=None)：协议 reader 或旧式 callable reader 均可； + 无 reader、无 repo/ref、reader 失败、reader 返回与声 + 称不一致时 verified 一律 False；声称 CI 全绿但实时 + 数据失败/缺失时记 mismatch。比对规则与 + verify_gate_evidence 逐项一致（共用 _compare），不另 + 造口径；空声称不由本函数放行——缺失证据由硬门禁 + （merge_gate_hard）阻断，本函数只核对“声称了的项”。 +- 新增 live_gate_decision：端点层判定函数，verified + 为 False 时返回 allowed=False、http_status=409、 + status="blocked"、blocked=True（mismatch 与未核对用 + 不同 code 区分：live_evidence_mismatch / + live_evidence_unverified）；verified 为 True 时返回 + allowed=True、http_status=200、status="allowed"。 +- 诚实残留（不得冒充已接线）：HTTP 端点 _merge_task / + _release_task 本批未接 live_gate_decision——当前无 + 生产 reader 可注入：以 reader=None 接线会挡住全部合 + 并/发布（fail-closed 的本意不是停摆），以桩 reader + 接线则是假求证。真实 live 验证仍待 B6 真实 App 安装 + 与 B1 等用户侧事项，未在本批完成。本批不改 #238 + auto_apply（仍 False）、不改 #247“纠正为 pass 本身 + 不解封”、不改 v1.1 契约、不改任何阈值（阈值仍 + test_default_unapproved）。 """ +from dataclasses import dataclass + +try: + from typing import Protocol + from typing import runtime_checkable +except ImportError: # pragma: no cover - Python < 3.8 + Protocol = object # type: ignore + + def runtime_checkable(cls): + return cls + + +def _compare_claimed_vs_live(evidence, live): + """声称证据与实时数据的逐项比对（两核对函数共用，同一口径）。 + + 只比对调用方声称的项；实时数据缺失某项时记 mismatch + （声称全绿却查无此项，不是“查不到所以通过”）。 + 返回 mismatches 列表（空即逐项全过）。 + """ + mismatches = [] + live_checks = live.get("ci_checks") or {} + if not isinstance(live_checks, dict): + live_checks = {} + evidence = evidence or {} + if evidence.get("ci_status") == "success": + failed_live = sorted( + name for name, status in live_checks.items() + if status != "success") + if failed_live: + mismatches.append( + "ci_not_green_live: %s" % ", ".join(failed_live)) + for check in (evidence.get("ci_checks") or []): + if not isinstance(check, dict): + continue + name = check.get("name") + if not name: + continue + if check.get("status") == "success" \ + and live_checks.get(name) != "success": + mismatches.append( + "check_not_green_live: %s" % name) + for name in (evidence.get("required_checks") or []): + if live_checks.get(name) != "success": + mismatches.append( + "required_check_not_green_live: %s" % name) + claimed_protection = evidence.get("branch_protection") + live_protection = live.get("branch_protection") + if claimed_protection == "protected" \ + and live_protection != "protected": + mismatches.append( + "branch_protection_not_protected_live: %s" + % live_protection) + return mismatches + def verify_gate_evidence(evidence, reader=None, repo=None, ref=None): """只读核对调用方门禁证据与实时数据。 @@ -103,38 +190,246 @@ def verify_gate_evidence(evidence, reader=None, repo=None, ref=None): return {"checked": False, "verified": False, "mismatches": [], "reason": "reader_failed: non-dict result"} - mismatches = [] - live_checks = live.get("ci_checks") or {} - if not isinstance(live_checks, dict): - live_checks = {} - evidence = evidence or {} - if evidence.get("ci_status") == "success": - failed_live = sorted( - name for name, status in live_checks.items() - if status != "success") - if failed_live: - mismatches.append( - "ci_not_green_live: %s" % ", ".join(failed_live)) - for check in (evidence.get("ci_checks") or []): - if not isinstance(check, dict): - continue - name = check.get("name") - if not name: - continue - if check.get("status") == "success" \ - and live_checks.get(name) != "success": - mismatches.append( - "check_not_green_live: %s" % name) - for name in (evidence.get("required_checks") or []): - if live_checks.get(name) != "success": - mismatches.append( - "required_check_not_green_live: %s" % name) - claimed_protection = evidence.get("branch_protection") - live_protection = live.get("branch_protection") - if claimed_protection == "protected" \ - and live_protection != "protected": - mismatches.append( - "branch_protection_not_protected_live: %s" - % live_protection) + mismatches = _compare_claimed_vs_live(evidence, live) return {"checked": True, "verified": not mismatches, "mismatches": mismatches, "reason": None} + + +@dataclass(frozen=True) +class LiveChecksResult: + """get_checks 的返回结构：状态 + 原始证据。 + + checks：{检查名: 状态}，状态与声称口径一致 + （"success" 为绿，其它一律非绿）。 + raw：原始证据体（真实 reader 未来填 GitHub Checks + API 的原始 payload；假 reader 可填任意可追溯值， + 也可为 None）。比对只用 checks，raw 只供审计追溯， + 绝不参与“通过”判定。 + """ + checks: dict + raw: object = None + + +@dataclass(frozen=True) +class LiveProtectionResult: + """get_branch_protection 的返回结构：状态 + 原始证据。 + + status："protected" / "unprotected" / 其它；None 表 + 示实时数据无此项（声称 protected 时即 mismatch）。 + raw：原始证据体（真实 reader 未来填分支保护 API + 的原始 payload；假 reader 可填任意可追溯值，也可 + 为 None），同样只供追溯、不参与判定。 + """ + status: object = None + raw: object = None + + +@runtime_checkable +class LiveEvidenceReader(Protocol): + """可注入的实时证据 reader 接口（v7 机制第 6 项）。 + + 生产代码只定义接口、不实现真实 reader：真实实现 + （走 GitHub Checks / 分支保护只读接口）须在 B6 真 + 实 App 安装与 B1 等用户侧事项就绪后另行接入，本 + 模块绝不内置 HTTP 调用、token 读取或 GitHub API + 调用。测试与验收一律用实现本协议的假 reader。 + """ + + def get_checks(self, repo, ref): + """返回 LiveChecksResult（或等价 dict，见 + _normalize_checks）：repo 在 ref 处的 CI 检查 + 实时状态与原始证据。""" + raise NotImplementedError + + def get_branch_protection(self, repo, branch): + """返回 LiveProtectionResult（或等价 dict / 纯 + 状态字符串，见 _normalize_protection）：repo 的 + branch 分支保护实时状态与原始证据。""" + raise NotImplementedError + + +def _normalize_checks(result): + """归一化 get_checks 返回值 -> (checks dict, raw) 或 None。 + + 接受 LiveChecksResult、含 "checks"/"ci_checks" dict + 的 dict、以及本身就是 {名: 状态} 的 dict；其它形状 + （非 dict checks、非字符串状态）返回 None，调用方 + 按 reader_failed 处理，不猜、不放行。 + """ + if isinstance(result, LiveChecksResult): + checks, raw = result.checks, result.raw + elif isinstance(result, dict): + raw = result.get("raw") + if isinstance(result.get("checks"), dict): + checks = result["checks"] + elif isinstance(result.get("ci_checks"), dict): + checks = result["ci_checks"] + else: + checks = result + else: + return None + if not isinstance(checks, dict): + return None + if any(not isinstance(status, str) for status in checks.values()): + return None + return checks, raw + + +def _normalize_protection(result): + """归一化 get_branch_protection 返回值。 + + 返回 (status, raw, malformed)：status 为字符串或 + None（无此项）；malformed 为 True 时调用方按 + reader_failed 处理（例如返回了整数等非法形状）。 + 纯字符串即状态本身；无 "status"/"branch_protection" + 键的 dict 视为无此项（None），不是 malformed—— + 声称 protected 时它仍会记 mismatch。 + """ + if isinstance(result, LiveProtectionResult): + status, raw = result.status, result.raw + elif isinstance(result, str): + status, raw = result, None + elif result is None: + status, raw = None, None + elif isinstance(result, dict): + raw = result.get("raw") + if isinstance(result.get("status"), str): + status = result["status"] + elif isinstance(result.get("branch_protection"), str): + status = result["branch_protection"] + else: + status = None + else: + return None, None, True + if status is not None and not isinstance(status, str): + return None, None, True + return status, raw, False + + +def _is_live_reader(reader): + """是否协议 reader：同时有 get_checks 与 + get_branch_protection 两个方法（鸭子类型，不强制 + isinstance，因为假 reader 未必显式继承协议）。""" + return callable(getattr(reader, "get_checks", None)) \ + and callable(getattr(reader, "get_branch_protection", None)) + + +def verify_live_evidence(claimed, reader=None, repo=None, ref=None, + branch=None): + """实时证据比对（v7 机制第 6 项，fail-closed）。 + + claimed：调用方声称的门禁证据 dict（与硬门禁同款 + 字段：ci_status、ci_checks[{name, status}]、 + required_checks、branch_protection）。 + reader：LiveEvidenceReader 协议实例，或旧式 callable + (repo, ref) -> {"ci_checks": {...}, "branch_protection": + str}（与 verify_gate_evidence 同款；兼容旧假 reader）。 + repo/ref：求证目标；branch：分支保护的求证分支， + 未给时用 ref 作分支名（调用方须保证 ref 在此语义 + 下就是分支；SHA 求证时请显式传 branch，绝不由本 + 函数猜分支与 SHA 的对应关系）。 + 返回 {"checked", "verified", "mismatches", "reason", + "live"}：live 为归一化后的实时数据（含原始证据 + raw）或 None（未核对时）。verified 为 False 的情形： + 无 reader、无 repo/ref、reader 抛异常、reader 返回 + 恶化形状、实时数据与声称不一致——一律不是“通过”。 + 比对规则与 verify_gate_evidence 共用 _compare，口 + 径一致；本函数不发网络请求、不读 token。 + """ + if reader is None: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "no_trusted_reader", "live": None} + if not repo or not ref: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "no_verification_target", "live": None} + if _is_live_reader(reader): + try: + checks_result = reader.get_checks(repo, ref) + except Exception as exc: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: %s" % exc, + "live": None} + normalized = _normalize_checks(checks_result) + if normalized is None: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: malformed checks", + "live": None} + checks, checks_raw = normalized + try: + protection_result = reader.get_branch_protection( + repo, branch if branch else ref) + except Exception as exc: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: %s" % exc, + "live": None} + protection, protection_raw, malformed = \ + _normalize_protection(protection_result) + if malformed: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: malformed protection", + "live": None} + live = {"ci_checks": checks, + "branch_protection": protection, + "checks_raw": checks_raw, + "protection_raw": protection_raw} + elif callable(reader): + try: + live_raw = reader(repo, ref) + except Exception as exc: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: %s" % exc, + "live": None} + if not isinstance(live_raw, dict): + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: non-dict result", + "live": None} + live = live_raw + else: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: unsupported reader type", + "live": None} + mismatches = _compare_claimed_vs_live(claimed, live) + return {"checked": True, "verified": not mismatches, + "mismatches": mismatches, "reason": None, + "live": live} + + +def live_gate_decision(claimed, reader=None, repo=None, ref=None, + branch=None): + """端点层 live 门禁判定（v7 机制第 6 项，fail-closed）。 + + verified=False 一律阻断，返回 allowed=False、 + http_status=409、status="blocked"、blocked=True： + code 为 live_evidence_mismatch（已核对但不一致） + 或 live_evidence_unverified（未核对：无 reader、 + 无目标、reader 失败）。verified=True 时返回 + allowed=True、http_status=200、status="allowed"、 + code="live_evidence_verified"。返回里附完整 live + 比对结果（live 键），供端点写进 409 响应与审计。 + 注意：本函数是判定缝，不是端点接线——HTTP 端点 + 须在有生产 reader 可注入时才调用它；无 reader 时 + 调用它会阻断（这是 fail-closed 的正确语义），所 + 以在生产 reader 就绪前端点不得接线（见模块头残 + 留）。本函数本身是纯函数，不写审计、不改状态。 + """ + result = verify_live_evidence(claimed, reader, repo, ref, + branch=branch) + if result["verified"]: + return {"allowed": True, "blocked": False, + "status": "allowed", "http_status": 200, + "code": "live_evidence_verified", + "live": result} + code = "live_evidence_mismatch" if result["checked"] \ + else "live_evidence_unverified" + return {"allowed": False, "blocked": True, + "status": "blocked", "http_status": 409, + "code": code, "live": result} diff --git a/cloud/tests/test_scorecard_mech_live.py b/cloud/tests/test_scorecard_mech_live.py new file mode 100644 index 0000000..6d6f9f0 --- /dev/null +++ b/cloud/tests/test_scorecard_mech_live.py @@ -0,0 +1,311 @@ +"""v7 机制第 6 项：生产证据 fail-closed live 接线定向测试。 + +只测这一项，不碰其它 7 项，不改阈值（阈值仍 +test_default_unapproved，见 cloud/audit_scorecard.py， +本文件不测阈值）、不改 #238 auto_apply、不改 #247 +“纠正为 pass 本身不解封”、不改 v1.1 契约。 + +被测对象（cloud/gate_verification.py）： +- LiveEvidenceReader 协议：get_checks(repo, ref) / + get_branch_protection(repo, branch)，返回结构含状态 + 与原始证据；生产代码里绝无真实 HTTP 调用、真实 + token 读取、真实 GitHub API 调用（本文件以 AST 断 + 言钉住这一点）。 +- verify_live_evidence(claimed, reader, repo, ref)：无 + reader、无 repo/ref、reader 失败、reader 返回与声称 + 不一致时 verified 一律 False；声称 CI 全绿但假 reader + 返回失败/缺失时记 mismatch。 +- live_gate_decision：verified=False 时返回 409 语义 + （allowed=False、http_status=409、status="blocked"、 + blocked=True），verified=True 时返回 200 语义。 + +诚实边界：本文件测的是判定函数，不冒充 HTTP 端点 +已接线——_merge_task / _release_task 本批未接 +live_gate_decision（无生产 reader 可注入；见 +gate_verification 模块头残留），真实 live 验证仍待 +B6 真实 App 安装与 B1 等用户侧事项。 +""" + +import ast +import unittest +from pathlib import Path + +try: + from cloud.gate_verification import LiveChecksResult + from cloud.gate_verification import LiveEvidenceReader + from cloud.gate_verification import LiveProtectionResult + from cloud.gate_verification import live_gate_decision + from cloud.gate_verification import verify_live_evidence +except ImportError: + import sys + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.gate_verification import LiveChecksResult + from cloud.gate_verification import LiveEvidenceReader + from cloud.gate_verification import LiveProtectionResult + from cloud.gate_verification import live_gate_decision + from cloud.gate_verification import verify_live_evidence + + +def _claimed_green(): + return { + "ci_status": "success", + "ci_checks": [{"name": "ci", "status": "success"}], + "required_checks": ["ci"], + "branch_protection": "protected", + } + + +class FakeReader(LiveEvidenceReader): + """实现协议的假 reader：不发网络请求，只返回预设值。 + + 同时记录收到的 repo/ref/branch，断言接线把求证目标 + 原样传给 reader（不猜目标）。 + """ + + def __init__(self, checks, protection, checks_raw=None, + protection_raw=None): + self.checks = checks + self.protection = protection + self.checks_raw = checks_raw + self.protection_raw = protection_raw + self.seen = [] + + def get_checks(self, repo, ref): + self.seen.append(("checks", repo, ref)) + return LiveChecksResult(checks=dict(self.checks), + raw=self.checks_raw) + + def get_branch_protection(self, repo, branch): + self.seen.append(("protection", repo, branch)) + return LiveProtectionResult(status=self.protection, + raw=self.protection_raw) + + +class FailingReader(LiveEvidenceReader): + def __init__(self, fail_on): + self.fail_on = fail_on + + def get_checks(self, repo, ref): + if self.fail_on == "checks": + raise RuntimeError("github down") + return LiveChecksResult(checks={"ci": "success"}) + + def get_branch_protection(self, repo, branch): + if self.fail_on == "protection": + raise RuntimeError("github down") + return LiveProtectionResult(status="protected") + + +class LiveEvidenceTestCase(unittest.TestCase): + def test_matching_fake_reader_verified(self): + reader = FakeReader({"ci": "success"}, "protected", + checks_raw={"total_count": 1}, + protection_raw={"required": True}) + result = verify_live_evidence( + _claimed_green(), reader, "o/r", "sha1", + branch="main") + self.assertTrue(result["checked"]) + self.assertTrue(result["verified"]) + self.assertEqual(result["mismatches"], []) + self.assertIsNone(result["reason"]) + self.assertEqual(result["live"]["ci_checks"], + {"ci": "success"}) + self.assertEqual(result["live"]["branch_protection"], + "protected") + self.assertEqual(result["live"]["checks_raw"], + {"total_count": 1}) + self.assertEqual(result["live"]["protection_raw"], + {"required": True}) + self.assertIn(("checks", "o/r", "sha1"), reader.seen) + self.assertIn(("protection", "o/r", "main"), reader.seen) + + def test_lied_all_green_vs_failing_live_mismatch_to_409(self): + # 谎报全绿：声称 ci success + required ci，但实时数据 + # 里 ci 是 failure、且 required 的 lint 查无此项。 + claimed = _claimed_green() + claimed["required_checks"] = ["ci", "lint"] + reader = FakeReader({"ci": "failure"}, "protected") + result = verify_live_evidence(claimed, reader, "o/r", "sha1") + self.assertTrue(result["checked"]) + self.assertFalse(result["verified"]) + self.assertTrue(any("ci_not_green_live" in m + for m in result["mismatches"])) + self.assertTrue(any("check_not_green_live: ci" in m + for m in result["mismatches"])) + self.assertTrue(any("required_check_not_green_live: lint" in m + for m in result["mismatches"])) + decision = live_gate_decision(claimed, reader, "o/r", "sha1") + self.assertFalse(decision["allowed"]) + self.assertTrue(decision["blocked"]) + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["status"], "blocked") + self.assertEqual(decision["code"], "live_evidence_mismatch") + self.assertFalse(decision["live"]["verified"]) + + def test_protection_mismatch_blocks(self): + reader = FakeReader({"ci": "success"}, "unprotected") + result = verify_live_evidence( + _claimed_green(), reader, "o/r", "sha1") + self.assertTrue(result["checked"]) + self.assertFalse(result["verified"]) + self.assertTrue(any("branch_protection" in m + for m in result["mismatches"])) + decision = live_gate_decision( + _claimed_green(), reader, "o/r", "sha1") + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["status"], "blocked") + + def test_no_reader_is_not_a_pass(self): + result = verify_live_evidence( + _claimed_green(), None, "o/r", "sha1") + self.assertFalse(result["checked"]) + self.assertFalse(result["verified"]) + self.assertEqual(result["reason"], "no_trusted_reader") + self.assertIsNone(result["live"]) + decision = live_gate_decision( + _claimed_green(), None, "o/r", "sha1") + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["status"], "blocked") + self.assertEqual(decision["code"], "live_evidence_unverified") + + def test_no_repo_or_ref_is_not_a_pass(self): + reader = FakeReader({"ci": "success"}, "protected") + for repo, ref in ((None, "sha1"), ("o/r", None), + ("o/r", ""), ("", "sha1")): + result = verify_live_evidence( + _claimed_green(), reader, repo, ref) + self.assertFalse(result["checked"]) + self.assertFalse(result["verified"]) + self.assertEqual(result["reason"], + "no_verification_target") + decision = live_gate_decision( + _claimed_green(), reader, repo, ref) + self.assertEqual(decision["http_status"], 409) + + def test_reader_failure_is_not_a_pass(self): + for fail_on in ("checks", "protection"): + result = verify_live_evidence( + _claimed_green(), FailingReader(fail_on), + "o/r", "sha1") + self.assertFalse(result["checked"]) + self.assertFalse(result["verified"]) + self.assertTrue( + result["reason"].startswith("reader_failed")) + decision = live_gate_decision( + _claimed_green(), FailingReader(fail_on), + "o/r", "sha1") + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["code"], + "live_evidence_unverified") + + def test_malformed_reader_result_is_not_a_pass(self): + class BadChecks(LiveEvidenceReader): + def get_checks(self, repo, ref): + return 42 + + def get_branch_protection(self, repo, branch): + return LiveProtectionResult(status="protected") + + class BadStatus(LiveEvidenceReader): + def get_checks(self, repo, ref): + return LiveChecksResult(checks={"ci": 1}) + + def get_branch_protection(self, repo, branch): + return LiveProtectionResult(status="protected") + + for reader in (BadChecks(), BadStatus()): + result = verify_live_evidence( + _claimed_green(), reader, "o/r", "sha1") + self.assertFalse(result["checked"]) + self.assertFalse(result["verified"]) + self.assertTrue( + result["reason"].startswith("reader_failed")) + + def test_matching_fake_reader_decision_allows(self): + reader = FakeReader({"ci": "success"}, "protected") + decision = live_gate_decision( + _claimed_green(), reader, "o/r", "sha1", + branch="main") + self.assertTrue(decision["allowed"]) + self.assertFalse(decision["blocked"]) + self.assertEqual(decision["http_status"], 200) + self.assertEqual(decision["status"], "allowed") + self.assertEqual(decision["code"], "live_evidence_verified") + + def test_branch_defaults_to_ref_and_explicit_branch_wins(self): + reader = FakeReader({"ci": "success"}, "protected") + verify_live_evidence(_claimed_green(), reader, "o/r", "sha1") + self.assertIn(("protection", "o/r", "sha1"), reader.seen) + reader2 = FakeReader({"ci": "success"}, "protected") + verify_live_evidence(_claimed_green(), reader2, "o/r", "sha1", + branch="main") + self.assertIn(("protection", "o/r", "main"), reader2.seen) + + def test_callable_legacy_reader_still_works(self): + def live_ok(repo, ref): + self.assertEqual((repo, ref), ("o/r", "sha1")) + return {"ci_checks": {"ci": "success"}, + "branch_protection": "protected"} + + result = verify_live_evidence( + _claimed_green(), live_ok, "o/r", "sha1") + self.assertTrue(result["verified"]) + + def live_bad(repo, ref): + return {"ci_checks": {"ci": "failure"}, + "branch_protection": "protected"} + + result = verify_live_evidence( + _claimed_green(), live_bad, "o/r", "sha1") + self.assertFalse(result["verified"]) + self.assertTrue(any("ci_not_green_live" in m + for m in result["mismatches"])) + decision = live_gate_decision( + _claimed_green(), live_bad, "o/r", "sha1") + self.assertEqual(decision["http_status"], 409) + + def test_no_real_github_call_in_production_module(self): + # 注意：旧模块头 docstring 本来就以文字描述过“真实 + # reader 用 github_app 的 _github_api_request”（未来 + # 设想，非代码），所以断言须区分文字与代码：代码 + # 位置（import / 调用 / 非 docstring 字符串）里不 + # 得出现真实调用，docstring 文字不算。 + path = Path(__file__).resolve().parents[1] / "gate_verification.py" + tree = ast.parse(path.read_text()) + imported = set() + docstring_nodes = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Module, ast.ClassDef, + ast.FunctionDef, + ast.AsyncFunctionDef)): + body = node.body + if body and isinstance(body[0], ast.Expr) \ + and isinstance(body[0].value, ast.Constant) \ + and isinstance(body[0].value.value, str): + docstring_nodes.add(id(body[0].value)) + if isinstance(node, ast.Import): + imported.update(a.name.split(".")[0] + for a in node.names) + elif isinstance(node, ast.ImportFrom): + if node.module: + imported.add(node.module.split(".")[0]) + for banned in ("urllib", "requests", "http", "socket"): + self.assertNotIn(banned, imported, + "production module must not import %s" + % banned) + code_idents = set() + for node in ast.walk(tree): + if isinstance(node, ast.Attribute): + code_idents.add(node.attr) + elif isinstance(node, ast.Name): + code_idents.add(node.id) + elif isinstance(node, ast.Constant) \ + and isinstance(node.value, str) \ + and id(node) not in docstring_nodes: + self.assertNotIn("api.github.com", node.value) + self.assertNotIn("_github_api_request", code_idents) + + +if __name__ == "__main__": + unittest.main()