From 19fe4fc7e1e029bc40d09b304d4c5e0a8ccbad37 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 15:29:33 +0000 Subject: [PATCH] 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 | 141 ++++++++++++++- cloud/tests/test_scorecard_mech_evidence.py | 188 ++++++++++++++++++++ 2 files changed, 326 insertions(+), 3 deletions(-) create mode 100644 cloud/tests/test_scorecard_mech_evidence.py diff --git a/cloud/audit_scorecard.py b/cloud/audit_scorecard.py index 1299151..656e709 100644 --- a/cloud/audit_scorecard.py +++ b/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(): diff --git a/cloud/tests/test_scorecard_mech_evidence.py b/cloud/tests/test_scorecard_mech_evidence.py new file mode 100644 index 0000000..d71e6c7 --- /dev/null +++ b/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()