From ca3d75e7263707cc09faf665f30b325eeb172b4c Mon Sep 17 00:00:00 2001 From: dev-belly Date: Mon, 12 Oct 2026 21:57:49 +0800 Subject: [PATCH] Bind audit reviews to evidence and reject stale submissions --- README.md | 6 +- docs/CASE_STUDY.md | 2 + docs/REVIEW_INTEGRITY.md | 57 +++++++ pyproject.toml | 2 +- src/controltrace/__init__.py | 2 +- src/controltrace/app.py | 35 ++++- src/controltrace/cli.py | 2 + src/controltrace/exports.py | 24 ++- src/controltrace/reviews.py | 38 +++++ src/controltrace/store.py | 111 ++++++++++---- src/controltrace/verify.py | 31 +++- tests/test_app.py | 23 ++- tests/test_review_evidence.py | 270 ++++++++++++++++++++++++++++++++++ uv.lock | 2 +- 14 files changed, 555 insertions(+), 50 deletions(-) create mode 100644 docs/REVIEW_INTEGRITY.md create mode 100644 tests/test_review_evidence.py diff --git a/README.md b/README.md index 1cb41f2..f073d67 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,10 @@ uv run controltrace verify # 校验 ZIP 文件哈希，并从包内源数据 复核历史按解析后的 UTC 时间排列，保留小数秒精度；时间相同则按稳定复核 ID 排序，因此补录较早的记录不会替换较晚的结论。`verify` 同时检查每条复核的字段、复核人、时间、状态与结论：`pending` 不能带结论，`in_review` 可不填结论或标记待补证，`closed` 必须有最终结论和说明。即使文件哈希及 CSV、底稿内容相互一致，非法复核或逆序历史也会被拒绝。此检查仍不能证明复核人的身份或原始材料真实性。升级到新版本后，请重新导出旧版本底稿。 +**复核不能只跟着发现 ID 走。** 0.2.8 将每次新复核绑定到自动发现的完整证据指纹。相同证据重跑保留当前结论；同一 ID 的关联来源记录或判断口径变化后，原结论保留在历史中，当前状态返回待复核。界面保存时核对首次展示的编辑版本；其他人追加复核或证据发生变化后，须显式重新加载，防止旧表单覆盖新证据。版本检查、复核及指纹关联在一个事务内提交，并对同一发现的并发写入进行冲突检查。[复核完整性设计](docs/REVIEW_INTEGRITY.md)解释业务场景、Python API、迁移和可复现实验。 + +新 ZIP 使用格式 3，增加 `review_evidence.json`，`verify` 根据重放的当前证据选择可适用的复核。旧数据库的复核历史继续保留，但因原证据版本未知，升级后需要重新评估当前证据。格式 2 / 0.2.7 底稿仍可核验，结果明确说明它没有证据版本关联；更早底稿请使用原版本。新包没有归档历史源快照，也不提供身份认证或签名。 + ## 数据字典与案例 - [数据字典](docs/DATA_DICTIONARY.md)：各源表、主键、关联键、时间和来源字段。 @@ -87,7 +91,7 @@ python -m ruff check src tests python -m pytest -q ``` -**本机验证（2026-10-11，Windows / Python 3.12）：** 完整 pytest 为 **74 passed**，Ruff 与 git diff --check 通过；实际执行生成、规则测试、CSV/ZIP 导出及包内重放，仍有 12 条预置线索（9 条演示异常、3 条人工判断），验证 28 个源文件。新增回归确认截止日后创建的账号不造成离职/转岗误报，并覆盖 UTC 偏移及恰在截止时点创建的账号；缺少截止日前账号证据仍列为人工核实。既有测试覆盖日期边界、重复事件、非默认种子、缺失关联、复核时间和状态、界面筛选及重哈希后非法复核的拒绝。[GitHub Actions CI](https://github.com/dev-belly/ControlTrace/actions/workflows/ci.yml) 在 Ubuntu / Python 3.11、3.12 和 Windows / Python 3.12 运行完整 Ruff 与 pytest。规则版本为 0.2.7，旧版底稿应使用原版本校验，升级后重新导出。 +**本机验证（2026-10-12，Windows / Python 3.12.13）：** 完整 pytest 为 **90 passed**，Ruff 与 git diff --check 通过；实际 CSV/ZIP 导出及包内重放有 12 条预置线索（9 条演示异常、3 条人工判断），验证 28 个源文件。新增回归覆盖 CH002 审批时间修正后同一发现重新待复核、无关线索保留结论、两个复核者同版本并发、事务失败回滚、界面显式重新加载、旧数据库迁移、旧格式兼容及重哈希后的伪造当前结论拒绝。既有测试覆盖日期边界、缺失关联、复核 UTC 时间、合法状态和非法 JSON。[GitHub Actions CI](https://github.com/dev-belly/ControlTrace/actions/workflows/ci.yml) 配置在 Ubuntu / Python 3.11、3.12 和 Windows / Python 3.12 运行 Ruff 与完整 pytest。当前包版本为 0.2.8；格式 2 / 0.2.7 底稿可继续核验，新导出采用格式 3。 ## 已知局限 diff --git a/docs/CASE_STUDY.md b/docs/CASE_STUDY.md index 22a17e4..64ea0c2 100644 --- a/docs/CASE_STUDY.md +++ b/docs/CASE_STUDY.md @@ -2,6 +2,8 @@ > 合成案例，非真实企业审计。本文记录一次真实运行 ControlTrace 演示程序并在界面保存示范复核的过程。`confirmed_exception` 仅指本合成数据符合演示控制的时间违例口径，不是对任何真实组织的审计结论。 +本文截图与示范底稿保存的是 0.2.7 的历史运行。0.2.8 新增复核与证据版本关联，以及旧表单冲突检查；旧数据库记录的原证据版本未知，须重新评估后记录新复核。该历史记录不自动获得新版本的证据关联保证。具体行为见[复核完整性设计](REVIEW_INTEGRITY.md)。 + ![D002 的发现详情、证据 ID 和规则依据](../assets/controltrace-finding.png) ## 1. 控制目标与总体 diff --git a/docs/REVIEW_INTEGRITY.md b/docs/REVIEW_INTEGRITY.md new file mode 100644 index 0000000..00a7d4f --- /dev/null +++ b/docs/REVIEW_INTEGRITY.md @@ -0,0 +1,57 @@ +# 复核结论如何绑定证据 + +> 合成案例；这里的关闭状态是线索处理结果，不是企业审计意见。 + +## 业务问题：ID 没变，证据可能已经变了 + +发现 ID 根据控制、原因代码和对象 ID 生成。它方便关联同一条线索，但不代表证据内容不变。例如 D002 的变更工单后来修正了审批时间，即使依然命中“审批晚于上线”，发现 ID 仍然相同。仅按发现 ID 取最后一条复核，会把基于原抽取的 `closed` 结论自动套到修正后的证据上。 + +从 0.2.8 起，每次保存复核都会同时保存其自动发现指纹。指纹覆盖规则口径、关联源记录原值、时间线和自动判断。复核人、处理状态等界面附加字段不参与指纹，规则重跑的刷新时间也不参与。因此相同证据重跑保留结论，关联证据变化则返回 `pending`，原复核仍保留在历史中。 + +当前结论是**在当前指纹下，按 UTC 复核时间和复核 ID 排序的最后一条记录**。历史补录沿用原来的时间排序规则。其他线索的复核不影响本线索的编辑版本。若完整证据恢复为已复核的同一个指纹，对应历史结论也会重新适用；这里比较的是确切内容，不推断来源的可信性。 + +## 保存前核对编辑版本 + +界面首次展示复核表单时保存 `review_revision`，它概括当前证据指纹及该线索的全部历史记录。另一位复核者保存了新记录，或规则重跑后证据变化，当前表单就必须显式重新加载，避免把旧表单中的结论写到新证据上。 + +Python 调用者编辑之前读出的发现时，也应传入这个版本： + +```python +from controltrace.store import get_finding, save_review + +finding = get_finding("data/controltrace.duckdb", "F-2C5FD7D5834D") +assert finding is not None +save_review( + "data/controltrace.duckdb", + finding["finding_id"], + "Demo Reviewer", + "closed", + "confirmed_exception", + "Rechecked the current synthetic approval and deployment records.", + expected_revision=finding["review_revision"], +) +``` + +版本核对、复核记录、证据指纹关联在同一 DuckDB 事务中提交。事务还实际递增该发现行的写入计数，使两个同时读取同一版本的连接发生写入冲突；仅执行字段值不变的更新可能被数据库优化掉，不能充当并发保护。失败会回滚复核及关联，使用者需重新读取当前记录。 + +`expected_revision` 是可选参数，用来保留旧 Python API；省略时，调用者明确按保存时的当前自动发现追加记录，不能获得旧表单检查。界面始终传入版本。这是本机同进程内多个连接的冲突控制，不是分布式协作或多人认证服务。DuckDB 的跨进程文件写入限制仍然适用。 + +## 旧数据与可复现底稿 + +旧数据库没有保存复核时的证据指纹。迁移保留所有原记录，但把它们视为证据版本未知，不从今天的源数据补造历史关联；使用者要重新评估当前证据后记录新结论。 + +新底稿采用格式 3，保留原 `reviews.json` 字段，同时增加 `review_evidence.json`：按发现和复核 ID 保存关联指纹，未知历史为 `null`。CSV 标明当前证据指纹和不适用历史数量，Markdown 明确显示当前结论与全部历史。校验器从包内源记录重放自动发现，重算指纹并独立选择当前结论；即使 CSV、底稿和文件哈希被一致修改，旧证据结论冒充当前结论仍会被拒绝。 + +格式 2、版本 0.2.7 的旧底稿继续支持原始重放；返回 `review_evidence_bound=false`，命令行也说明旧格式没有证据版本关联。更早版本仍需其原版本校验器。这种兼容不会把旧底稿升级成拥有新保证的证据。 + +指纹和包内清单不能证明复核者身份、源系统真实性或防御同时重写所有内容的攻击者。新包只带当前源快照，保留的历史指纹并不等于保存了旧源快照。认证、签名和历史源版本归档仍是实际业务接入前需要单独解决的工作。 + +## 可以现场展示的验证 + +[回归测试](../tests/test_review_evidence.py)直接修改同一发现关联源行，验证发现 ID 不变、指纹改变、旧关闭记录保留、当前状态重新待复核；还覆盖两人同版本并发、绑定失败回滚、旧数据库迁移和重新计算哈希后的伪造关闭底稿。[界面测试](../tests/test_app.py)验证旧表单禁用保存，并在使用者主动重新加载后展示当前复核记录。 + +```bash +uv run --extra dev pytest tests/test_review_evidence.py tests/test_app.py -q +``` + +讨论项目时可以从这个具体问题展开：稳定业务 ID 与证据版本不同；追加历史与当前有效结论不同；事务原子性与读取后写入的并发冲突不同。每个判断都有代码路径和失败场景，而不是依赖合成数据的命中数。 diff --git a/pyproject.toml b/pyproject.toml index 13b321b..4003e8a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "controltrace" -version = "0.2.7" +version = "0.2.8" description = "Synthetic IT audit evidence and internal control testing workbench" readme = "README.md" requires-python = ">=3.11" diff --git a/src/controltrace/__init__.py b/src/controltrace/__init__.py index 7e761ba..7a1537f 100644 --- a/src/controltrace/__init__.py +++ b/src/controltrace/__init__.py @@ -1,3 +1,3 @@ """ControlTrace synthetic IT audit demonstration.""" -__version__ = "0.2.7" +__version__ = "0.2.8" diff --git a/src/controltrace/app.py b/src/controltrace/app.py index 32e2681..b1b60a8 100644 --- a/src/controltrace/app.py +++ b/src/controltrace/app.py @@ -52,8 +52,7 @@ def _display(value: Any) -> str: def _load_findings() -> list[dict[str, Any]]: - if not DB_PATH.exists(): - initialize_db(DB_PATH) + initialize_db(DB_PATH) findings = list_findings(DB_PATH) if not findings: run_tests(DB_PATH) @@ -241,22 +240,38 @@ def _load_findings() -> list[dict[str, Any]]: st.subheader("Human review") current = selected.get("latest_review") or {} + if selected["review_stale_count"]: + st.warning( + f"{selected['review_stale_count']} historical review(s) concern different or unknown " + "evidence. Assess the current source records before recording a new decision." + ) st.caption( f"Current: {current.get('status', 'pending')} · " f"{current.get('conclusion') or 'no conclusion'}" ) finding_id = selected["finding_id"] + revision_key = f"review_base_{finding_id}" + if revision_key not in st.session_state: + st.session_state[revision_key] = selected["review_revision"] + base_revision = st.session_state[revision_key] + stale_form = base_revision != selected["review_revision"] + if stale_form: + st.warning("This finding's evidence or review changed while you were editing. Reload it.") + if st.button("Reload current finding", key=f"reload_{finding_id}"): + st.session_state[revision_key] = selected["review_revision"] + st.rerun() + input_key = f"{finding_id}_{base_revision}" reviewer = st.text_input( "Reviewer name or initials", value=current.get("reviewer", ""), - key=f"reviewer_{finding_id}", + key=f"reviewer_{input_key}", ) status_options = ["pending", "in_review", "closed"] status = st.selectbox( "Processing status", status_options, index=status_options.index(current.get("status", "pending")), - key=f"review_status_{finding_id}", + key=f"review_status_{input_key}", ) conclusion_options = { "pending": [None], @@ -273,23 +288,27 @@ def _load_findings() -> list[dict[str, Any]]: else 0 ), format_func=lambda value: value.replace("_", " ") if value else "No conclusion", - key=f"review_conclusion_{finding_id}_{status}", + key=f"review_conclusion_{input_key}_{status}", ) notes = st.text_area( "Review notes / exception basis", value=current.get("notes", ""), - key=f"review_notes_{finding_id}", + key=f"review_notes_{input_key}", ) - submitted = st.button("Save review", type="primary", width="stretch") + submitted = st.button("Save review", type="primary", width="stretch", disabled=stale_form) if submitted: if not reviewer.strip() or not notes.strip(): st.error("Reviewer and notes are required to preserve a usable review trail.") else: try: - save_review(DB_PATH, finding_id, reviewer.strip(), status, conclusion, notes.strip()) + save_review( + DB_PATH, finding_id, reviewer.strip(), status, conclusion, notes.strip(), + expected_revision=base_revision, + ) except ValueError as error: st.error(str(error)) else: + del st.session_state[revision_key] st.rerun() if selected.get("review_history"): with st.expander(f"Decision history ({len(selected['review_history'])})"): diff --git a/src/controltrace/cli.py b/src/controltrace/cli.py index a5e1745..d429eaa 100644 --- a/src/controltrace/cli.py +++ b/src/controltrace/cli.py @@ -43,6 +43,8 @@ def main(argv: list[str] | None = None) -> int: f"{result['replayed_findings']} observations " f"({result['exported_findings']} in this {result['scope']} export)." ) + if not result["review_evidence_bound"]: + print("Legacy bundle: review history is valid but was not bound to an evidence snapshot.") return 0 from controltrace.rules import run_tests from controltrace.store import initialize_db diff --git a/src/controltrace/exports.py b/src/controltrace/exports.py index a6036e9..eee70de 100644 --- a/src/controltrace/exports.py +++ b/src/controltrace/exports.py @@ -51,6 +51,8 @@ def exception_csv(findings: list[dict[str, Any]]) -> bytes: "occurred_at", "audit_cutoff", "entity_type", "entity_id", "title", "summary", "rationale", "evidence_ids", "reviewer", "reviewed_at", "status", "conclusion", "notes", ] + if any("evidence_sha256" in finding for finding in findings): + fields += ["evidence_sha256", "review_stale_count"] rows = [] for finding in findings: review = finding.get("latest_review") or {} @@ -68,7 +70,9 @@ def exception_csv(findings: list[dict[str, Any]]) -> bytes: return _csv_bytes(rows, fields) -def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str: +def finding_workpaper( + finding: dict[str, Any], dataset_sha256: str, *, version: str = __version__, +) -> str: """A self-contained trail from control objective to source evidence and human decision.""" rule = finding.get("rule") or RULES[finding["control_id"]] review = finding.get("latest_review") or {} @@ -79,7 +83,7 @@ def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str: "", f"- Audit cutoff: `{DEMO_CUTOFF}`", f"- Dataset SHA-256: `{dataset_sha256}`", - f"- ControlTrace version: `{__version__}`", + f"- ControlTrace version: `{version}`", f"- Control: `{finding['control_id']}` — {finding['title']}", f"- Classification: `{finding['classification']}`; risk: `{finding['risk']}`", f"- System: `{finding['system']}`; subject: `{finding['entity_type']}:{finding['entity_id']}`", @@ -100,6 +104,8 @@ def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str: if isinstance(value, list): value = "; ".join(str(item) for item in value) lines.append(f"- **{label}:** {value}") + if "evidence_sha256" in finding: + lines.append(f"- Finding evidence SHA-256: `{finding['evidence_sha256']}`") lines += ["", "## Automated observation", "", finding["summary"], "", finding["rationale"]] lines += ["", "## Event timeline", ""] for event in finding.get("timeline", []): @@ -121,6 +127,15 @@ def finding_workpaper(finding: dict[str, Any], dataset_sha256: str) -> str: "### Review history", "", ] + if "review_bindings" in finding: + lines += [ + "Current decision applies only to this finding's evidence SHA-256. " + f"Historical decisions on different or unknown evidence: {finding['review_stale_count']}. " + "Their original source snapshots are not included in this bundle.", + "", + " " + _json({"review_evidence_sha256": finding["review_bindings"]}), + "", + ] for item in finding.get("review_history", []): lines.append(" " + _json(item)) lines += [ @@ -171,6 +186,9 @@ def workpaper_zip(db_path: str | Path, finding_ids: set[str] | None = None) -> b "reviews.json": _json({ finding["finding_id"]: finding["review_history"] for finding in findings }).encode("utf-8"), + "review_evidence.json": _json({ + finding["finding_id"]: finding["review_bindings"] for finding in findings + }).encode("utf-8"), "rules.json": _json(RULES).encode("utf-8"), "README.txt": ( "Synthetic ControlTrace workpapers. Check manifest.json hashes before using " @@ -189,7 +207,7 @@ def workpaper_zip(db_path: str | Path, finding_ids: set[str] | None = None) -> b }) manifest = { "notice": "Synthetic case study; no real enterprise audit data", - "bundle_format_version": 2, + "bundle_format_version": 3, "audit_cutoff": DEMO_CUTOFF, "generator_seed": int(meta.get("seed", DEFAULT_SEED)), "controltrace_version": __version__, diff --git a/src/controltrace/reviews.py b/src/controltrace/reviews.py index 2a4b7f6..1963102 100644 --- a/src/controltrace/reviews.py +++ b/src/controltrace/reviews.py @@ -2,6 +2,8 @@ from __future__ import annotations +import hashlib +import json from datetime import datetime from controltrace.data import parse_utc @@ -11,6 +13,42 @@ REVIEW_FIELDS = { "review_id", "finding_id", "reviewer", "reviewed_at", "status", "conclusion", "notes", } +REVIEW_ANNOTATIONS = { + "latest_review", "review_history", "review_status", "review_bindings", + "evidence_sha256", "review_revision", "review_stale_count", +} + + +def finding_fingerprint(finding: dict) -> str: + """Identify the exact automated observation, rule and linked source values.""" + payload = {key: value for key, value in finding.items() if key not in REVIEW_ANNOTATIONS} + encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, allow_nan=False).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def bind_review_history(finding: dict, history: list[dict], bindings: dict) -> dict: + """Only a decision made on the current evidence can become the current verdict. + + Unbound legacy reviews remain in history, but cannot claim to have assessed + the present evidence. Their original evidence is unknown. + """ + fingerprint = finding_fingerprint(finding) + applicable = [item for item in history if bindings.get(item["review_id"]) == fingerprint] + latest = applicable[-1] if applicable else None + revision = hashlib.sha256(json.dumps({ + "evidence": fingerprint, + "reviews": [(item["review_id"], bindings.get(item["review_id"])) for item in history], + }, sort_keys=True).encode("utf-8")).hexdigest() + return { + **finding, + "evidence_sha256": fingerprint, + "review_history": history, + "review_bindings": {item["review_id"]: bindings.get(item["review_id"]) for item in history}, + "latest_review": latest, + "review_status": latest["status"] if latest else "pending", + "review_stale_count": len(history) - len(applicable), + "review_revision": revision, + } def validate_review_decision( diff --git a/src/controltrace/store.py b/src/controltrace/store.py index a6e473a..aa7b10f 100644 --- a/src/controltrace/store.py +++ b/src/controltrace/store.py @@ -20,9 +20,9 @@ ) from controltrace.reviews import REVIEW_CONCLUSIONS as REVIEW_CONCLUSIONS from controltrace.reviews import REVIEW_STATUSES as REVIEW_STATUSES -from controltrace.reviews import validate_review_decision +from controltrace.reviews import bind_review_history, finding_fingerprint, validate_review_decision -_OTHER_TABLES = ("meta", "findings", "reviews") +_OTHER_TABLES = ("meta", "findings", "reviews", "review_evidence") ALL_TABLES = SOURCE_TABLES + _OTHER_TABLES _REVIEW_ORDER_BY = "CAST(reviewed_at AS TIMESTAMPTZ), review_id" @@ -39,6 +39,27 @@ def _rows(cursor: duckdb.DuckDBPyConnection) -> list[dict]: return [dict(zip(names, values, strict=True)) for values in cursor.fetchall()] +def _ensure_review_evidence(con: duckdb.DuckDBPyConnection) -> None: + # Do not backfill old decisions from today's finding: their original + # observation may have changed before this migration. + con.execute(""" + CREATE TABLE IF NOT EXISTS review_evidence ( + review_id VARCHAR PRIMARY KEY, + finding_sha256 VARCHAR NOT NULL + ) + """) + if con.execute( + "SELECT 1 FROM information_schema.tables WHERE table_name = 'findings'" + ).fetchone(): + con.execute( + "ALTER TABLE findings ADD COLUMN IF NOT EXISTS review_write_revision BIGINT DEFAULT 0" + ) + + +def _review_bindings(con: duckdb.DuckDBPyConnection) -> dict[str, str]: + return dict(con.execute("SELECT review_id, finding_sha256 FROM review_evidence").fetchall()) + + def initialize_db(path: str | Path, seed: int | None = None) -> None: """Initialize once. Rerunning with the same seed never resets review records. @@ -48,6 +69,7 @@ def initialize_db(path: str | Path, seed: int | None = None) -> None: """ con = _connect(path) try: + _ensure_review_evidence(con) con.execute("CREATE TABLE IF NOT EXISTS meta (key VARCHAR PRIMARY KEY, value VARCHAR NOT NULL)") prior = con.execute("SELECT value FROM meta WHERE key = 'seed'").fetchone() if prior: @@ -85,7 +107,8 @@ def initialize_db(path: str | Path, seed: int | None = None) -> None: entity_type VARCHAR NOT NULL, entity_id VARCHAR NOT NULL, payload_json VARCHAR NOT NULL, - refreshed_at VARCHAR NOT NULL + refreshed_at VARCHAR NOT NULL, + review_write_revision BIGINT DEFAULT 0 ) """) con.execute(""" @@ -116,7 +139,9 @@ def get_table_rows(path: str | Path, table: str) -> list[dict]: """Read a whitelisted table as JSON-serializable dictionaries.""" if table not in ALL_TABLES: raise ValueError(f"Unknown table: {table}") - order_key = PRIMARY_KEYS.get(table, {"meta": "key", "findings": "finding_id"}.get(table)) + order_key = PRIMARY_KEYS.get(table, { + "meta": "key", "findings": "finding_id", "review_evidence": "review_id", + }.get(table)) order_by = ( f' ORDER BY "{order_key}"' if order_key else f" ORDER BY {_REVIEW_ORDER_BY}" ) @@ -221,14 +246,12 @@ def list_findings(path: str | Path) -> list[dict]: "ORDER BY control_id, occurred_at, entity_id, issue_code" )) history = _review_history(con, [row["finding_id"] for row in raw]) + bindings = _review_bindings(con) findings = [] for row in raw: finding = json.loads(row["payload_json"]) reviews = history.get(row["finding_id"], []) - finding["latest_review"] = reviews[-1] if reviews else None - finding["review_history"] = reviews - finding["review_status"] = reviews[-1]["status"] if reviews else "pending" - findings.append(finding) + findings.append(bind_review_history(finding, reviews, bindings)) return findings finally: con.close() @@ -245,10 +268,7 @@ def get_finding(path: str | Path, finding_id: str) -> dict | None: return None finding = json.loads(raw[0]) reviews = _review_history(con, [finding_id]).get(finding_id, []) - finding["latest_review"] = reviews[-1] if reviews else None - finding["review_history"] = reviews - finding["review_status"] = reviews[-1]["status"] if reviews else "pending" - return finding + return bind_review_history(finding, reviews, _review_bindings(con)) finally: con.close() @@ -261,30 +281,61 @@ def save_review( conclusion: str | None = None, notes: str = "", reviewed_at: str | None = None, + *, + expected_revision: str | None = None, ) -> dict: - """Append an analyst decision; never rewrite earlier reviews or source evidence.""" + """Append a decision bound to current evidence, optionally rejecting a stale form. + + Callers editing a previously displayed finding should pass its + ``review_revision``. Validation and both inserts share one transaction. + """ if reviewed_at is None: reviewed_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") reviewed_time = validate_review_decision(reviewer, status, conclusion, notes, reviewed_at) reviewed_at = reviewed_time.isoformat().replace("+00:00", "Z") con = _connect(path) try: - if con.execute( - "SELECT 1 FROM findings WHERE finding_id = ?", [finding_id] - ).fetchone() is None: - raise ValueError(f"Unknown finding: {finding_id}") - review = { - "review_id": f"REV-{uuid4().hex[:12].upper()}", - "finding_id": finding_id, - "reviewer": reviewer.strip(), - "reviewed_at": reviewed_at, - "status": status, - "conclusion": conclusion, - "notes": notes.strip(), - } - con.execute( - "INSERT INTO reviews VALUES (?, ?, ?, ?, ?, ?, ?)", list(review.values()) - ) - return review + con.execute("BEGIN TRANSACTION") + try: + raw = con.execute( + "SELECT payload_json FROM findings WHERE finding_id = ?", [finding_id] + ).fetchone() + if raw is None: + raise ValueError(f"Unknown finding: {finding_id}") + finding = json.loads(raw[0]) + history = _review_history(con, [finding_id]).get(finding_id, []) + current = bind_review_history(finding, history, _review_bindings(con)) + if expected_revision is not None and expected_revision != current["review_revision"]: + raise ValueError("Finding evidence or review changed; reload before saving") + # A real row update is essential: a no-op can be optimized away, + # allowing two snapshot readers to both pass the revision check. + con.execute( + "UPDATE findings SET review_write_revision = review_write_revision + 1 " + "WHERE finding_id = ?", [finding_id] + ) + review = { + "review_id": f"REV-{uuid4().hex[:12].upper()}", + "finding_id": finding_id, + "reviewer": reviewer.strip(), + "reviewed_at": reviewed_at, + "status": status, + "conclusion": conclusion, + "notes": notes.strip(), + } + con.execute( + "INSERT INTO reviews VALUES (?, ?, ?, ?, ?, ?, ?)", list(review.values()) + ) + con.execute( + "INSERT INTO review_evidence VALUES (?, ?)", + [review["review_id"], finding_fingerprint(finding)], + ) + con.execute("COMMIT") + return review + except duckdb.TransactionException as error: + con.execute("ROLLBACK") + raise ValueError("Finding evidence or review changed; reload before saving") from error + except Exception: + con.execute("ROLLBACK") + raise finally: con.close() diff --git a/src/controltrace/verify.py b/src/controltrace/verify.py index 589a41c..a142e3c 100644 --- a/src/controltrace/verify.py +++ b/src/controltrace/verify.py @@ -7,6 +7,7 @@ import io import json import math +import re import zipfile from pathlib import Path from typing import Any @@ -14,7 +15,7 @@ from controltrace import __version__ from controltrace.data import DEMO_CUTOFF, SOURCE_TABLES from controltrace.exports import _csv_bytes, _json, exception_csv, finding_workpaper -from controltrace.reviews import REVIEW_FIELDS, validate_review_decision +from controltrace.reviews import REVIEW_FIELDS, bind_review_history, validate_review_decision from controltrace.rules import RULES, evaluate @@ -59,8 +60,10 @@ def verify_bundle(path: str | Path) -> dict[str, Any]: _require("manifest.json" in names, "Bundle has no manifest.json") manifest = _read_json(archive, "manifest.json") _require(isinstance(manifest, dict), "Manifest must be a JSON object") - _require(manifest.get("bundle_format_version") == 2, + bundle_format = manifest.get("bundle_format_version") + _require(bundle_format in (2, 3), "Unsupported bundle format; export again with this ControlTrace version") + evidence_bound = bundle_format == 3 expected_files = manifest.get("files_sha256") _require(isinstance(expected_files, dict), "Manifest has no complete file hash list") _require(set(expected_files) == set(names) - {"manifest.json"}, @@ -83,7 +86,8 @@ def verify_bundle(path: str | Path) -> dict[str, Any]: "Dataset fingerprint does not match source hashes") _require(manifest.get("audit_cutoff") == DEMO_CUTOFF, "Bundle cutoff does not match the current control rules") - _require(manifest.get("controltrace_version") == __version__, + bundle_version = manifest.get("controltrace_version") + _require(bundle_version == (__version__ if evidence_bound else "0.2.7"), "Bundle was created by a different ControlTrace version") _require(_read_json(archive, "rules.json") == RULES, "Rule catalog differs from the installed ControlTrace version") @@ -120,6 +124,12 @@ def verify_bundle(path: str | Path) -> dict[str, Any]: histories = _read_json(archive, "reviews.json") _require(isinstance(histories, dict) and set(histories) == set(selected_ids), "Review history does not match selected findings") + bindings = {} + if evidence_bound: + _require("review_evidence.json" in expected_files, "Bundle has no review evidence bindings") + bindings = _read_json(archive, "review_evidence.json") + _require(isinstance(bindings, dict) and set(bindings) == set(selected_ids), + "Review evidence bindings do not match selected findings") selected_findings = [] review_ids = set() for finding_id in selected_ids: @@ -153,10 +163,22 @@ def verify_bundle(path: str | Path) -> dict[str, Any]: "review_history": history, "latest_review": history[-1] if history else None, } + if evidence_bound: + finding_bindings = bindings[finding_id] + _require(isinstance(finding_bindings, dict) + and set(finding_bindings) == {item["review_id"] for item in history}, + f"Review evidence bindings do not match history: {finding_id}") + _require(all(value is None or isinstance(value, str) + and re.fullmatch(r"[0-9a-f]{64}", value) + for value in finding_bindings.values()), + f"Invalid review evidence fingerprint: {finding_id}") + finding = bind_review_history(finding, history, finding_bindings) selected_findings.append(finding) workpaper_name = f"workpapers/{finding_id}.md" _require(workpaper_name in expected_files, f"Missing workpaper for {finding_id}") - _require(archive.read(workpaper_name) == finding_workpaper(finding, dataset_hash).encode("utf-8"), + _require(archive.read(workpaper_name) == finding_workpaper( + finding, dataset_hash, version=bundle_version, + ).encode("utf-8"), f"Workpaper differs from replayed evidence or review: {finding_id}") _require(archive.read("findings.csv") == exception_csv(selected_findings), "Findings CSV differs from replayed evidence or review") @@ -167,4 +189,5 @@ def verify_bundle(path: str | Path) -> dict[str, Any]: "replayed_findings": len(all_ids), "exported_findings": len(selected_ids), "scope": scope, + "review_evidence_bound": evidence_bound, } diff --git a/tests/test_app.py b/tests/test_app.py index 5c44534..52d7b8a 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -4,7 +4,7 @@ from streamlit.testing.v1 import AppTest -from controltrace.store import get_table_rows +from controltrace.store import get_table_rows, list_findings, save_review APP_PATH = Path(__file__).parents[1] / "src" / "controltrace" / "app.py" @@ -64,3 +64,24 @@ def test_filter_changes_reset_table_selection_without_crashing(tmp_path, monkeyp app.multiselect[0].set_value([]).run() assert not app.exception assert any("No observations match" in item.value for item in app.info) + + +def test_stale_form_requires_explicit_reload_before_saving(tmp_path, monkeypatch): + db_path = tmp_path / "case.duckdb" + monkeypatch.setenv("CONTROLTRACE_DB", str(db_path)) + app = AppTest.from_file(str(APP_PATH)).run(timeout=30) + app.text_input[0].set_value("Reviewer A").run() + app.text_area[0].set_value("Unsaved original form").run() + finding = list_findings(db_path)[0] + save_review(db_path, finding["finding_id"], "Reviewer B", notes="New current review") + app.run() + assert not app.exception + assert any("changed while you were editing" in item.value for item in app.warning) + save_button = next(button for button in app.button if button.label == "Save review") + assert save_button.disabled + assert len(get_table_rows(db_path, "reviews")) == 1 + next(button for button in app.button if button.label == "Reload current finding").click().run() + assert not app.exception + assert app.text_input[0].value == "Reviewer B" + assert app.text_area[0].value == "New current review" + assert not next(button for button in app.button if button.label == "Save review").disabled diff --git a/tests/test_review_evidence.py b/tests/test_review_evidence.py new file mode 100644 index 0000000..8abbdd2 --- /dev/null +++ b/tests/test_review_evidence.py @@ -0,0 +1,270 @@ +"""A stable finding ID cannot carry a verdict across different source evidence.""" + +from __future__ import annotations + +import hashlib +import io +import json +import threading +import zipfile +from concurrent.futures import ThreadPoolExecutor + +import duckdb +import pytest + +from controltrace import store +from controltrace.data import PRIMARY_KEYS +from controltrace.exports import _json, exception_csv, finding_workpaper, workpaper_zip +from controltrace.reviews import REVIEW_ANNOTATIONS +from controltrace.rules import run_tests +from controltrace.store import get_finding, get_table_rows, initialize_db, save_review +from controltrace.verify import BundleVerificationError, verify_bundle + + +def _change_linked_source(db_path, finding): + record = finding["related_records"][0] + table = record["table"] + with duckdb.connect(str(db_path)) as connection: + connection.execute( + f'UPDATE "{table}" SET source_system = ? WHERE "{PRIMARY_KEYS[table]}" = ?', + ["Corrected synthetic extract", record["id"]], + ) + + +def _closed_finding(db_path): + initialize_db(db_path) + finding = run_tests(db_path)[0] + save_review( + db_path, finding["finding_id"], "Reviewer A", "closed", "confirmed_exception", + "Checked the original linked source", expected_revision=finding["review_revision"], + ) + return get_finding(db_path, finding["finding_id"]) + + +def test_identical_rerun_keeps_review_and_revision(tmp_path): + db_path = tmp_path / "case.duckdb" + finding = _closed_finding(db_path) + run_tests(db_path) + assert get_finding(db_path, finding["finding_id"]) == finding + + +def test_changed_evidence_reopens_same_id_without_losing_history(tmp_path): + db_path = tmp_path / "case.duckdb" + original = _closed_finding(db_path) + _change_linked_source(db_path, original) + run_tests(db_path) + changed = get_finding(db_path, original["finding_id"]) + assert changed["evidence_sha256"] != original["evidence_sha256"] + assert changed["review_history"] == original["review_history"] + assert changed["review_bindings"] == original["review_bindings"] + assert changed["latest_review"] is None + assert changed["review_status"] == "pending" + assert changed["review_stale_count"] == 1 + with pytest.raises(ValueError, match="reload before saving"): + save_review( + db_path, original["finding_id"], "Reviewer B", "closed", "false_positive", + "This form still displayed the original extract", + expected_revision=original["review_revision"], + ) + assert get_table_rows(db_path, "reviews") == original["review_history"] + replacement = save_review( + db_path, changed["finding_id"], "Reviewer B", "closed", "false_positive", + "Reassessed the corrected extract", expected_revision=changed["review_revision"], + ) + current = get_finding(db_path, changed["finding_id"]) + assert current["latest_review"] == replacement + assert current["review_bindings"][replacement["review_id"]] == changed["evidence_sha256"] + assert current["review_stale_count"] == 1 + + +def test_corrected_approval_time_reopens_only_findings_with_changed_evidence(tmp_path): + db_path = tmp_path / "case.duckdb" + unrelated = _closed_finding(db_path) + finding_id = "F-2C5FD7D5834D" + finding = get_finding(db_path, finding_id) + save_review( + db_path, finding_id, "Reviewer A", "closed", "confirmed_exception", + "CH002 approval is later than deployment D002", + expected_revision=finding["review_revision"], + ) + original = get_finding(db_path, finding_id) + with duckdb.connect(str(db_path)) as connection: + connection.execute( + "UPDATE change_tickets SET approved_at = ? WHERE ticket_id = ?", + ["2025-06-21T13:00:00Z", "CH002"], + ) + run_tests(db_path) + changed = get_finding(db_path, finding_id) + assert changed["issue_code"] == original["issue_code"] == "deployment_approval_late" + assert changed["evidence_sha256"] != original["evidence_sha256"] + assert changed["review_status"] == "pending" + assert changed["review_history"] == original["review_history"] + assert get_finding(db_path, unrelated["finding_id"]) == unrelated + + +def test_stale_review_rejected_after_another_reviewer_saves(tmp_path): + db_path = tmp_path / "case.duckdb" + original = _closed_finding(db_path) + save_review(db_path, original["finding_id"], "Reviewer B", notes="Requested new evidence") + with pytest.raises(ValueError, match="reload before saving"): + save_review( + db_path, original["finding_id"], "Reviewer C", notes="A stale response", + expected_revision=original["review_revision"], + ) + assert len(get_table_rows(db_path, "reviews")) == 2 + + +def test_simultaneous_reviewers_cannot_both_commit_same_revision(tmp_path, monkeypatch): + db_path = tmp_path / "case.duckdb" + original = _closed_finding(db_path) + barrier = threading.Barrier(2) + original_bindings = store._review_bindings + + def synchronized_bindings(connection): + result = original_bindings(connection) + if threading.current_thread() is not threading.main_thread(): + barrier.wait(timeout=10) + return result + + monkeypatch.setattr(store, "_review_bindings", synchronized_bindings) + + def submit(reviewer): + try: + save_review( + db_path, original["finding_id"], reviewer, notes="Concurrent response", + expected_revision=original["review_revision"], + ) + return "saved" + except ValueError as error: + assert "reload before saving" in str(error) + return "conflict" + + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(submit, ["Reviewer B", "Reviewer C"])) + assert sorted(results) == ["conflict", "saved"] + assert len(get_table_rows(db_path, "reviews")) == 2 + assert len(get_table_rows(db_path, "review_evidence")) == 2 + + +def test_binding_failure_rolls_back_review_insert(tmp_path): + db_path = tmp_path / "case.duckdb" + original = _closed_finding(db_path) + before = get_table_rows(db_path, "reviews") + with duckdb.connect(str(db_path)) as connection: + connection.execute("DROP TABLE review_evidence") + connection.execute(""" + CREATE TABLE review_evidence ( + review_id VARCHAR PRIMARY KEY, finding_sha256 VARCHAR CHECK (FALSE) + ) + """) + with pytest.raises(duckdb.ConstraintException): + save_review(db_path, original["finding_id"], "Reviewer B", notes="Must remain atomic") + assert get_table_rows(db_path, "reviews") == before + assert get_table_rows(db_path, "review_evidence") == [] + + +def test_legacy_database_retains_history_without_inventing_evidence_bindings(tmp_path): + db_path = tmp_path / "legacy.duckdb" + original = _closed_finding(db_path) + with duckdb.connect(str(db_path)) as connection: + connection.execute("DROP TABLE review_evidence") + connection.execute("ALTER TABLE findings DROP COLUMN review_write_revision") + initialize_db(db_path) + migrated = get_finding(db_path, original["finding_id"]) + assert migrated["review_history"] == original["review_history"] + assert list(migrated["review_bindings"].values()) == [None] + assert migrated["latest_review"] is None + assert migrated["review_status"] == "pending" + assert migrated["review_stale_count"] == 1 + + +def _save_members(path, members): + manifest = json.loads(members["manifest.json"]) + manifest["files_sha256"] = { + name: hashlib.sha256(value).hexdigest() + for name, value in members.items() if name != "manifest.json" + } + members["manifest.json"] = _json(manifest).encode("utf-8") + with zipfile.ZipFile(path, "w") as archive: + for name, value in members.items(): + archive.writestr(name, value) + + +def test_new_bundle_replays_pending_after_evidence_changes(tmp_path): + db_path = tmp_path / "case.duckdb" + original = _closed_finding(db_path) + _change_linked_source(db_path, original) + run_tests(db_path) + bundle = workpaper_zip(db_path, {original["finding_id"]}) + path = tmp_path / "pending.zip" + path.write_bytes(bundle) + assert verify_bundle(path)["review_evidence_bound"] is True + with zipfile.ZipFile(io.BytesIO(bundle)) as archive: + members = {name: archive.read(name) for name in archive.namelist()} + assert b"pending" in members["findings.csv"] + assert b"Reviewer A" not in members["findings.csv"] + # A fully rehashed CSV/workpaper claiming the old closed decision must + # still disagree with independently replayed evidence applicability. + current = get_finding(db_path, original["finding_id"]) + forged = {**current, "latest_review": original["latest_review"]} + members["findings.csv"] = exception_csv([forged]) + manifest = json.loads(members["manifest.json"]) + members[f"workpapers/{original['finding_id']}.md"] = finding_workpaper( + forged, manifest["dataset_sha256"], + ).encode("utf-8") + forged_path = tmp_path / "forged-closed.zip" + _save_members(forged_path, members) + with pytest.raises(BundleVerificationError, match="Workpaper differs"): + verify_bundle(forged_path) + + +def test_legacy_v2_bundle_is_verified_with_explicit_unbound_status(tmp_path): + db_path = tmp_path / "case.duckdb" + finding = _closed_finding(db_path) + bundle = workpaper_zip(db_path, {finding["finding_id"]}) + with zipfile.ZipFile(io.BytesIO(bundle)) as archive: + members = {name: archive.read(name) for name in archive.namelist()} + legacy = {key: value for key, value in finding.items() if key not in REVIEW_ANNOTATIONS} + legacy["review_history"] = finding["review_history"] + legacy["latest_review"] = finding["latest_review"] + members.pop("review_evidence.json") + manifest = json.loads(members["manifest.json"]) + manifest["bundle_format_version"] = 2 + manifest["controltrace_version"] = "0.2.7" + members["manifest.json"] = _json(manifest).encode("utf-8") + members["findings.csv"] = exception_csv([legacy]) + members[f"workpapers/{finding['finding_id']}.md"] = finding_workpaper( + legacy, manifest["dataset_sha256"], version="0.2.7", + ).encode("utf-8") + path = tmp_path / "legacy-v2.zip" + _save_members(path, members) + assert verify_bundle(path)["review_evidence_bound"] is False + + +@pytest.mark.parametrize("binding", ["a" * 63, "G" * 64, 42, [], {}]) +def test_verifier_rejects_invalid_review_binding_after_rehash(tmp_path, binding): + db_path = tmp_path / "case.duckdb" + finding = _closed_finding(db_path) + bundle = workpaper_zip(db_path, {finding["finding_id"]}) + with zipfile.ZipFile(io.BytesIO(bundle)) as archive: + members = {name: archive.read(name) for name in archive.namelist()} + review_id = finding["latest_review"]["review_id"] + members["review_evidence.json"] = _json({ + finding["finding_id"]: {review_id: binding}, + }).encode("utf-8") + path = tmp_path / "invalid-binding.zip" + _save_members(path, members) + with pytest.raises(BundleVerificationError, match="Invalid review evidence fingerprint"): + verify_bundle(path) + + +def test_review_of_another_finding_does_not_invalidate_form_revision(tmp_path): + db_path = tmp_path / "case.duckdb" + initialize_db(db_path) + first, second = run_tests(db_path)[:2] + save_review(db_path, second["finding_id"], "Reviewer B", notes="Reviewing another case") + save_review( + db_path, first["finding_id"], "Reviewer A", notes="My case is still unchanged", + expected_revision=first["review_revision"], + ) + assert len(get_table_rows(db_path, "reviews")) == 2 diff --git a/uv.lock b/uv.lock index b232cac..e763dd3 100644 --- a/uv.lock +++ b/uv.lock @@ -220,7 +220,7 @@ wheels = [ [[package]] name = "controltrace" -version = "0.2.7" +version = "0.2.8" source = { editable = "." } dependencies = [ { name = "duckdb" },