189 lines
7.8 KiB
Python
189 lines
7.8 KiB
Python
"""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()
|