283 lines
12 KiB
Python
283 lines
12 KiB
Python
"""评分卡标尺版本化：仅追加的持久存储（v7 机制第 1 项）。
|
||
|
||
问题：cloud/audit_scorecard.py 的 DEFAULT_THRESHOLDS 只存在于
|
||
代码里，标尺一改，旧标尺下的分数就再也算不出来、也对不上
|
||
——审计员每次重审都在新标尺下重算旧分，基线漂移就是这么来的。
|
||
|
||
解法（复用 audit_loop_driver 的键思路，不复用它的表）：
|
||
- audit_loop_driver 以 (module_id, contract_version) 为键，
|
||
合同版本一变，计数与预算全部重置、旧版本状态原地不动。
|
||
这里同理以 rubric_version 为键：标尺改动 = 写一个新版本行，
|
||
旧版本行原地不动；同一证据在旧版本下重算，分数逐字相同。
|
||
- 不复用 audit_loop_state 表：那张表存的是循环状态
|
||
（pushback 轮次、失败计数），标尺不是循环状态，混表会让
|
||
两类数据的生命周期互相牵连。故新开两张表（见下）。
|
||
|
||
两张表（SQLite / Postgres 兼容写法，沿用 audit_loop_wiring 的
|
||
conn + ``?`` 占位符 + CREATE TABLE IF NOT EXISTS 模式；类型只用
|
||
TEXT / REAL，不用任何一方专属类型，不用自增列）：
|
||
- scorecard_rubric_versions：标尺版本行，主键 rubric_version。
|
||
列：rubric_version、thresholds_snapshot（归一化阈值表的
|
||
canonical JSON）、baseline_head（该版本锚定的基线提交，
|
||
调用方显式传入，本模块不跑 git、不猜 HEAD）、
|
||
thresholds_source（user_approved 仅当全部条目都已拍板，
|
||
其余一律 test_default_unapproved）、created_at。
|
||
- scorecard_snapshots：已算出的卡片行，主键
|
||
(rubric_version, evidence_sha256)。列除主键外还有
|
||
baseline_head 与 thresholds_snapshot（从版本行反规范化而来，
|
||
让单行快照自带审计三要素，不必联表也能核对）、
|
||
scorecard_json（build_scorecard 的完整输出）、created_at。
|
||
选型理由：只存版本行 + 现算也能复现旧分（scorer 是纯函数），
|
||
但快照行让“旧行分数不变”成为可直接读回的事实而非“重算
|
||
碰巧一致”的推论；快照不存证据原文，只存证据的 SHA-256，
|
||
避免证据体在库里再留一份副本。
|
||
|
||
仅追加（append-only），这是本模块的核心承诺：
|
||
- 对外 API 只有 create / get / list / latest / score / record /
|
||
get_snapshot，没有任何改旧行、删旧行的方法；重复建同一
|
||
版本号直接报 ValueError（主键冲突转译，不做 upsert）。
|
||
- 同一 (rubric_version, evidence) 重复 record 时返回已存行的
|
||
卡片原样，不覆盖、不合并。
|
||
- 本模块防的是“走正常 API 的误写”（与 ledger 的触发器同款
|
||
诚实边界）：能直接拿连接跑 SQL 的人当然能改库，库内机制
|
||
拦不住超级用户；可信锚点仍须落在库外（如导出的行哈希）。
|
||
|
||
事务边界：沿用 audit_loop_wiring 的做法——本模块的写函数
|
||
在写后 commit；调用方若想把多写包进同一事务，应直接用
|
||
底层的 INSERT 语句而非本模块的写函数。读函数不 commit。
|
||
|
||
诚实边界：
|
||
- 阈值本身仍不由本模块拍板：thresholds_source 的判定与
|
||
audit_scorecard 一致，测试默认值永远标
|
||
test_default_unapproved；user_approved 只能由调用方显式
|
||
传入全部已拍板条目。
|
||
- 未在真实 Postgres 上验证：SQL 只用两端通用的写法，
|
||
并以 SQLite（文件库 + :memory:）与假连接（只录语句）
|
||
测试；PG 接通后须以真实 PG 复测一次才能宣称两端可用。
|
||
- 本模块不改 #238 auto_apply（保持 False）、不改 #247
|
||
（纠正为 pass 本身不解封）、不改 v1.1 契约与 7 维定义。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import time
|
||
|
||
from cloud.audit_scorecard import build_scorecard, thresholds_source
|
||
|
||
_RUBRIC_TABLE_SQL = (
|
||
"CREATE TABLE IF NOT EXISTS scorecard_rubric_versions("
|
||
"rubric_version TEXT PRIMARY KEY, "
|
||
"thresholds_snapshot TEXT NOT NULL, "
|
||
"baseline_head TEXT NOT NULL, "
|
||
"thresholds_source TEXT NOT NULL, "
|
||
"created_at REAL NOT NULL)"
|
||
)
|
||
|
||
_SNAPSHOT_TABLE_SQL = (
|
||
"CREATE TABLE IF NOT EXISTS scorecard_snapshots("
|
||
"rubric_version TEXT NOT NULL, "
|
||
"baseline_head TEXT NOT NULL, "
|
||
"evidence_sha256 TEXT NOT NULL, "
|
||
"thresholds_snapshot TEXT NOT NULL, "
|
||
"scorecard_json TEXT NOT NULL, "
|
||
"created_at REAL NOT NULL, "
|
||
"PRIMARY KEY(rubric_version, evidence_sha256))"
|
||
)
|
||
|
||
|
||
def ensure_rubric_tables(conn):
|
||
"""建两张表（幂等）。写法与 audit_loop_wiring 保持一致。"""
|
||
conn.execute(_RUBRIC_TABLE_SQL)
|
||
conn.execute(_SNAPSHOT_TABLE_SQL)
|
||
try:
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _canonical_json(value):
|
||
return json.dumps(value, sort_keys=True, separators=(",", ":"))
|
||
|
||
|
||
def _evidence_sha256(evidence):
|
||
return hashlib.sha256(
|
||
_canonical_json(evidence).encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _normalize_thresholds(thresholds):
|
||
"""阈值表归一化：每项 -> {"value": 数值, "source": 来源}。
|
||
|
||
纯数值条目按 audit_scorecard 的旧规则视作未拍板；
|
||
dict 条目缺 source 时同样视作未拍板，绝不默认已拍板。
|
||
"""
|
||
if not isinstance(thresholds, dict) or not thresholds:
|
||
raise ValueError("thresholds must be a non-empty dict")
|
||
normalized = {}
|
||
for key, entry in thresholds.items():
|
||
if isinstance(entry, dict):
|
||
if "value" not in entry:
|
||
raise ValueError(
|
||
"threshold %r must carry a value" % (key,))
|
||
value = entry["value"]
|
||
source = entry.get("source", "test_default_unapproved")
|
||
else:
|
||
value = entry
|
||
source = "test_default_unapproved"
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||
raise ValueError(
|
||
"threshold %r value must be a number" % (key,))
|
||
if not isinstance(source, str) or not source:
|
||
raise ValueError(
|
||
"threshold %r source must be a non-empty string" % (key,))
|
||
normalized[key] = {"value": value, "source": source}
|
||
return normalized
|
||
|
||
|
||
def _row_to_version(row):
|
||
return {
|
||
"rubric_version": row[0],
|
||
"thresholds_snapshot": row[1],
|
||
"thresholds": json.loads(row[1]),
|
||
"baseline_head": row[2],
|
||
"thresholds_source": row[3],
|
||
"created_at": row[4],
|
||
}
|
||
|
||
|
||
def create_rubric_version(conn, rubric_version, thresholds, baseline_head,
|
||
created_at=None):
|
||
"""写一个新标尺版本行。版本号已存在时报 ValueError，旧行不动。
|
||
|
||
thresholds：audit_scorecard 阈值表形态（dict 条目或纯数值）。
|
||
baseline_head：调用方显式传入的基线提交标识，不可为空。
|
||
"""
|
||
if not isinstance(rubric_version, str) or not rubric_version:
|
||
raise ValueError("rubric_version must be a non-empty string")
|
||
if not isinstance(baseline_head, str) or not baseline_head:
|
||
raise ValueError("baseline_head must be a non-empty string")
|
||
normalized = _normalize_thresholds(thresholds)
|
||
snapshot = _canonical_json(normalized)
|
||
source = thresholds_source(normalized)
|
||
stamp = time.time() if created_at is None else created_at
|
||
ensure_rubric_tables(conn)
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO scorecard_rubric_versions"
|
||
"(rubric_version, thresholds_snapshot, baseline_head,"
|
||
" thresholds_source, created_at) VALUES (?, ?, ?, ?, ?)",
|
||
(rubric_version, snapshot, baseline_head, source, stamp))
|
||
except Exception as exc:
|
||
# 只有主键冲突才转译为“版本已存在”；其它错误原样抛出，
|
||
# 绝不能把真实写失败伪装成版本冲突、更不能转成 upsert。
|
||
text = str(exc).lower()
|
||
if "unique" in text or "duplicate" in text:
|
||
raise ValueError(
|
||
"rubric_version %r already exists; create a new version"
|
||
" instead of modifying it" % (rubric_version,)) from exc
|
||
raise
|
||
conn.commit()
|
||
return {
|
||
"rubric_version": rubric_version,
|
||
"thresholds_snapshot": snapshot,
|
||
"thresholds": normalized,
|
||
"baseline_head": baseline_head,
|
||
"thresholds_source": source,
|
||
"created_at": stamp,
|
||
}
|
||
|
||
|
||
def get_rubric_version(conn, rubric_version):
|
||
"""读一个版本行；不存在返回 None（不抛错、不造默认值）。"""
|
||
ensure_rubric_tables(conn)
|
||
row = conn.execute(
|
||
"SELECT rubric_version, thresholds_snapshot, baseline_head,"
|
||
" thresholds_source, created_at FROM scorecard_rubric_versions"
|
||
" WHERE rubric_version = ?",
|
||
(rubric_version,)).fetchone()
|
||
if row is None:
|
||
return None
|
||
return _row_to_version(row)
|
||
|
||
|
||
def list_rubric_versions(conn):
|
||
"""全部版本行，按 (created_at, rubric_version) 升序（写序）。"""
|
||
ensure_rubric_tables(conn)
|
||
rows = conn.execute(
|
||
"SELECT rubric_version, thresholds_snapshot, baseline_head,"
|
||
" thresholds_source, created_at FROM scorecard_rubric_versions"
|
||
" ORDER BY created_at ASC, rubric_version ASC").fetchall()
|
||
return [_row_to_version(row) for row in rows]
|
||
|
||
|
||
def latest_rubric_version(conn):
|
||
"""最新版本行（created_at 最大，时间并列时版本号最大）；无行返回 None。"""
|
||
ensure_rubric_tables(conn)
|
||
row = conn.execute(
|
||
"SELECT rubric_version, thresholds_snapshot, baseline_head,"
|
||
" thresholds_source, created_at FROM scorecard_rubric_versions"
|
||
" ORDER BY created_at DESC, rubric_version DESC LIMIT 1").fetchone()
|
||
if row is None:
|
||
return None
|
||
return _row_to_version(row)
|
||
|
||
|
||
def scorecard_for_rubric(conn, rubric_version, evidence):
|
||
"""用指定版本行的快照阈值现算一张卡（只读，不写快照表）。
|
||
|
||
未知版本号报 ValueError，不回落到默认值——回落会让旧分
|
||
在不知情下变成新标尺的分，正是版本化要杜绝的漂移。
|
||
"""
|
||
version = get_rubric_version(conn, rubric_version)
|
||
if version is None:
|
||
raise ValueError("unknown rubric_version %r" % (rubric_version,))
|
||
card = build_scorecard(evidence, thresholds=version["thresholds"],
|
||
rubric_version=rubric_version)
|
||
card["baseline_head"] = version["baseline_head"]
|
||
return card
|
||
|
||
|
||
def record_scorecard(conn, rubric_version, evidence, created_at=None):
|
||
"""算卡并写入快照行；同 (版本, 证据) 已有行时返回旧行原样。
|
||
|
||
返回值恒为卡片 dict（新算的或旧存的），调用方凭它核对。
|
||
"""
|
||
version = get_rubric_version(conn, rubric_version)
|
||
if version is None:
|
||
raise ValueError("unknown rubric_version %r" % (rubric_version,))
|
||
card = build_scorecard(evidence, thresholds=version["thresholds"],
|
||
rubric_version=rubric_version)
|
||
card["baseline_head"] = version["baseline_head"]
|
||
digest = _evidence_sha256(evidence)
|
||
ensure_rubric_tables(conn)
|
||
row = conn.execute(
|
||
"SELECT scorecard_json FROM scorecard_snapshots"
|
||
" WHERE rubric_version = ? AND evidence_sha256 = ?",
|
||
(rubric_version, digest)).fetchone()
|
||
if row is not None:
|
||
return json.loads(row[0])
|
||
conn.execute(
|
||
"INSERT INTO scorecard_snapshots"
|
||
"(rubric_version, baseline_head, evidence_sha256,"
|
||
" thresholds_snapshot, scorecard_json, created_at)"
|
||
" VALUES (?, ?, ?, ?, ?, ?)",
|
||
(rubric_version, version["baseline_head"], digest,
|
||
version["thresholds_snapshot"], _canonical_json(card),
|
||
time.time() if created_at is None else created_at))
|
||
conn.commit()
|
||
return card
|
||
|
||
|
||
def get_scorecard_snapshot(conn, rubric_version, evidence):
|
||
"""读一张已存快照卡；无行返回 None（不现算、不回落）。"""
|
||
digest = _evidence_sha256(evidence)
|
||
ensure_rubric_tables(conn)
|
||
row = conn.execute(
|
||
"SELECT scorecard_json FROM scorecard_snapshots"
|
||
" WHERE rubric_version = ? AND evidence_sha256 = ?",
|
||
(rubric_version, digest)).fetchone()
|
||
if row is None:
|
||
return None
|
||
return json.loads(row[0])
|