153 lines
6.9 KiB
Python
153 lines
6.9 KiB
Python
# ============================================================
|
||
# 统一 EvidenceItem 与可追溯链 v1（moduleId: core-evidence, 可重生 ✅）
|
||
# plan.md §3.4 / 矩阵「结论、动作、报告可追溯和可复算」：
|
||
# 把 run-id、算法版本、种子、知识版本、用户确认串成一条可复算链
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import hashlib # 链哈希
|
||
import json # 规范化序列化
|
||
from dataclasses import dataclass, field, asdict # 结构化记录
|
||
from typing import Any # 类型标注
|
||
|
||
# 证据引用语法：<kind>:<id>（与既有 schedule-version:<id> 协议兼容）
|
||
_EVIDENCE_KINDS = frozenset({
|
||
"schedule-version", "run", "algorithm", "seed", "knowledge", "confirm", "report",
|
||
})
|
||
|
||
|
||
def evidence_ref(kind: str, ref_id: Any) -> str:
|
||
"""生成统一证据引用：<kind>:<id>（kind 白名单，防注入）。"""
|
||
if kind not in _EVIDENCE_KINDS:
|
||
raise ValueError(f"unknown evidence kind: {kind!r}")
|
||
return f"{kind}:{ref_id}"
|
||
|
||
|
||
def parse_evidence_ref(ref: str) -> tuple[str, str]:
|
||
"""解析证据引用为 (kind, id)；非法引用抛 ValueError。"""
|
||
if not isinstance(ref, str) or ":" not in ref:
|
||
raise ValueError(f"malformed evidence ref: {ref!r}")
|
||
kind, _, ref_id = ref.partition(":")
|
||
if kind not in _EVIDENCE_KINDS:
|
||
raise ValueError(f"unknown evidence kind: {kind!r}")
|
||
return kind, ref_id
|
||
|
||
|
||
@dataclass
|
||
class EvidenceItem:
|
||
"""一条结构化证据：可追溯链的原子元素。
|
||
|
||
kind: 证据种类（schedule-version/run/algorithm/seed/knowledge/confirm/report）
|
||
ref: 人类可读引用（如 run-<id>）
|
||
version: 版本号（算法/知识/种子版本）
|
||
runId: 求解运行 ID
|
||
engine: 引擎类型（RULE/CP/GA/HYBRID/EXTERNAL）
|
||
seed: 随机种子（复算用）
|
||
inputsHash: 输入包哈希（复算用）
|
||
knowledgeVersion: 知识库版本（RAG 出处）
|
||
confirmId: 用户确认令牌（P2/P3 门禁）
|
||
note: 审批意见/备注
|
||
meta: 额外元数据
|
||
"""
|
||
kind: str
|
||
ref: str
|
||
version: str | None = None
|
||
runId: str | None = None
|
||
engine: str | None = None
|
||
seed: Any = None
|
||
inputsHash: str | None = None
|
||
knowledgeVersion: str | None = None
|
||
confirmId: str | None = None
|
||
note: str | None = None
|
||
meta: dict[str, Any] = field(default_factory=dict)
|
||
|
||
def __post_init__(self) -> None:
|
||
if self.kind not in _EVIDENCE_KINDS:
|
||
raise ValueError(f"unknown evidence kind: {self.kind!r}")
|
||
|
||
|
||
def _canonical_json(value: Any) -> str:
|
||
return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), default=str)
|
||
|
||
|
||
def trace_chain(items: list[EvidenceItem | dict[str, Any]]) -> dict[str, Any]:
|
||
"""把多条证据串成可复算链。
|
||
|
||
chainHash = SHA256(规范化 JSON 串联)，任何元素/顺序变化都会断链。
|
||
返回 {items, chainHash, count}。
|
||
"""
|
||
normalized: list[dict[str, Any]] = []
|
||
for item in items:
|
||
if isinstance(item, EvidenceItem):
|
||
normalized.append(asdict(item))
|
||
elif isinstance(item, dict):
|
||
normalized.append(dict(item))
|
||
else:
|
||
raise TypeError(f"evidence item must be EvidenceItem or dict, got {type(item)}")
|
||
chain_hash = hashlib.sha256(_canonical_json(normalized).encode("utf-8")).hexdigest()
|
||
return {"items": normalized, "chainHash": chain_hash, "count": len(normalized)}
|
||
|
||
|
||
def verify_chain(items: list[EvidenceItem | dict[str, Any]], expected_hash: str) -> bool:
|
||
"""校验链哈希（可复算/可追溯）。"""
|
||
return trace_chain(items)["chainHash"] == expected_hash
|
||
|
||
|
||
def verify_audit_trace(events: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""审计链完整性校验（矩阵 114：trace 断链检测，重算链哈希逐条比对）。
|
||
|
||
遍历 ALGO_RUN 审计的 traceChainHash，重算链哈希逐条比对：
|
||
- 结构校验：哈希格式（64 位十六进制）、traceCount 与 traceSummary 条数一致、
|
||
summary 条目 kind 白名单 + ref 非空；
|
||
- 链重算：若审计 rationale 携带 canonical traceItems（ALGO_RUN 写入时落盘），
|
||
用 trace_chain 重算并与期望 traceChainHash 比对（verify_chain 语义）；
|
||
旧记录（无 traceItems）仅结构校验，避免误报。
|
||
返回 {"ok", "checked", "broken"}；broken 明细：{eventId, expected, actual, reason}。
|
||
"""
|
||
import re as _re
|
||
|
||
_hex64 = _re.compile(r"^[0-9a-f]{64}$")
|
||
broken: list[dict[str, Any]] = []
|
||
checked = 0
|
||
for ev in events:
|
||
rationale = ev.get("rationale") or {}
|
||
expected = rationale.get("traceChainHash")
|
||
if not expected:
|
||
continue # 非 trace 审计（无链）跳过
|
||
checked += 1
|
||
event_id = ev.get("id")
|
||
summary = rationale.get("traceSummary")
|
||
count = rationale.get("traceCount")
|
||
if not (isinstance(expected, str) and _hex64.match(expected)):
|
||
broken.append({"eventId": event_id, "expected": str(expected),
|
||
"actual": "invalid-hash", "reason": "bad-hash-format"})
|
||
continue
|
||
if not isinstance(summary, list):
|
||
broken.append({"eventId": event_id, "expected": str(count),
|
||
"actual": "missing-summary", "reason": "missing-summary"})
|
||
continue
|
||
if not isinstance(count, int) or count != len(summary):
|
||
broken.append({"eventId": event_id, "expected": str(count),
|
||
"actual": str(len(summary)), "reason": "count-mismatch"})
|
||
continue
|
||
bad = next((it for it in summary
|
||
if not isinstance(it, dict)
|
||
or it.get("kind") not in _EVIDENCE_KINDS
|
||
or not isinstance(it.get("ref"), str) or not it["ref"]), None)
|
||
if bad is not None:
|
||
broken.append({"eventId": event_id, "expected": "valid-kind/ref",
|
||
"actual": str(bad), "reason": "bad-summary-entry"})
|
||
continue
|
||
items = rationale.get("traceItems") # canonical 全量条目（写入时落盘）
|
||
if isinstance(items, list) and items:
|
||
actual = trace_chain(items)["chainHash"] # 重算链哈希（可复算）
|
||
if actual != expected:
|
||
broken.append({"eventId": event_id, "expected": expected,
|
||
"actual": actual, "reason": "chain-hash-mismatch"})
|
||
return {"ok": len(broken) == 0, "checked": checked, "broken": broken}
|
||
|
||
|
||
def schedule_version_item(version_id: Any, **kwargs: Any) -> EvidenceItem:
|
||
"""构造 schedule-version 证据项（与既有协议兼容）。"""
|
||
return EvidenceItem(kind="schedule-version", ref=str(version_id), version=str(version_id), **kwargs)
|