host
3978dbbbf2
- 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)
312 lines
13 KiB
Python
312 lines
13 KiB
Python
"""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()
|