"""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()