code-to10 v7 mech-4: scorecard evidence ledger head + row ids, injectable signature verifier (fail-closed)

(cherry picked from commit bcd385c6979db77abea458af66e21ef891453064)

cloud/audit_scorecard.py

@@ -71,6 +71,9 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
+
 PASS_SCORE = 9.0
 
 DIMENSIONS = [
@@ -478,8 +481,132 @@ _SCORERS = {
 }
 
 
+# ---- v7 机制第 4 项：分数证据绑定与签名（只做这一项） ----
+# 诚实边界（先写清，再写代码）：
+# - 本节是纯函数：只算哈希与验签结果，不写库、不写账本、
+#   不移动 cloud/ledger.py 的链头、不改变总体达标规则。
+#   evidence_ledger_head 是调用方传入的证据行即时算出的链头，
+#   不是已持久化/已外部锚定的头；持久化与外部锚定是调用方
+#   与运营方的事，本模块绝不谎称已上链入库。
+# - 防重盖戳靠签名，不靠本节的哈希：哈希只能证明“这份证据体
+#   就是算分时那份”（改动即哈希变），不能证明“这份证据体是
+#   新取的”。没有有效签名的证据，调用方仍可重盖戳再交——
+#   与 cloud/evidence_binding.py 的说明同款局限，本节不冒充解决。
+# - 生产签名密钥供给仍是待定的运营方事项：本模块不含任何
+#   生产密钥、不读生产 env、不新增密钥落盘路径、不默认信任。
+#   机制只是代码就绪，不是生产已激活；真实 PG 上的链验证未做，
+#   本项测试只用纯函数内存计算（连 SQLite 账本集成都没接）。
+# - 签名验证是 fail-closed：无验证器、签名缺失、验证器拒绝、
+#   验证器抛异常，一律 verified=False，绝不默认 verified=True。
+#   验签结果不改变 build_scorecard 的 overall：总体达标仍只看
+#   7 维分数，签名是调用方在传证据行之前自行核验的附加层，
+#   把验签接入门禁/达标是另一次契约决定，不在本项悄悄做。
+EVIDENCE_CHAIN_GENESIS = hashlib.sha256(
+    b"scorecard-evidence-genesis-v1").hexdigest()
+
+
+def canonical_evidence_bytes(evidence_row):
+    """证据体全文的规范字节：dict 按 sort_keys 紧凑 JSON 编码。
+
+    全文即整个 dict，不只取其中某个字段；字段增减、值改动、
+    键顺序之外的任何内容变化都会改变字节，从而改变哈希。
+    非 JSON 可序列化内容抛 ValueError，不静默丢字段。
+    """
+    if isinstance(evidence_row, bytes):
+        return evidence_row
+    if isinstance(evidence_row, str):
+        return evidence_row.encode("utf-8")
+    try:
+        return json.dumps(evidence_row, sort_keys=True,
+                          separators=(",", ":"),
+                          ensure_ascii=False).encode("utf-8")
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            "evidence row must be JSON-serializable: %s" % (exc,))
+
+
+def hash_evidence_body(evidence_row):
+    """证据体全文哈希：规范字节的 SHA-256 hex。"""
+    return hashlib.sha256(
+        canonical_evidence_bytes(evidence_row)).hexdigest()
+
+
+def compute_evidence_ledger(evidence_rows):
+    """对调用方传入的证据行算链头与行 id 列表（纯函数）。
+
+    链式构造（仿 cloud/ledger.py 的头推进，不写库）：
+      head_0 = EVIDENCE_CHAIN_GENESIS
+      head_{i+1} = sha256((head_i + body_hash_i).encode())
+    因此任一证据体改动、行顺序变化都会改变最终头哈希；
+    只改一行 id 而不动其余内容时，该行体哈希随之变，头也变。
+    返回 {"head": str|None, "row_ids": [...], "body_hashes": [...]}。
+    - evidence_rows 为 None 或空：head=None、row_ids=[]、
+      body_hashes=[]，绝不编造一个头或 id。
+    - 行 id 只取行 dict 里现成的 "id"（且不为 None）；没有 id
+      的行仍参与头计算，但不为其发明 id（row_ids 长度可小于
+      行数，调用方要可审计映射须每行自带 id）。
+    - 非 dict 行抛 ValueError：不静默跳过、不猜 id。
+    """
+    if evidence_rows is None:
+        return {"head": None, "row_ids": [], "body_hashes": []}
+    rows = list(evidence_rows)
+    if not rows:
+        return {"head": None, "row_ids": [], "body_hashes": []}
+    row_ids, body_hashes = [], []
+    head = EVIDENCE_CHAIN_GENESIS
+    for row in rows:
+        if not isinstance(row, dict):
+            raise ValueError("evidence row must be a dict")
+        body_hash = hash_evidence_body(row)
+        body_hashes.append(body_hash)
+        head = hashlib.sha256(
+            (head + body_hash).encode("utf-8")).hexdigest()
+        if row.get("id") is not None:
+            row_ids.append(row["id"])
+    return {"head": head, "row_ids": row_ids, "body_hashes": body_hashes}
+
+
+def verify_signature(evidence_bytes, signature, verifier=None):
+    """签名验证器：可注入接口，fail-closed。
+
+    参数：
+    - evidence_bytes: bytes（或 str，按 UTF-8 编码）；其它类型
+      抛 TypeError（这是调用方编程错误，不是“验签失败”）。
+    - signature: 调用方出示的签名（bytes/str 或其它验证器能
+      接受的形式）；为 None 或空 bytes/str 即缺失。
+    - verifier: 调用方注入的可调用对象，签名
+      verifier(evidence_bytes: bytes, signature) -> bool。
+      生产验证器由运营方在密钥供给到位后注入，本模块不提供、
+      不默认、不从 env/文件读任何验证器或密钥。
+    返回 {"verified": bool, "reason": str}，reason 取值：
+      "verified"（验证器返回真）、"verifier_rejected"（返回假）、
+      "no_verifier"（未注入验证器）、"missing_signature"
+      （签名缺失）、"verifier_error"（验证器抛异常，fail-closed）。
+    默认无验证器时 verified=False，绝不默认 True。
+    """
+    if not isinstance(evidence_bytes, (bytes, str)):
+        raise TypeError("evidence_bytes must be bytes or str")
+    if verifier is None:
+        return {"verified": False, "reason": "no_verifier"}
+    if not callable(verifier):
+        raise TypeError("verifier must be callable or None")
+    if signature is None:
+        return {"verified": False, "reason": "missing_signature"}
+    if isinstance(signature, (bytes, str)) and len(signature) == 0:
+        return {"verified": False, "reason": "missing_signature"}
+    payload = (evidence_bytes if isinstance(evidence_bytes, bytes)
+               else evidence_bytes.encode("utf-8"))
+    try:
+        result = verifier(payload, signature)
+    except Exception:
+        return {"verified": False, "reason": "verifier_error"}
+    if result:
+        return {"verified": True, "reason": "verified"}
+    return {"verified": False, "reason": "verifier_rejected"}
+
+
 def build_scorecard(evidence, thresholds=None, rubric_version=None,
-                    baseline_head=None):
+                    baseline_head=None, evidence_rows=None):
     """算出 7 维评分卡。evidence 缺键按空证据处理（该维记 None）。
 
     rubric_version：调用方显式传入的标尺版本号（见
@@ -494,11 +621,18 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None,
     编造、绝不探测版本控制状态。baseline 缺失本身不改写 overall
     语义（overall 仍按 7 维算），漂移与否由 check_baseline_stale
     单独判定，调用方不得只看 overall 就把旧分当现分用。
+
+    evidence_rows（v7 机制第 4 项）：调用方显式传入的证据行
+    列表（每行 dict），本函数只据此算 evidence_ledger_head 与
+    evidence_row_ids（见 compute_evidence_ledger），不写库、
+    不验签、不改变 overall。无证据行时 head 为 None、
+    row_ids 为空列表，不得编造。
     """
     evidence = evidence or {}
     th = thresholds or {}
     if baseline_head is not None and baseline_head == "":
         baseline_head = None
+    ledger = compute_evidence_ledger(evidence_rows)
     dimensions = []
     missing_all, untested_all = [], []
     for dim_id, _name in DIMENSIONS:
@@ -526,6 +660,8 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None,
         "thresholds_source": thresholds_source(th),
         "thresholds": _effective_thresholds(th),
         "rubric_version": rubric_version,
+        "evidence_ledger_head": ledger["head"],
+        "evidence_row_ids": ledger["row_ids"],
     }
 
 
@@ -575,8 +711,7 @@ def check_baseline_stale(scorecard, current_main_head):
         "reason": reason,
         "usable_as_pass_basis": not stale,
         "overall_usable_as_pass_basis": not stale,
-        "overall_must_not_be_used_as_pass_basis": stale,
-    }
+        "overall_must_not_be_used_as_pass_basis": stale,    }
 
 
 def example_scorecard_as_of_2026_10_31():

cloud/tests/test_scorecard_mech_evidence.py

@@ -0,0 +1,188 @@
+"""v7 机制第 4 项：分数证据绑定与签名（定向测试）。
+
+只测本项，不碰其它 7 项、不改阈值口径、不改总体达标规则。
+
+测试签名器只在本文件出现：标准库 hmac + 一个测试用密钥
+（test-only，绝不是生产密钥；生产代码 cloud/audit_scorecard.py
+里没有任何密钥、env 读取或密钥文件路径，验证器一律由调用方
+注入）。生产签名密钥供给仍是待定的运营方事项，本项只是
+代码就绪，不是生产已激活；真实 PG 上的链验证未做，本测试
+只用纯函数内存计算。
+
+钉住的断言：
+- 证据体改动则哈希变化（全文哈希，不是只取某个字段）；
+- 行顺序变化、行增减也改变链头；
+- 无证据行时 head 为 None、row_ids 为空列表，绝不编造；
+- 没有 id 的行不被发明 id，但仍参与头计算；
+- 验签失败 verified=False；无验证器 verified=False；
+  验证器抛异常 verified=False（fail-closed）；
+- 注入正确验证器 verified=True；证据体或签名被改则
+  同一验证器返回 verified=False。
+"""
+
+import hashlib
+import hmac
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.audit_scorecard import (
+        build_scorecard,
+        canonical_evidence_bytes,
+        compute_evidence_ledger,
+        hash_evidence_body,
+        verify_signature,
+    )
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.audit_scorecard import (
+        build_scorecard,
+        canonical_evidence_bytes,
+        compute_evidence_ledger,
+        hash_evidence_body,
+        verify_signature,
+    )
+
+# 仅测试用：hmac 测试签名器（test-only key，绝非生产密钥）。
+_TEST_KEY = b"mech4-test-only-key-do-not-use-in-production"
+
+
+def _test_sign(evidence_bytes):
+    return hmac.new(_TEST_KEY, evidence_bytes, hashlib.sha256).hexdigest()
+
+
+def _test_verifier(evidence_bytes, signature):
+    if not isinstance(signature, str):
+        return False
+    expected = _test_sign(evidence_bytes)
+    return hmac.compare_digest(expected, signature)
+
+
+def _row(row_id, score, note="ok"):
+    return {"id": row_id, "dimension": "security_ledger",
+            "score": score, "note": note}
+
+
+class EvidenceHashTestCase(unittest.TestCase):
+    def test_body_change_changes_hash(self):
+        before = hash_evidence_body(_row("ev-1", 9.5))
+        after = hash_evidence_body(_row("ev-1", 9.5, note="changed"))
+        self.assertNotEqual(before, after)
+        # 只改一个不起眼的字段也必须变：全文哈希，不是字段子集。
+        after2 = hash_evidence_body(_row("ev-1", 9.5, note="ok "))
+        self.assertNotEqual(before, after2)
+
+    def test_hash_is_deterministic_for_same_body(self):
+        self.assertEqual(hash_evidence_body(_row("ev-1", 9.5)),
+                         hash_evidence_body(_row("ev-1", 9.5)))
+
+    def test_order_change_changes_head(self):
+        rows = [_row("ev-1", 9.5), _row("ev-2", 10.0)]
+        forward = compute_evidence_ledger(rows)
+        reversed_ = compute_evidence_ledger(list(reversed(rows)))
+        self.assertNotEqual(forward["head"], reversed_["head"])
+        # 行 id 列表按传入顺序返回，不是排序后。
+        self.assertEqual(forward["row_ids"], ["ev-1", "ev-2"])
+        self.assertEqual(reversed_["row_ids"], ["ev-2", "ev-1"])
+
+    def test_row_addition_changes_head(self):
+        one = compute_evidence_ledger([_row("ev-1", 9.5)])
+        two = compute_evidence_ledger([_row("ev-1", 9.5),
+                                       _row("ev-2", 10.0)])
+        self.assertNotEqual(one["head"], two["head"])
+
+    def test_no_rows_gives_none_head_and_empty_ids(self):
+        for empty in (None, []):
+            ledger = compute_evidence_ledger(empty)
+            self.assertIsNone(ledger["head"])
+            self.assertEqual(ledger["row_ids"], [])
+            self.assertEqual(ledger["body_hashes"], [])
+
+    def test_row_without_id_gets_no_invented_id(self):
+        row = {"dimension": "security_ledger", "score": 9.5}
+        ledger = compute_evidence_ledger([row])
+        self.assertIsNotNone(ledger["head"])
+        self.assertEqual(ledger["row_ids"], [])
+        self.assertEqual(len(ledger["body_hashes"]), 1)
+
+    def test_non_dict_row_rejected_not_silently_skipped(self):
+        with self.assertRaises(ValueError):
+            compute_evidence_ledger(["not-a-dict"])
+
+
+class ScorecardLedgerFieldsTestCase(unittest.TestCase):
+    def test_scorecard_carries_head_and_row_ids(self):
+        rows = [_row("ev-1", 9.5), _row("ev-2", 10.0)]
+        report = build_scorecard({}, evidence_rows=rows)
+        ledger = compute_evidence_ledger(rows)
+        self.assertEqual(report["evidence_ledger_head"], ledger["head"])
+        self.assertEqual(report["evidence_row_ids"], ["ev-1", "ev-2"])
+        # 头字段不改变总体规则：空证据仍是不达标。
+        self.assertEqual(report["overall"], "not_pass")
+
+    def test_scorecard_without_rows_does_not_fabricate(self):
+        report = build_scorecard({})
+        self.assertIsNone(report["evidence_ledger_head"])
+        self.assertEqual(report["evidence_row_ids"], [])
+        # 旧调用（不传 evidence_rows）仍能跑，字段始终在位。
+        self.assertIn("evidence_ledger_head", report)
+        self.assertIn("evidence_row_ids", report)
+
+
+class VerifySignatureTestCase(unittest.TestCase):
+    def test_no_verifier_is_not_verified(self):
+        payload = canonical_evidence_bytes(_row("ev-1", 9.5))
+        result = verify_signature(payload, _test_sign(payload))
+        self.assertFalse(result["verified"])
+        self.assertEqual(result["reason"], "no_verifier")
+
+    def test_missing_signature_is_not_verified_even_with_verifier(self):
+        payload = canonical_evidence_bytes(_row("ev-1", 9.5))
+        for missing in (None, "", b""):
+            result = verify_signature(payload, missing,
+                                      verifier=_test_verifier)
+            self.assertFalse(result["verified"])
+            self.assertEqual(result["reason"], "missing_signature")
+
+    def test_wrong_signature_is_not_verified(self):
+        payload = canonical_evidence_bytes(_row("ev-1", 9.5))
+        result = verify_signature(payload, "deadbeef",
+                                  verifier=_test_verifier)
+        self.assertFalse(result["verified"])
+        self.assertEqual(result["reason"], "verifier_rejected")
+
+    def test_tampered_body_fails_with_same_signature(self):
+        payload = canonical_evidence_bytes(_row("ev-1", 9.5))
+        signature = _test_sign(payload)
+        tampered = canonical_evidence_bytes(_row("ev-1", 9.5,
+                                                 note="changed"))
+        self.assertNotEqual(payload, tampered)
+        result = verify_signature(tampered, signature,
+                                  verifier=_test_verifier)
+        self.assertFalse(result["verified"])
+
+    def test_injected_correct_verifier_is_verified(self):
+        payload = canonical_evidence_bytes(_row("ev-1", 9.5))
+        result = verify_signature(payload, _test_sign(payload),
+                                  verifier=_test_verifier)
+        self.assertTrue(result["verified"])
+        self.assertEqual(result["reason"], "verified")
+
+    def test_verifier_exception_is_fail_closed(self):
+        def _boom(_payload, _signature):
+            raise RuntimeError("test verifier boom")
+
+        payload = canonical_evidence_bytes(_row("ev-1", 9.5))
+        result = verify_signature(payload, "anything", verifier=_boom)
+        self.assertFalse(result["verified"])
+        self.assertEqual(result["reason"], "verifier_error")
+
+    def test_non_callable_verifier_is_programming_error(self):
+        payload = canonical_evidence_bytes(_row("ev-1", 9.5))
+        with self.assertRaises(TypeError):
+            verify_signature(payload, "x", verifier="not-callable")
+
+
+if __name__ == "__main__":
+    unittest.main()