可验证报告服务端:GET /api/runs/{id}/bundle.html 把已封存运行的报告导出成自包含的 HTML,浏览器断网也能核对

- bundle/build.py(纯函数):封存范围内每条事件带 seq、类型、节点和 data_sha256(与 runner 封存同一算法),浏览器据此重算清单;
  白名单事件(运行开始、报告自查、口径卡升版、出具判定、报告和口径卡的节点完成、数据源没有遮罩的查询返回)另带规范字符串,
  运行里有带遮罩的快照时,成果节点和运行结束事件的预览可能带出查询结果,也只给哈希;报告文档和它引用的口径卡、查询快照按原字节
  base64 嵌入,遮罩快照(查询时记下的或数据源现在设的)只给 id;封存之后追加的事件单列、不进清单
- template.html + verifier.js:样式和脚本内联、零外链,内容安全策略按哈希只放行核对脚本;数据块转义 < > & 和 U+2028/2029。
  浏览器逐项核对事件内容、封存清单、工件哈希、引用链、报告里每个数字(照搬 cell_value / same_value 判等),
  crypto.subtle 拿不到时用自检过的纯 JS SHA-256;醒目写出封存哈希,并写明只证明内部一致、真伪要去系统里对照
- 接口复用证据接口的封存判定:未封存、或运行还没结束(接着跑的运行留着旧封存,跑完会换掉清单哈希)回 409,封存核对不一致回 409;
  附件名「<运行名>-可验证报告.html」按 RFC 5987 编码;只读,不追加事件
- artifact_store 加 read_bytes:原样取回工件字节,默认复验哈希,verify=False 交给浏览器去核
This commit is contained in:
ThearchyHelios
2026-10-05 03:58:02 +08:00
parent 35bb23d372
commit a0016493de
8 changed files with 2300 additions and 3 deletions
+66 -3
View File
@@ -1,10 +1,73 @@
"""可验证报告:把一次运行的报告和证据导出成可独立核对的 HTML 包。
"""可验证报告:GET /api/runs/{run_id}/bundle.html,把一次已封存运行的报告导出成一个自包含的 HTML 文件。
目前只登记了路由前缀,接口由可验证报告的工作包补上。路径挂在 /api/runs/{run_id}/… 下面;
注意 runs.py 里已经有 /{run_id}/verify(封存复验),别撞名。
拿到文件的人不连服务、不装任何东西,在浏览器里就能核对每个数字的出处、运行记录和工件有没有被改过
(页面怎么核见 bundle/verifier.js,数据怎么拼见 bundle/build.py)。
只给已封存、而且封存核对一致的运行:封存判定照搬证据接口(api/evidence._sealed),不另写一套。没封存的
运行事件还会变,导出去的东西马上就过时;封存核对不一致的,导出去的页面只会把一条断掉的链画成核对通过。
另外要求运行已经结束(成功、失败、取消):失败后接着跑的运行留着上一次的封存,跑完会重新封存、换掉清单哈希,
早年的运行停在审批时也封存过——这时导出去的文件,封存哈希和系统里之后记的对不上,页面上「去系统里对照」
那一步就作废了。
只读:不追加任何事件,不改运行记录。路径挂在 /api/runs/{run_id}/… 下面,runs.py 里已经有 /{run_id}/verify,
这里叫 bundle.html,不撞名。
"""
from __future__ import annotations
import asyncio
from datetime import datetime, timezone
from fastapi import APIRouter
from fastapi.responses import Response
from sqlalchemy import select
from app.api.coded import CodedHTTPException
from app.api.evidence import RUN_NOT_FOUND, _mask_fingerprint, _sealed
from app.bundle.build import attachment_name, build_bundle, content_disposition, render_html
from app.core.artifact_store import read_bytes
from app.db.base import SessionLocal
from app.db.models import Run, RunEvent
router = APIRouter(prefix="/api/runs", tags=["bundle"])
BUNDLE_UNSEALED = "bundle_unsealed"
BUNDLE_SEAL_BROKEN = "bundle_seal_broken"
#: 运行结束的状态:封存是最后一次的,之后不会再换(除非失败的运行又被接着跑,那是另一次封存)
_FINISHED = frozenset({"succeeded", "failed", "cancelled"})
UNSEALED = "运行尚未结束封存,结束后才能导出"
SEAL_BROKEN = "封存核对不一致:{why}。不能导出可验证报告,请先在记录页核对这次运行"
@router.get("/{run_id}/bundle.html")
async def export_bundle(run_id: str) -> Response:
"""可验证报告的 HTML 文件,作为附件下载(附件名「<运行名>-可验证报告.html」)。"""
sealed = await _sealed(run_id)
if not sealed.sealed or sealed.status not in _FINISHED:
raise CodedHTTPException(409, UNSEALED, BUNDLE_UNSEALED)
if not sealed.trusted:
why = str(sealed.verdict.get("message") or "封存后有事件被修改、删除或插入").rstrip("。")
raise CodedHTTPException(409, SEAL_BROKEN.format(why=why), BUNDLE_SEAL_BROKEN)
bound = sealed.seal["manifest_seq"]
async with SessionLocal() as session:
run = await session.get(Run, run_id)
if run is None: # 两次读之间被删掉了
raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND)
info = {"id": run.id, "workflow_id": run.workflow_id, "workflow_name": run.workflow_name,
"version": run.version, "version_hash": run.version_hash, "run_class": run.run_class,
"status": run.status, "started_at": run.started_at, "finished_at": run.finished_at}
# 封存之后追加的事件(按需裁判、审批质询):_sealed 只留了封存范围内的,这几条另取,单列在页面上
later = [tuple(r) for r in await session.execute(
select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data)
.where(RunEvent.run_id == run_id, RunEvent.seq > bound).order_by(RunEvent.seq))]
# 数据源现在设的遮罩:事后加的遮罩同样生效,和证据面板一个道理
masks = await _mask_fingerprint()
exported_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
# 读工件、算哈希、base64 是同步的活,快照大的时候不该占着事件循环
page = await asyncio.to_thread(lambda: render_html(build_bundle(
info, [*sealed.events, *later], manifest_seq=bound, manifest_hash=str(sealed.manifest_hash or ""),
legacy=bool(sealed.seal.get("legacy")), graph=sealed.graph, source_masks=masks,
read=lambda artifact: read_bytes(artifact, verify=False), exported_at=exported_at)))
return Response(page, media_type="text/html; charset=utf-8", headers={
"Content-Disposition": content_disposition(attachment_name(info), run_id=run_id),
# 导出的是这一刻的封存件,不该被哪一层缓存下来、换给下一个请求
"Cache-Control": "no-store",
})
+8
View File
@@ -0,0 +1,8 @@
"""可验证报告:把一次已封存运行的报告导出成一个自包含的 HTML 文件,拿到文件的人在浏览器里就能核对。
- build.py:从事件、工件拼出页面数据,填进模板(纯函数,不碰数据库)
- template.html:页面骨架和样式
- verifier.js:浏览器端的核对和渲染(原生 JS,无依赖)
接口在 api/bundle.py(GET /api/runs/{run_id}/bundle.html)。
"""
+394
View File
@@ -0,0 +1,394 @@
"""可验证报告的构建:把一次已封存运行的报告、证据和运行记录打成一个自包含的 HTML 文件。
拿到文件的人不连服务、不装任何东西,在浏览器里就能核对(verifier.js):
- 事件:封存范围内(seq ≤ manifest_seq)的每一条都带 {seq, type, node_id, data_sha256},data_sha256 是
sha256(canonical_json(data)),和 runner 封存时算 manifest_hash 用的是同一个算法,浏览器按同样的拼法重算
清单哈希、和封存哈希比。白名单里的事件另带服务端算好的规范字符串 canonical(逐字等于 canonical_json(data)),
浏览器先核它的哈希,再从里面读出报告文档、口径卡、查询快照的 id。其余事件只给哈希:工具返回、节点产出的
预览里可能有遮罩列的原值和提示词。
- 工件:每份报告文档(各报告节点最后一次自查记下的 doc_artifact)、它引用到的口径卡和查询快照,按原字节
base64 嵌入,浏览器重算 sha256 和 id 比;带遮罩的快照不嵌内容、只给 id。没被报告引用的工件不嵌。
- 封存之后追加的事件(按需裁判、审批质询这类批注)单列:类型、序号、内容摘要,不进清单。
它只证明「内部一致」:数字、文档、运行记录彼此对得上,而且没被改过。真伪要拿页面上的封存哈希和系统里这次
运行的封存核对(GET /api/runs/{id}/verify)对照,页面上醒目地写着这一点。
主体是纯函数:不碰数据库,工件经 read 取原字节,数据源现在设的遮罩由调用方查好传进来(source_masks)。
"""
from __future__ import annotations
import base64
import hashlib
import html
import json
import re
from collections.abc import Callable, Iterable, Mapping, Sequence
from datetime import datetime
from functools import lru_cache
from pathlib import Path
from typing import Any
from urllib.parse import quote
from app.core.artifact_store import canonical_json, content_hash
from app.core.events import EventType
from app.engine.evidence import iter_units
from app.engine.merge_query import MERGE_SOURCE
from app.engine.schema import NodeType
SCHEMA = "agentlab.bundle/1"
#: 不论哪个节点都带规范字符串的事件:运行开始(实际生效的运行参数)、报告自查(报告文档的 id)、
#: 口径卡升版处置、出具判定(档位)
_SHOWN_EVENTS = frozenset({EventType.RUN_STARTED, EventType.REPORT_CHECKED, EventType.CALIBER_UPGRADE,
EventType.ISSUANCE})
#: node.finished 带规范字符串的节点:报告(预览就是嵌进来的那份文档的正文)、口径卡(指标集的 id 记在台账里)
_SHOWN_NODES = frozenset({NodeType.REPORT, NodeType.METRICS})
#: 目录条目的种类 → 嵌进来的工件种类。运行输入、检索、表和字段的条目不嵌:前者没有工件,后几样不在核对范围里
_CITED_KINDS = {"metric": "metric_set", "query": "query"}
#: 没挂在工作流上的运行(画布、问数据当场搭的):老数据叫「临时图」,新的叫「未保存的工作流」,也可能是空的。
#: 和前端 lib/terms.ts 的 runName 同一个规则
UNSAVED_NAME = "未保存的工作流"
_UNSAVED_NAMES = frozenset({"", "临时图", UNSAVED_NAME})
FILE_SUFFIX = "-可验证报告.html"
#: 文件名里不能出现的字(Windows 最严,按它来)和控制字符
_BAD_FILENAME = re.compile(r'[\\/:*?"<>|\x00-\x1f\x7f]')
_NAME_MAX = 80
Reader = Callable[[str], "bytes | None"]
# --------------------------------------------------------------------------
# 工件:原字节、解析、遮罩
# --------------------------------------------------------------------------
class _Store:
"""这一次构建里读过的工件:原字节、解析出来的内容、查询快照带不带遮罩。每份只读一次盘。"""
def __init__(self, read: Reader, source_masks: Mapping[str, Any]) -> None:
self._read = read
self._source_masks = source_masks
self._raw: dict[str, bytes | None] = {}
self._json: dict[str, Any] = {}
self._masked: dict[str, bool] = {}
def raw(self, artifact: str) -> bytes | None:
if artifact not in self._raw:
try:
data = self._read(artifact)
except (OSError, ValueError):
data = None
self._raw[artifact] = data if isinstance(data, (bytes, bytearray)) else None
return self._raw[artifact]
def content(self, artifact: str) -> Any:
"""解析出来的内容,读不出来、不是 JSON 的是 None。不看哈希:被改过的快照也要知道它记的遮罩。"""
if artifact not in self._json:
raw = self.raw(artifact)
try:
self._json[artifact] = json.loads(raw) if raw is not None else None
except ValueError:
self._json[artifact] = None
return self._json[artifact]
def intact(self, artifact: str) -> bool:
raw = self.raw(artifact)
return raw is not None and content_hash(bytes(raw)) == artifact
def masked(self, artifact: str) -> bool:
"""这份查询快照算不算带遮罩:查询当时记下了遮罩列(mask_columns),或者它的数据源现在设了遮罩
(合并查询看它合并的那几个数据源)。比证据面板粗:面板只遮那几列,这里整份不随文件附带——文件会
离开系统,遮罩只减少暴露,宁可多挡。读不出来的快照不知道带没带遮罩,按带遮罩算。"""
if artifact not in self._masked:
self._masked[artifact] = _snapshot_masked(self.content(artifact), self._source_masks)
return self._masked[artifact]
def _snapshot_masked(content: Any, source_masks: Mapping[str, Any]) -> bool:
if not isinstance(content, dict):
return True
recorded = content.get("mask_columns")
if isinstance(recorded, list) and any(isinstance(c, str) and c.strip() for c in recorded):
return True
sources = content.get("sources") if content.get("source") == MERGE_SOURCE else [content.get("source")]
return any(source_masks.get(str(name or "")) for name in sources or [])
def _query_artifacts(events: Iterable[tuple[int, str, str | None, Any]]) -> list[str]:
"""封存范围内交回过的查询快照:tool.end / merge.end 的 query_artifact,节点台账里的 query 条目。"""
out: list[str] = []
for _, etype, _, data in events:
if not isinstance(data, dict):
continue
if etype in (EventType.TOOL_END, EventType.MERGE_END) and isinstance(data.get("query_artifact"), str):
out.append(data["query_artifact"])
elif etype == EventType.NODE_FINISHED:
out += [str(e["artifact"]) for e in data.get("evidence") or []
if isinstance(e, dict) and e.get("kind") == "query" and e.get("artifact")]
return list(dict.fromkeys(a for a in out if a))
# --------------------------------------------------------------------------
# 报告文档和它引用到的工件
# --------------------------------------------------------------------------
def _report_events(events: Iterable[tuple[int, str, str | None, Any]]) -> list[dict[str, Any]]:
"""每个报告节点最后一次自查记下的文档,按第一次出现的顺序。和证据接口(api/evidence._reports)同一个认法:
报告重写过的,认最后一次。"""
latest: dict[str, dict[str, Any]] = {}
for seq, etype, node_id, data in events:
if etype == EventType.REPORT_CHECKED and node_id and isinstance(data, dict) and data.get("doc_artifact"):
latest[node_id] = {"node_id": node_id, "doc_artifact": str(data["doc_artifact"]), "seq": seq}
return list(latest.values())
def _cited_artifacts(doc: dict[str, Any]) -> list[tuple[str, str, str | None]]:
"""报告正文引用到的口径卡和查询快照:(工件 id, 种类, 来源节点),按正文顺序、去重。
看每个解析成功的引用(行内的数字、单元格,句末的 [[see:]])在文档目录里指向的条目:指标条目指向口径卡,
查询条目指向快照。整表在组装时已经展开成每格一个单元格引用,不用另看。"""
catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {}
out: list[tuple[str, str, str | None]] = []
for _, unit in iter_units(doc):
if not isinstance(unit, dict):
continue
cites = [s.get("cite") for s in unit.get("segments") or [] if isinstance(s, dict)]
for cite in [*cites, *(unit.get("see") or [])]:
if not isinstance(cite, dict) or cite.get("status") != "resolved":
continue
entry = catalog.get(cite.get("alias"))
kind = _CITED_KINDS.get(entry.get("kind")) if isinstance(entry, dict) else None
if kind and isinstance(entry.get("artifact"), str) and entry["artifact"]:
owner = entry.get("node_id")
out.append((entry["artifact"], kind, str(owner) if owner else None))
return list(dict.fromkeys(out))
class _Artifacts:
"""嵌进页面的工件清单,按先后:报告文档在前,接着是它引用到的。同一份只列一次。"""
def __init__(self, store: _Store) -> None:
self.store = store
self.items: list[dict[str, Any]] = []
self._seen: set[str] = set()
def add(self, artifact: str, kind: str, node_id: str | None) -> None:
if artifact in self._seen:
return
self._seen.add(artifact)
item: dict[str, Any] = {"id": artifact, "kind": kind}
if node_id:
item["node_id"] = node_id
raw = self.store.raw(artifact)
if raw is None:
# 服务端的文件不在了:只给 id,浏览器照实说核对不了
item["missing"] = True
elif kind == "query" and self.store.masked(artifact):
item["masked"] = True
else:
# 原样嵌入,不在这里复验哈希:被改过的文件照实交出去,浏览器自己算出对不上
item["b64"] = base64.b64encode(bytes(raw)).decode("ascii")
self.items.append(item)
# --------------------------------------------------------------------------
# 事件
# --------------------------------------------------------------------------
def _shows_content(etype: str, node_id: str | None, data: Any, node_types: Mapping[str, str], store: _Store,
masked_run: bool) -> bool:
"""这条封存范围内的事件带不带规范字符串(白名单)。"""
if etype in _SHOWN_EVENTS:
return True
if etype == EventType.RUN_FINISHED:
# 运行结束事件里是成果:字段可以直接引用查询结果,运行里有带遮罩的快照时只给哈希
return not masked_run
if etype == EventType.NODE_FINISHED:
kind = node_types.get(node_id or "")
return kind in _SHOWN_NODES or (kind == NodeType.OUTPUT and not masked_run)
if etype == EventType.TOOL_END:
# 查询类:把查询快照的 id 和运行记录连起来。预览里是查询结果的前一截,数据源带遮罩的只给哈希
artifact = data.get("query_artifact") if isinstance(data, dict) else None
return isinstance(artifact, str) and bool(artifact) and not store.masked(artifact)
return False
def _clip(text: Any, limit: int = 60) -> str:
flat = re.sub(r"\s+", " ", str(text or "")).strip()
return flat if len(flat) <= limit else flat[:limit] + "…"
def _summary(etype: str, data: Any) -> str:
"""封存之后追加的事件说一句是什么。只挑不会带出数据的字段:判了几句、问的是什么、日志那一句。"""
if not isinstance(data, dict):
return ""
if etype == EventType.EVIDENCE_JUDGED:
verdicts = data.get("verdicts")
count = len(verdicts) if isinstance(verdicts, dict) else 0
model = data.get("model")
return f"按需裁判 {count} 句结论" + (f"(模型 {_clip(model, 40)})" if isinstance(model, str) and model else "")
if etype == EventType.APPROVAL_INQUIRY:
question = data.get("question")
return f"质询:{_clip(question)}" if isinstance(question, str) and question.strip() else "审批质询"
if etype == EventType.LOG and isinstance(data.get("message"), str):
return _clip(data["message"])
return ""
def _tier(events: Sequence[tuple[int, str, str | None, Any]]) -> dict[str, Any] | None:
"""出具档位:封存范围内最后一条出具判定事件里的 tier,带上序号(浏览器拿那条已核对的事件对照)。"""
for seq, etype, _, data in reversed(events):
if etype == EventType.ISSUANCE and isinstance(data, dict) and isinstance(data.get("tier"), str):
return {"value": data["tier"], "seq": seq}
return None
def _iso(value: Any) -> str | None:
if value is None:
return None
return value.isoformat() if isinstance(value, datetime) else str(value)
# --------------------------------------------------------------------------
# 构建
# --------------------------------------------------------------------------
def display_name(run: Mapping[str, Any]) -> str:
"""页面标题、附件名用的运行名:工作流名;没保存的工作流一律叫「未保存的工作流」。"""
name = str(run.get("workflow_name") or "").strip()
if not run.get("workflow_id") and name in _UNSAVED_NAMES:
return UNSAVED_NAME
return name or f"运行 {str(run.get('id') or '')[:8]}".strip()
def attachment_name(run: Mapping[str, Any]) -> str:
"""附件名「<运行名>-可验证报告.html」。文件名里不能有的字换成下划线。"""
name = _BAD_FILENAME.sub("_", display_name(run)).strip(" .")[:_NAME_MAX].strip(" .")
return f"{name or UNSAVED_NAME}{FILE_SUFFIX}"
def content_disposition(filename: str, *, run_id: str) -> str:
"""附件头:中文名按 RFC 5987 写在 filename* 里;filename 给不认 filename* 的老客户端一个纯 ASCII 的名字。"""
ascii_id = re.sub(r"[^A-Za-z0-9]", "", run_id)[:8]
return f"attachment; filename=\"agentlab-{ascii_id}-bundle.html\"; filename*=UTF-8''{quote(filename, safe='')}"
def build_bundle(
run: Mapping[str, Any],
events: Sequence[tuple[int, str, str | None, Any]],
*,
manifest_seq: int,
manifest_hash: str,
read: Reader,
exported_at: str,
legacy: bool = False,
graph: Mapping[str, Any] | None = None,
source_masks: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""一次已封存运行 → 页面里的数据(bundle-data)。
- run:运行记录的几样(id、workflow_id、workflow_name、version、version_hash、run_class、status、
started_at、finished_at)
- events:这次运行的全部事件 (seq, type, node_id, data),按数据库里 seq 的顺序。和封存核对
(runner.verify_manifest)读的是同一个顺序:seq 相同的几条,浏览器按这里的先后排,清单才算得一样
- manifest_seq / manifest_hash:封存范围(老运行是按结束事件推算的那个,legacy 为真)和封存哈希
- read:工件 id → 原字节(不复验哈希),没有就是 None
- source_masks:数据源名 → 它现在设的遮罩列(只列设了的)
"""
graph = graph if isinstance(graph, Mapping) else {}
node_types: dict[str, str] = {}
nodes: dict[str, dict[str, str]] = {}
for item in graph.get("nodes") or []:
if isinstance(item, Mapping) and item.get("id"):
nid = str(item["id"])
data = item.get("data") if isinstance(item.get("data"), Mapping) else {}
node_types[nid] = str(item.get("type") or "")
nodes[nid] = {"label": str(data.get("label") or nid), "type": node_types[nid]}
sealed = [e for e in events if e[0] <= manifest_seq]
later = [e for e in events if e[0] > manifest_seq]
store = _Store(read, source_masks or {})
artifacts = _Artifacts(store)
reports = _report_events(sealed)
cited_queries: list[str] = []
for report in reports:
doc_id, node_id = report["doc_artifact"], report["node_id"]
artifacts.add(doc_id, "report_doc", node_id)
doc = store.content(doc_id) if store.intact(doc_id) else None
# 文档要真是这次运行里这个节点写的:内容寻址只保证没被改,不保证没被张冠李戴。对不上的照实嵌进去让
# 浏览器标出来,但它的目录不可信,不顺着它去找别的工件
if not isinstance(doc, dict) or doc.get("run_id") != run.get("id") or doc.get("node_id") != node_id:
continue
for artifact, kind, owner in _cited_artifacts(doc):
artifacts.add(artifact, kind, owner)
if kind == "query":
cited_queries.append(artifact)
# 运行里有没有带遮罩的查询:有的话,成果节点和运行结束事件的预览可能直接带出查询结果,只给哈希
masked_run = any(store.masked(a) for a in dict.fromkeys([*_query_artifacts(sealed), *cited_queries]))
shown_events = []
for seq, etype, node_id, data in sealed:
text = canonical_json(data)
item: dict[str, Any] = {"seq": seq, "type": etype, "node_id": node_id, "data_sha256": content_hash(text)}
if _shows_content(etype, node_id, data, node_types, store, masked_run):
item["canonical"] = text
shown_events.append(item)
return {
"schema": SCHEMA,
"meta": {
"run_id": run.get("id"), "workflow_id": run.get("workflow_id"), "workflow_name": display_name(run),
"version": run.get("version"), "version_hash": run.get("version_hash"),
"run_class": run.get("run_class"), "status": run.get("status"),
"started_at": _iso(run.get("started_at")), "finished_at": _iso(run.get("finished_at")),
"manifest_seq": manifest_seq, "manifest_hash": manifest_hash, "legacy": bool(legacy),
"tier": _tier(sealed), "exported_at": exported_at,
},
"nodes": nodes,
"events": shown_events,
"appended": [{"seq": seq, "type": etype, "node_id": node_id, "summary": _summary(etype, data)}
for seq, etype, node_id, data in later],
"reports": reports,
"artifacts": artifacts.items,
}
# --------------------------------------------------------------------------
# 页面
# --------------------------------------------------------------------------
#: 数据放在 <script type="application/json"> 里:< > & 换成转义,正文里的 </script>、<!-- 截不断数据块;
#: U+2028 / U+2029 在老的 JS 解析器里是换行,一并转义。这几个字只会出现在 JSON 的字符串里,转义后 JSON 不变
_JSON_ESCAPES = str.maketrans({"<": "\\u003c", ">": "\\u003e", "&": "\\u0026",
"\u2028": "\\u2028", "\u2029": "\\u2029"})
_PLACEHOLDER = re.compile(r"__BUNDLE_(TITLE|CSP|DATA|SCRIPT)__")
@lru_cache(maxsize=1)
def _assets() -> tuple[str, str]:
here = Path(__file__).parent
return (here / "template.html").read_text(encoding="utf-8"), (here / "verifier.js").read_text(encoding="utf-8")
def bundle_json(bundle: Mapping[str, Any]) -> str:
"""页面数据块里的 JSON 文本。"""
return json.dumps(bundle, ensure_ascii=False, separators=(",", ":")).translate(_JSON_ESCAPES)
def render_html(bundle: Mapping[str, Any]) -> str:
"""把数据和核对脚本填进模板:样式、脚本都内联,零外链。
内容安全策略按哈希只放行这一段核对脚本,默认什么都不许连:页面打开后不会、也不能向任何地方发请求。
占位符一次替换完,填进去的内容不会再被当成占位符。"""
template, script = _assets()
digest = base64.b64encode(hashlib.sha256(script.encode("utf-8")).digest()).decode("ascii")
title = f"{(bundle.get('meta') or {}).get('workflow_name') or UNSAVED_NAME} · 可验证报告"
values = {"TITLE": html.escape(title), "CSP": f"'sha256-{digest}'", "DATA": bundle_json(bundle),
"SCRIPT": script}
return _PLACEHOLDER.sub(lambda m: values[m.group(1)], template)
+161
View File
@@ -0,0 +1,161 @@
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta http-equiv="Content-Security-Policy" content="default-src 'none'; script-src __BUNDLE_CSP__; style-src 'unsafe-inline'; img-src data:; base-uri 'none'; form-action 'none'">
<meta name="color-scheme" content="light dark">
<title>__BUNDLE_TITLE__</title>
<style>
:root {
color-scheme: light dark;
--bg: #ffffff; --panel: #f5f6f8; --raise: #ffffff; --border: #dfe2e8; --fg: #16181d; --dim: #41454e; --faint: #5f6470;
--accent: #2a57c2; --ok: #1b7339; --ok-soft: #e5f4ea; --bad: #b92424; --bad-soft: #fcebeb; --na: #565b66; --na-soft: #eceef2;
--mark: #fff1b8; --code: #eff1f5; --shadow: 0 10px 32px rgba(16, 20, 28, .16);
--sans: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei", "Noto Sans CJK SC", sans-serif;
--mono: ui-monospace, SFMono-Regular, Menlo, Consolas, "Liberation Mono", monospace;
}
@media (prefers-color-scheme: dark) {
:root {
--bg: #111318; --panel: #181b21; --raise: #1c1f26; --border: #2e323b; --fg: #e8eaee; --dim: #c1c5ce; --faint: #969ca8;
--accent: #8aaeff; --ok: #62d08c; --ok-soft: #14301f; --bad: #ff8080; --bad-soft: #3c1c1f; --na: #a9afbb; --na-soft: #262a31;
--mark: #4d4214; --code: #23262e; --shadow: 0 10px 32px rgba(0, 0, 0, .55);
}
}
* { box-sizing: border-box; }
[hidden] { display: none !important; }
html { -webkit-text-size-adjust: 100%; text-size-adjust: 100%; }
body { margin: 0; background: var(--bg); color: var(--fg); font: 15px/1.65 var(--sans); overflow-wrap: break-word; }
code, pre, .mono { font-family: var(--mono); }
.page { max-width: 920px; margin: 0 auto; padding: 28px 16px 72px; }
.brand { margin: 0; font-size: 13px; font-weight: 600; letter-spacing: .04em; color: var(--accent); }
h1 { margin: 2px 0 0; font-size: 24px; line-height: 1.35; }
.overall { display: flex; align-items: center; gap: 10px; margin: 16px 0 4px; padding: 12px 14px; border-radius: 10px;
border: 1px solid var(--border); background: var(--panel); font-size: 18px; font-weight: 650; }
.overall[data-overall="pass"] { color: var(--ok); background: var(--ok-soft); border-color: transparent; }
.overall[data-overall="fail"] { color: var(--bad); background: var(--bad-soft); border-color: transparent; }
.overall-mark { font-size: 20px; line-height: 1; }
.overall-sub { margin: 4px 2px 0; font-size: 13px; color: var(--dim); }
.seal { margin: 18px 0; padding: 14px 16px; border: 2px solid var(--accent); border-radius: 10px; }
.seal h2 { margin: 0 0 6px; font-size: 13px; font-weight: 600; color: var(--dim); }
.hash { display: block; font-size: 15px; font-weight: 600; line-height: 1.5; word-break: break-all; color: var(--fg); }
.seal p { margin: 10px 0 0; font-size: 13px; color: var(--dim); }
.meta { display: grid; grid-template-columns: max-content minmax(0, 1fr); gap: 4px 16px; margin: 0; font-size: 13px; }
.meta dt { color: var(--faint); }
.meta dd { margin: 0; min-width: 0; color: var(--dim); overflow-wrap: anywhere; }
.block { margin-top: 36px; }
.block > h2 { margin: 0 0 10px; font-size: 18px; }
.muted { color: var(--faint); }
.bad { color: var(--bad); }
.note { margin: 6px 0 12px; font-size: 13px; color: var(--dim); }
.warn { margin: 0 0 10px; padding: 8px 10px; border-radius: 8px; background: var(--bad-soft); color: var(--bad); font-size: 13px; }
.report { margin: 14px 0; padding: 16px; border: 1px solid var(--border); border-radius: 10px; background: var(--raise); }
.report-head { display: flex; flex-wrap: wrap; align-items: center; gap: 6px 10px; margin-bottom: 10px; }
.report-head h3 { margin: 0; font-size: 15px; }
.report-head .mono { font-size: 12px; }
.doc { font-size: 15px; }
.doc p { margin: 8px 0; }
.doc .doc-h { margin: 18px 0 6px; line-height: 1.4; }
.doc .doc-h1 { font-size: 19px; } .doc .doc-h2 { font-size: 17px; } .doc .doc-h3 { font-size: 15px; }
.doc ul, .doc ol { margin: 8px 0; padding-left: 1.6em; }
.doc li { margin: 2px 0; }
.doc blockquote { margin: 8px 0; padding-left: 12px; border-left: 3px solid var(--accent); color: var(--dim); }
.doc pre { margin: 8px 0; padding: 10px 12px; border-radius: 8px; background: var(--code); overflow-x: auto; font-size: 13px; line-height: 1.55; }
.doc code { font-size: .92em; padding: 0 4px; border-radius: 4px; background: var(--code); }
.doc pre code { padding: 0; background: none; }
.doc hr { border: 0; border-top: 1px solid var(--border); margin: 14px 0; }
.doc .link { color: var(--accent); }
.table-wrap { max-width: 100%; margin: 8px 0; overflow-x: auto; border: 1px solid var(--border); border-radius: 8px; }
.table-wrap table { width: 100%; border-collapse: collapse; font-size: 13px; }
.table-wrap th, .table-wrap td { padding: 5px 9px; border-bottom: 1px solid var(--border); text-align: left; vertical-align: top; }
.table-wrap th { background: var(--panel); color: var(--dim); font-weight: 600; white-space: nowrap; }
.table-wrap tr:last-child td { border-bottom: 0; }
.table-wrap .n { text-align: right; font-variant-numeric: tabular-nums; }
.table-wrap .hit { background: var(--mark); }
.num { font: inherit; color: inherit; background: none; border: 0; margin: 0; padding: 0 1px; border-radius: 3px; cursor: pointer;
text-decoration-line: underline; text-decoration-thickness: 1.5px; text-underline-offset: 3px; }
.num[data-result="pass"] { text-decoration-style: solid; text-decoration-color: var(--ok); }
.num[data-result="fail"] { text-decoration-style: wavy; text-decoration-color: var(--bad); background: var(--bad-soft); }
.num[data-result="na"] { text-decoration-style: dotted; text-decoration-color: var(--na); }
.num:hover { background: var(--panel); }
.num:focus-visible, .close:focus-visible, .items button:focus-visible, summary:focus-visible { outline: 2px solid var(--accent); outline-offset: 2px; }
.num.is-active { background: var(--mark); }
.badge { display: inline-flex; align-items: center; gap: 3px; padding: 2px 8px; border-radius: 999px; font-size: 12px;
font-weight: 600; line-height: 1.5; white-space: nowrap; }
.badge[data-result="pass"] { color: var(--ok); background: var(--ok-soft); }
.badge[data-result="fail"] { color: var(--bad); background: var(--bad-soft); }
.badge[data-result="na"] { color: var(--na); background: var(--na-soft); }
.check { margin: 12px 0; border: 1px solid var(--border); border-radius: 10px; overflow: hidden; }
.check-head { display: flex; flex-wrap: wrap; align-items: center; gap: 6px 10px; padding: 10px 12px; background: var(--panel); }
.check-head h3 { margin: 0; font-size: 14px; }
.check-sum { font-size: 13px; color: var(--dim); }
.items { margin: 0; padding: 0; list-style: none; }
.items li { display: flex; flex-wrap: wrap; align-items: baseline; gap: 4px 8px; padding: 7px 12px; border-top: 1px solid var(--border); font-size: 13px; }
.items .what { flex: 1 1 220px; min-width: 0; overflow-wrap: anywhere; }
.items .why { flex: 1 1 100%; color: var(--dim); }
.items button { font: inherit; color: var(--accent); background: none; border: 0; padding: 0; cursor: pointer; text-align: left; }
details > summary { padding: 8px 12px; border-top: 1px solid var(--border); font-size: 13px; color: var(--accent); cursor: pointer; }
.detail { position: fixed; top: 0; right: 0; bottom: 0; z-index: 10; width: min(420px, 100vw); overflow-y: auto;
background: var(--raise); border-left: 1px solid var(--border); box-shadow: var(--shadow); }
.detail-head { position: sticky; top: 0; display: flex; align-items: flex-start; gap: 8px; padding: 14px 16px 10px;
background: var(--raise); border-bottom: 1px solid var(--border); }
.detail-head h2 { flex: 1; min-width: 0; margin: 0; font-size: 17px; overflow-wrap: anywhere; }
.close { flex: none; width: 30px; height: 30px; border: 1px solid var(--border); border-radius: 8px; background: var(--panel);
color: var(--dim); font: 18px/1 var(--sans); cursor: pointer; }
.detail-body { padding: 12px 16px 28px; font-size: 14px; }
.kv { display: grid; grid-template-columns: max-content minmax(0, 1fr); gap: 7px 12px; margin: 0 0 12px; }
.kv dt { color: var(--faint); font-size: 13px; }
.kv dd { margin: 0; min-width: 0; overflow-wrap: anywhere; }
.sql { margin: 6px 0 12px; padding: 8px 10px; border-radius: 8px; background: var(--code); font-size: 12px; line-height: 1.5;
white-space: pre-wrap; overflow-wrap: anywhere; }
.sub { margin: 14px 0 4px; font-size: 13px; color: var(--faint); font-weight: 600; }
.foot { margin-top: 40px; padding-top: 14px; border-top: 1px solid var(--border); font-size: 12px; color: var(--faint); }
.foot p { margin: 4px 0; }
.noscript { max-width: 920px; margin: 24px auto; padding: 0 16px; color: var(--bad); }
@media (max-width: 760px) {
.detail { top: auto; left: 0; width: auto; max-height: 72vh; border-left: 0; border-top: 1px solid var(--border); border-radius: 14px 14px 0 0; }
}
@media (max-width: 520px) {
body { font-size: 14px; }
.page { padding: 20px 16px 56px; }
h1 { font-size: 20px; }
.overall { font-size: 16px; }
.hash { font-size: 13px; }
.meta { grid-template-columns: minmax(0, 1fr); gap: 0; }
.meta dt { margin-top: 6px; }
.report { padding: 12px; }
}
</style>
</head>
<body>
<div class="page">
<header>
<p class="brand">可验证报告</p>
<h1 id="title">可验证报告</h1>
<div id="overall" class="overall" data-overall="running" role="status" aria-live="polite">
<span class="overall-mark" aria-hidden="true">…</span><span id="overall-text">正在核对…</span>
</div>
<p id="overall-sub" class="overall-sub"></p>
<section class="seal" aria-labelledby="seal-label">
<h2 id="seal-label">封存清单哈希</h2>
<code id="seal-hash" class="hash"></code>
<p id="seal-note"></p>
</section>
<dl id="meta" class="meta"></dl>
</header>
<main>
<section id="reports" class="block" aria-labelledby="reports-title"><h2 id="reports-title">报告</h2></section>
<section id="checks" class="block" aria-labelledby="checks-title"><h2 id="checks-title">核对明细</h2></section>
<section id="appended" class="block" aria-labelledby="appended-title" hidden><h2 id="appended-title">封存之后追加的事件</h2></section>
</main>
<footer id="foot" class="foot"></footer>
</div>
<aside id="detail" class="detail" role="dialog" aria-modal="false" aria-labelledby="detail-title" hidden>
<div class="detail-head"><h2 id="detail-title"></h2><button type="button" id="detail-close" class="close" aria-label="关闭">×</button></div>
<div id="detail-body" class="detail-body"></div>
</aside>
<noscript><p class="noscript">这个文件在浏览器里自行核对,需要启用 JavaScript。</p></noscript>
<script type="application/json" id="bundle-data">__BUNDLE_DATA__</script>
<script>__BUNDLE_SCRIPT__</script>
</body>
</html>
+962
View File
@@ -0,0 +1,962 @@
/*
* 可验证报告的浏览器端核对:原生 JS,无依赖,不连网。
*
* 数据在 <script type="application/json" id="bundle-data"> 里(服务端 bundle/build.py 生成),逐项核对:
* ① 每条带规范字符串的事件:sha256(canonical) 等于它的 data_sha256;
* ② 封存清单:照 runner 的拼法(每行 seq|type|node_id|data_sha256,node_id 为空写空串,按 seq 排序,换行连接,
* 没有结尾换行)重算,等于封存哈希;
* ③ 每份嵌入的工件:sha256(原字节) 等于工件 id;
* ④ 引用链:报告文档的 id 是这个报告节点最后一次自查事件(已核对)记下的那份,文档记的运行、节点对得上;
* 口径卡和快照的 id 出现在已核对的事件内容里,或者已核对的报告文档的证据目录里;
* ⑤ 报告里的每个数字:引用的值等于口径卡里那个指标的值,或者快照里那一格的值。判等照搬服务端
* engine/expressions.py 的 cell_value / same_value;数字怎么显示是服务端渲染好的,这里不重新排版。
* 每项给出通过 / 不通过 / 无法核对和原因。这些只证明内部一致,真伪要拿封存哈希去系统里对照,页面上写明。
*
* SHA-256 优先用浏览器的 crypto.subtle;拿不到(非安全上下文、某些浏览器打开本地文件时)用下面的纯 JS 实现,
* 两者结果一致。纯 JS 实现启动时先拿标准测试向量自检,自检不过就不下任何结论。
* 来自数据的文字一律经 textContent 写进页面,不拼 HTML。
*/
(() => {
'use strict'
const PASS = 'pass'
const FAIL = 'fail'
const NA = 'na'
// ------------------------------------------------------------------ 文案
const RESULT = { pass: '通过', fail: '不通过', na: '无法核对' }
const MARK = { pass: '✓', fail: '✗', na: '?' }
const KIND_LABEL = { report_doc: '报告文档', metric_set: '口径卡', query: '查询快照' }
const TIER_LABEL = { formal: '完整出具', degraded: '降档出具', withheld: '不予出具' }
const CLASS_LABEL = { formal: '正式运行', exploratory: '探索运行' }
const EVENT_LABEL = {
'run.started': '运行开始', 'run.finished': '运行结束', 'run.failed': '运行失败', 'run.cancelled': '运行取消',
'run.interrupted': '等待人工介入', 'run.resumed': '恢复运行', 'node.started': '节点开始', 'node.finished': '节点完成',
'node.failed': '节点失败', 'node.skipped': '节点跳过', 'edge.taken': '分支走向', 'llm.start': '模型调用开始',
'llm.end': '模型调用结束', 'llm.thinking': '模型思考', 'tool.start': '工具调用', 'tool.end': '工具返回',
'tool.error': '工具出错', 'tool.gated': '门控判定', 'retrieve.end': '知识检索', 'memory.end': '记忆读写',
'merge.end': '合并查询', 'sandbox.start': '沙箱开始', 'sandbox.end': '沙箱结束', 'agent.step.start': 'Agent 步骤开始',
'agent.step.end': 'Agent 步骤结束', 'agent.route.start': '调度决策开始', 'agent.route.end': '调度决策结束',
'human.requested': '请求人工审批', 'human.resolved': '人工审批结果', 'approval.inquiry': '审批质询',
'issuance': '出具判定', 'caliber.upgrade': '口径卡升版处置', 'catalog.drift': '数据目录变化',
'report.checked': '报告自查', 'evidence.judged': '按需裁判', 'log': '日志', 'usage': '用量',
}
const T = {
allPass: '全部通过',
passWithNa: '可核对的项全部通过',
failed: (n) => `有 ${n} 项不通过`,
tally: (p, f, n) => `通过 ${p} 项 · 不通过 ${f} 项 · 无法核对 ${n} 项`,
broken: '页面数据无法解析,无法核对',
selfTestFailed: '本页的 SHA-256 实现在这个浏览器中自检未通过,不给出核对结论',
sealNote: (id) => '这个文件只证明内部一致:报告里的数字能追到口径卡或查询快照,文档、快照和运行记录彼此对得上。'
+ '它不能单独证明这些记录出自系统、未被整体替换。核对真伪:在系统中打开这次运行(运行 ID '
+ `${id}),确认「凭证详情」里的清单哈希与上面一致,并点「核对」确认事件记录与封存时一致`
+ `(接口 GET /api/runs/${id}/verify)。`,
metaRun: '运行 ID', metaWorkflow: '工作流', metaClass: '运行类别', metaVersionHash: '版本哈希', metaStarted: '开始',
metaFinished: '结束', metaRange: '封存范围', metaTier: '出具档位', metaExported: '导出时间',
// 序号不连续(模型的流式输出占序号、不落库),后面注明范围内实际有几条。说法同记录页的封存凭证
range: (seq, count) => `第 1–${seq} 条事件(范围内 ${count} 条)`,
rangeLegacy: '(早期运行,范围按结束事件推算)',
tierChecked: (seq) => `第 ${seq} 条事件,已核对`,
tierUnchecked: '与已核对的事件对不上',
noReports: '这次运行没有报告文档,下方只核对运行记录。',
reportTitle: (label) => `报告「${label}」`,
docUntrusted: '这份报告文档没有通过核对(见核对明细 ③ ④),正文仅供参考,其中的数字不予核对。',
docUnreadable: '报告文档无法解析,无法显示正文。',
checkEvents: '① 事件内容与哈希',
checkManifest: '② 封存清单',
checkArtifacts: '③ 工件与哈希',
checkChain: '④ 引用链',
checkNumbers: '⑤ 报告中的数字',
sumEvents: (n) => `${n} 条带内容的事件;其余事件只附哈希,参与 ② 的清单核对`,
sumManifest: (n) => `按 ${n} 条事件的哈希重算`,
sumCount: (n, unit) => `${n} ${unit}`,
showAll: (n) => `展开全部 ${n} 项`,
eventMismatch: '内容与记录的哈希不一致',
eventNotJson: '内容不是有效的 JSON',
manifestComputed: '重算结果',
manifestOk: '与封存清单哈希一致',
manifestMismatch: '与封存清单哈希不一致:事件被增删、改动或换了顺序',
beyondSeal: (seq) => `第 ${seq} 条事件的序号超出封存范围`,
artifactMismatch: '内容与工件 id(哈希)不一致',
artifactUndecodable: '内容无法解码',
artifactNotJson: '内容不是有效的 JSON',
masked: '含遮罩列,未随文件附带,只能核对到哈希',
artifactMissing: '导出时服务端已没有这份工件的文件',
artifactAbsent: '内容未随文件附带',
chainDoc: (label) => `报告「${label}」的文档`,
chainByEvent: (seq, what) => `第 ${seq} 条事件(${what})`,
chainByDoc: (label, alias) => `报告「${label}」的证据目录(${alias})`,
chainFrom: '引用自',
chainNoCheckEvent: '找不到这个报告节点的报告自查事件',
chainCheckEventFailed: '报告自查事件的内容没有通过核对',
chainOtherDoc: '这个节点最后一次报告自查记下的不是这份文档',
chainDocUnverified: '文档内容没有通过核对,无法确认它记下的运行和节点',
chainDocElsewhere: '文档记下的运行或节点与这次不符',
chainDocLeftOut: '运行记录里有这个报告节点的自查事件,文件里却没有附带它的文档',
chainUnreferenced: '已核对的事件和报告文档里都没有引用这份工件',
srcMetric: (name) => `口径卡指标「${name}」`,
srcCell: (alias, row, column) => `查询 ${alias} · 第 ${row + 1} 行 · ${column}`,
srcQuery: (alias) => `查询 ${alias}`,
srcInput: (field) => `运行输入「${field}」`,
srcNone: '没有出处',
srcOther: '其他引用',
numUncited: '这个数字没有引用出处(出具时已标为无出处)',
numUnresolved: (why) => `报告生成时这处引用未能解析${why ? `:${why}` : ''}`,
numInput: '引用的是运行输入,运行输入未随文件附带',
numOther: '这类引用不在核对范围内',
numDocUntrusted: '所在的报告文档没有通过核对',
numNoEntry: '报告的证据目录里没有这处引用指向的条目',
numArtifactAbsent: '引用的工件未随文件附带',
numArtifactFailed: '引用的工件没有通过哈希核对',
numNotChained: '引用的工件不在引用链上',
numNoMetric: '口径卡里没有这个指标',
numMetricDiffers: '引用的值与口径卡里的值不一致',
numNoCell: (why) => `快照里找不到这一格:${why}`,
numCellDiffers: '引用的值与快照里那一格的值不一致',
numAria: (text, result) => `数字 ${text},${result},查看出处`,
dResult: '核对结果', dSource: '出处', dNode: '来源节点', dCaliber: '口径', dCited: '引用的值', dCard: '口径卡里的值',
dCell: '快照里的值', dShown: '正文显示', dShownNote: '显示格式由服务端渲染', dArtifact: '所在工件', dReport: '所在报告',
dSql: 'SQL', dRow: (row) => `快照第 ${row + 1} 行`,
nullValue: '空值(NULL)',
appendedNote: '这些事件追加在封存之后,不在封存范围内,不参与清单核对,也不能当作证据。',
appendedTag: '封存之后追加,不在封存范围内',
shaSubtle: 'SHA-256 由浏览器内置实现计算(crypto.subtle)',
shaJs: 'SHA-256 由本页内置的实现计算(浏览器未提供 crypto.subtle),结果与浏览器内置实现一致',
renderNote: '数字的显示格式(千分位、单位、小数位)由服务端渲染;本页核对的是引用的值,不重新排版。',
schema: (s) => `文件格式 ${s}`,
}
// ------------------------------------------------------------------ SHA-256
const K = new Uint32Array([
0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5,
0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174,
0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da,
0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967,
0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85,
0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070,
0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3,
0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2,
])
/** 纯 JS 的 SHA-256(FIPS 180-4):字节 → 64 位小写十六进制 */
function sha256js(bytes) {
const n = bytes.length
const padded = new Uint8Array(Math.ceil((n + 9) / 64) * 64)
padded.set(bytes)
padded[n] = 0x80
const view = new DataView(padded.buffer)
// 消息长度按 64 位大端写在最后:位数可能超过 2^32,拆成高低两半
view.setUint32(padded.length - 8, Math.floor(n / 0x20000000))
view.setUint32(padded.length - 4, (n % 0x20000000) * 8)
const H = new Uint32Array([0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab, 0x5be0cd19])
const W = new Uint32Array(64)
for (let off = 0; off < padded.length; off += 64) {
for (let i = 0; i < 16; i++) W[i] = view.getUint32(off + i * 4)
for (let i = 16; i < 64; i++) {
const a = W[i - 15]
const b = W[i - 2]
const s0 = ((a >>> 7) | (a << 25)) ^ ((a >>> 18) | (a << 14)) ^ (a >>> 3)
const s1 = ((b >>> 17) | (b << 15)) ^ ((b >>> 19) | (b << 13)) ^ (b >>> 10)
W[i] = W[i - 16] + s0 + W[i - 7] + s1
}
let a = H[0]; let b = H[1]; let c = H[2]; let d = H[3]
let e = H[4]; let f = H[5]; let g = H[6]; let h = H[7]
for (let i = 0; i < 64; i++) {
const S1 = ((e >>> 6) | (e << 26)) ^ ((e >>> 11) | (e << 21)) ^ ((e >>> 25) | (e << 7))
const ch = (e & f) ^ (~e & g)
const t1 = (h + S1 + ch + K[i] + W[i]) | 0
const S0 = ((a >>> 2) | (a << 30)) ^ ((a >>> 13) | (a << 19)) ^ ((a >>> 22) | (a << 10))
const maj = (a & b) ^ (a & c) ^ (b & c)
const t2 = (S0 + maj) | 0
h = g; g = f; f = e; e = (d + t1) | 0
d = c; c = b; b = a; a = (t1 + t2) | 0
}
H[0] += a; H[1] += b; H[2] += c; H[3] += d; H[4] += e; H[5] += f; H[6] += g; H[7] += h
}
let out = ''
for (let i = 0; i < 8; i++) out += H[i].toString(16).padStart(8, '0')
return out
}
const encoder = new TextEncoder()
const decoder = new TextDecoder('utf-8', { fatal: true })
const utf8 = (text) => encoder.encode(text)
const bytesHex = (buf) => Array.from(new Uint8Array(buf), (x) => x.toString(16).padStart(2, '0')).join('')
// 标准测试向量(空串、abc、跨块的 448 位串、中文、1000 个 a)
const VECTORS = [
['', 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'],
['abc', 'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad'],
['abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq', '248d6a61d20638b8e5c026930c3e6039a33ce45964ff2167f6ecedd419db06c1'],
['可验证报告', 'e49666cf360bf3171c8e7d9645148517861e1966a3b4c6b6785c1cd3a2014a13'],
['a'.repeat(1000), '41edece42d63e8d9bf515a9ba6932e1c20cbc9f5a5d134645adb5db1b9737ea3'],
]
/** 用哪一种 SHA-256:{ kind: 'subtle' | 'js', digest(字节) → Promise<十六进制> };纯 JS 实现自检不过返回 null */
async function pickDigest() {
if (!VECTORS.every(([text, sum]) => sha256js(utf8(text)) === sum)) return null
const subtle = typeof crypto !== 'undefined' && crypto ? crypto.subtle : undefined
if (subtle && typeof subtle.digest === 'function') {
try {
if (bytesHex(await subtle.digest('SHA-256', utf8('abc'))) === VECTORS[1][1]) {
return { kind: 'subtle', digest: async (bytes) => bytesHex(await subtle.digest('SHA-256', bytes)) }
}
} catch (e) {
// 非安全上下文、打开本地文件时可能拿不到或者用不了:退回内置实现
}
}
return { kind: 'js', digest: async (bytes) => sha256js(bytes) }
}
function fromBase64(text) {
const bin = atob(text)
const out = new Uint8Array(bin.length)
for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i)
return out
}
// ------------------------------------------------------------------ 判等:照搬 engine/expressions.py
const isObj = (v) => v !== null && typeof v === 'object' && !Array.isArray(v)
const own = (obj, key) => (isObj(obj) && typeof key === 'string' && Object.prototype.hasOwnProperty.call(obj, key)
? obj[key] : undefined)
// 数据源把 DECIMAL 落成数字文本("45678.50"、"45678"、"1E-7"):按数算;带前导零的是编号,照文本(同 _NUMERIC_TEXT)
const NUMERIC_TEXT = /^\s*-?(?:0|[1-9]\d*)(?:\.\d+)?(?:E[+-]?\d+)?\s*$/
/** cell_value:快照里的一格 → 参与比对的值。kind 是查询时记下的列类型,text 列原样 */
function cellValue(raw, kind) {
if (kind === 'text') return raw
return typeof raw === 'string' && NUMERIC_TEXT.test(raw) ? Number(raw) : raw
}
/** math.isclose(a, b, rel_tol=1e-9, abs_tol=1e-9) */
function isClose(a, b) {
if (a === b) return true
if (!Number.isFinite(a) || !Number.isFinite(b)) return false
return Math.abs(a - b) <= Math.max(1e-9 * Math.max(Math.abs(a), Math.abs(b)), 1e-9)
}
const textOf = (v) => (typeof v === 'string' ? v : isObj(v) || Array.isArray(v) ? JSON.stringify(v) : String(v))
/** same_value:数按数比(数字文本也算数),布尔按 Python 的规矩(True == 1),其余压成一行按文本比 */
function sameValue(model, truth) {
if (model == null || truth == null) return model == null && truth == null
const m = cellValue(model)
const t = cellValue(truth)
if (typeof m === 'boolean' || typeof t === 'boolean') {
const num = (x) => (typeof x === 'boolean' ? Number(x) : typeof x === 'number' ? x : NaN)
return num(m) === num(t)
}
if (typeof m === 'number' && typeof t === 'number') return isClose(m, t)
return textOf(m).replace(/\s+/g, ' ').trim() === textOf(t).replace(/\s+/g, ' ').trim()
}
/** locate_cell:(那一格的原值, 列名)。列号换成列名;行是对象时按键取 */
function locateCell(table, row, column) {
if (!isObj(table) || !Array.isArray(table.rows)) throw new Error('快照里没有表格数据')
const rows = table.rows
if (!Number.isInteger(row)) throw new Error(`行号需为从 0 开始的整数,当前为「${row}」`)
if (!rows.length) throw new Error('查询结果为空(0 行)')
if (row < 0 || row >= rows.length) throw new Error(`请求第 ${row} 行,但查询结果只有 ${rows.length} 行(行号从 0 开始)`)
const record = rows[row]
let columns = (Array.isArray(table.columns) ? table.columns : []).map(String)
if (isObj(record) && !columns.length) columns = Object.keys(record)
let name = column
if (Number.isInteger(column)) {
if (column < 0 || column >= columns.length) throw new Error(`请求第 ${column} 列,但查询结果只有 ${columns.length} 列`)
name = columns[column]
} else if (typeof column !== 'string') {
throw new Error(`列需填写列名或从 0 开始的列号,当前为「${column}」`)
}
if (isObj(record)) {
if (!Object.prototype.hasOwnProperty.call(record, name)) throw new Error(`没有列「${name}」`)
return [record[name], name]
}
const index = columns.indexOf(name)
if (index < 0) throw new Error(`没有列「${name}」`)
if (!Array.isArray(record) || index >= record.length) throw new Error(`第 ${row} 行中没有「${name}」对应的单元格`)
return [record[index], name]
}
function columnKind(table, column) {
const kind = own(own(table, 'column_types'), column)
return typeof kind === 'string' ? kind : null
}
// ------------------------------------------------------------------ 核对
function push(map, key, value) {
if (!map.has(key)) map.set(key, [])
map.get(key).push(value)
}
/** 内容里出现过的工件 id(64 位十六进制的字符串),不论在哪一层 */
function collectIds(value, found) {
if (typeof value === 'string') {
if (/^[0-9a-f]{64}$/.test(value)) found.add(value)
} else if (Array.isArray(value)) {
for (const v of value) collectIds(v, found)
} else if (isObj(value)) {
for (const v of Object.values(value)) collectIds(v, found)
}
return found
}
function* segmentsOf(doc) {
for (const block of Array.isArray(doc.blocks) ? doc.blocks : []) {
for (const unit of isObj(block) && Array.isArray(block.units) ? block.units : []) {
for (const seg of isObj(unit) && Array.isArray(unit.segments) ? unit.segments : []) {
if (isObj(seg)) yield seg
}
}
}
}
async function verify(data, digest) {
const out = { events: [], manifest: null, artifacts: [], chain: [], numbers: [], docs: [], checked: new Map() }
const events = Array.isArray(data.events) ? data.events : []
const meta = isObj(data.meta) ? data.meta : {}
// ① 带规范字符串的事件
for (const ev of events) {
if (typeof ev.canonical !== 'string') continue
const item = { ev, result: FAIL, reason: '' }
if ((await digest(utf8(ev.canonical))) !== ev.data_sha256) {
item.reason = T.eventMismatch
} else {
try {
out.checked.set(ev, JSON.parse(ev.canonical))
item.result = PASS
} catch (e) {
item.reason = T.eventNotJson
}
}
out.events.push(item)
}
// ② 封存清单:seq 相同的几条按文件里的先后(服务端按数据库的顺序给,和封存时一样)
const order = events.map((ev, i) => [ev, i]).sort((x, y) => x[0].seq - y[0].seq || x[1] - y[1]).map(([ev]) => ev)
const lines = order.map((ev) => `${ev.seq}|${ev.type}|${ev.node_id == null ? '' : ev.node_id}|${ev.data_sha256}`)
const computed = await digest(utf8(lines.join('\n')))
const beyond = order.find((ev) => !(ev.seq <= meta.manifest_seq))
out.manifest = {
computed, count: lines.length,
result: computed === meta.manifest_hash && !beyond ? PASS : FAIL,
reason: computed !== meta.manifest_hash ? T.manifestMismatch : beyond ? T.beyondSeal(beyond.seq) : T.manifestOk,
}
// ③ 工件
const arts = new Map()
for (const a of Array.isArray(data.artifacts) ? data.artifacts : []) {
const item = { a, result: NA, reason: '', json: null, loose: null }
if (typeof a.b64 === 'string') {
let bytes = null
try {
bytes = fromBase64(a.b64)
} catch (e) {
item.result = FAIL
item.reason = T.artifactUndecodable
}
if (bytes) {
const intact = (await digest(bytes)) === a.id
try {
// 对不上的也解析出来(loose):被改过的报告仍显示正文,醒目地标着不可信
item.loose = JSON.parse(decoder.decode(bytes))
} catch (e) {
item.loose = null
}
if (!intact) {
item.result = FAIL
item.reason = T.artifactMismatch
} else if (item.loose === null) {
item.result = FAIL
item.reason = T.artifactNotJson
} else {
item.result = PASS
item.json = item.loose
}
}
} else {
item.reason = a.masked ? T.masked : a.missing ? T.artifactMissing : T.artifactAbsent
}
arts.set(a.id, item)
out.artifacts.push(item)
}
// ④ 引用链
const byEvent = new Map() // 工件 id → 引用它的已核对事件
for (const [ev, body] of out.checked) for (const id of collectIds(body, new Set())) push(byEvent, id, ev)
// 每个报告节点认最后一次记下了文档的自查(和服务端、证据接口同一个认法)。内容没核对通过的那条
// 不知道记没记文档,也算上:宁可报不通过,也不能跳过它去认一份更早的
const lastChecked = new Map()
for (const ev of order) {
if (ev.type !== 'report.checked' || !ev.node_id) continue
const body = out.checked.get(ev)
if (body !== undefined && !own(body, 'doc_artifact')) continue
lastChecked.set(ev.node_id, ev)
}
const reports = Array.isArray(data.reports) ? data.reports.filter(isObj) : []
const trusted = new Map() // 报告 → 已核对的文档
const chained = new Set()
for (const report of reports) {
const art = arts.get(report.doc_artifact)
const ev = lastChecked.get(report.node_id)
const link = { kind: 'report_doc', report, art, ev, result: FAIL, reason: '' }
if (!ev) link.reason = T.chainNoCheckEvent
else if (!out.checked.has(ev)) link.reason = T.chainCheckEventFailed
else if (own(out.checked.get(ev), 'doc_artifact') !== report.doc_artifact) link.reason = T.chainOtherDoc
else if (!art || art.result !== PASS) { link.result = NA; link.reason = T.chainDocUnverified }
else if (own(art.json, 'run_id') !== meta.run_id || own(art.json, 'node_id') !== report.node_id) link.reason = T.chainDocElsewhere
else link.result = PASS
out.chain.push(link)
if (link.result === PASS) {
trusted.set(report, art.json)
chained.add(report.doc_artifact)
}
}
// 运行记录里自查过、文件却没附带的报告:不能悄悄少一份
for (const [nodeId, ev] of lastChecked) {
if (!reports.some((r) => r.node_id === nodeId)) {
const body = out.checked.get(ev)
out.chain.push({ kind: 'report_doc', report: { node_id: nodeId, doc_artifact: own(body, 'doc_artifact') },
art: null, ev, result: FAIL, reason: body === undefined ? T.chainCheckEventFailed : T.chainDocLeftOut })
}
}
const byDoc = new Map() // 工件 id → [{ report, alias }]
for (const [report, doc] of trusted) {
const catalog = isObj(doc.catalog) ? doc.catalog : {}
for (const alias of Object.keys(catalog)) {
const entry = catalog[alias]
if (isObj(entry) && typeof entry.artifact === 'string') push(byDoc, entry.artifact, { report, alias })
}
}
for (const item of out.artifacts) {
if (item.a.kind === 'report_doc') continue
const evs = byEvent.get(item.a.id) || []
const docs = byDoc.get(item.a.id) || []
const ok = evs.length > 0 || docs.length > 0
out.chain.push({ kind: item.a.kind, art: item, events: evs, docs, result: ok ? PASS : FAIL,
reason: ok ? '' : T.chainUnreferenced })
if (ok) chained.add(item.a.id)
}
// ⑤ 报告里的数字
for (const report of reports) {
const art = arts.get(report.doc_artifact)
const doc = trusted.get(report) || (art && isObj(art.loose) ? art.loose : null)
out.docs.push({ report, art, doc, trusted: trusted.has(report) })
if (!doc) continue
for (const seg of segmentsOf(doc)) {
if (seg.kind !== 'number' && seg.kind !== 'value') continue
const check = trusted.has(report) ? checkNumber(seg, doc, arts, chained)
: { result: NA, reason: T.numDocUntrusted, source: describe(seg, doc) }
out.numbers.push({ ...check, seg, report, key: `${report.node_id}|${seg.id}` })
}
}
return out
}
/** 片段的出处说一句(核对前就能说的那部分) */
function describe(seg, doc) {
const cite = isObj(seg.cite) ? seg.cite : null
if (!cite) return T.srcNone
const entry = own(doc.catalog, cite.alias)
const loc = isObj(cite.locator) ? cite.locator : {}
if (cite.kind === 'metric') return T.srcMetric(own(entry, 'name') || loc.metric || cite.ref || '')
if (cite.kind === 'cell') {
return Number.isInteger(loc.row) ? T.srcCell(cite.alias, loc.row, String(loc.column ?? '')) : T.srcQuery(cite.alias)
}
if (cite.kind === 'input') return T.srcInput(loc.field || cite.ref || '')
return T.srcOther
}
function checkNumber(seg, doc, arts, chained) {
const cite = isObj(seg.cite) ? seg.cite : null
const source = describe(seg, doc)
if (!cite) return { result: NA, reason: T.numUncited, source }
if (cite.status !== 'resolved') return { result: NA, reason: T.numUnresolved(cite.reason), source }
if (cite.kind === 'input') return { result: NA, reason: T.numInput, source }
if (cite.kind !== 'metric' && cite.kind !== 'cell') return { result: NA, reason: T.numOther, source }
const entry = own(doc.catalog, cite.alias)
const info = { source, kind: cite.kind, cite }
const wanted = cite.kind === 'metric' ? 'metric' : 'query'
if (!isObj(entry) || entry.kind !== wanted || typeof entry.artifact !== 'string') {
return { ...info, result: FAIL, reason: T.numNoEntry }
}
info.artifact = entry.artifact
info.node = entry.node_id
const art = arts.get(entry.artifact)
if (!art) return { ...info, result: NA, reason: T.numArtifactAbsent }
if (art.a.masked || art.a.missing) return { ...info, result: NA, reason: art.reason }
if (art.result !== PASS) return { ...info, result: NA, reason: T.numArtifactFailed }
if (!chained.has(entry.artifact)) return { ...info, result: NA, reason: T.numNotChained }
if (cite.kind === 'metric') {
const metricId = own(cite.locator, 'metric') ?? own(entry.locator, 'metric')
const metrics = Array.isArray(art.json.metrics) ? art.json.metrics : []
const metric = metrics.find((m) => isObj(m) && m.id === metricId)
info.card = art.json
if (!metric) return { ...info, result: FAIL, reason: T.numNoMetric }
info.truth = metric.value
return sameValue(cite.value, metric.value) ? { ...info, result: PASS } : { ...info, result: FAIL, reason: T.numMetricDiffers }
}
const loc = isObj(cite.locator) ? cite.locator : {}
info.snapshot = art.json
try {
const [raw, column] = locateCell(art.json, loc.row, loc.column)
info.row = loc.row
info.column = column
info.truth = cellValue(raw, columnKind(art.json, column))
} catch (e) {
return { ...info, result: FAIL, reason: T.numNoCell(e.message) }
}
return sameValue(cite.value, info.truth) ? { ...info, result: PASS } : { ...info, result: FAIL, reason: T.numCellDiffers }
}
// ------------------------------------------------------------------ 渲染:工具
const $ = (id) => document.getElementById(id)
function h(tag, props, ...kids) {
const el = document.createElement(tag)
for (const [key, value] of Object.entries(props || {})) {
if (value == null || value === false) continue
if (key === 'class') el.className = value
else if (key.startsWith('on')) el.addEventListener(key.slice(2), value)
else el.setAttribute(key, value === true ? '' : String(value))
}
for (const kid of kids.flat(Infinity)) {
if (kid != null && kid !== false) el.append(kid instanceof Node ? kid : String(kid))
}
return el
}
const short = (id) => (typeof id === 'string' && id.length > 12 ? `${id.slice(0, 12)}…` : String(id ?? ''))
const badge = (result) => h('span', { class: 'badge', 'data-result': result }, `${MARK[result]} ${RESULT[result]}`)
const pad = (n) => String(n).padStart(2, '0')
/** 时间按读者所在时区显示,后面注明时区:文件会被转给别的时区的人 */
function when(iso) {
if (!iso) return '—'
const d = new Date(iso)
if (Number.isNaN(d.getTime())) return String(iso)
const offset = -d.getTimezoneOffset()
const zone = `UTC${offset < 0 ? '-' : '+'}${Math.floor(Math.abs(offset) / 60)}${Math.abs(offset) % 60 ? `:${pad(Math.abs(offset) % 60)}` : ''}`
return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())} ${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}(${zone})`
}
function valueText(v) {
if (v == null) return T.nullValue
if (typeof v === 'string') return `「${v}」`
return textOf(v)
}
function kv(rows) {
return h('dl', { class: 'kv' }, rows.filter(Boolean).map(([k, v]) => [h('dt', {}, k), h('dd', {}, v)]))
}
function worst(items) {
if (items.some((i) => i.result === FAIL)) return FAIL
if (items.some((i) => i.result === PASS)) return PASS
return NA
}
// ------------------------------------------------------------------ 渲染:报告正文
// 行内语法:代码、链接(只显示文字,不给链接:页面不连网)、粗体、删除线。和前端 Markdown.tsx 的规则一致,
// 斜体不认(要用后行断言,老浏览器不支持)
const INLINE = '(`+)([^`]+?)\\1|\\[([^\\]]+)\\]\\(([^)\\s]+)[^)]*\\)|\\*\\*([^*]+?)\\*\\*|__([^_]+?)__|~~([^~]+?)~~'
const BIT = { code: 1, bold: 2, strike: 4, link: 8 }
// 列表项的开头:行首的 -、*、+ 或者序号
const ITEM_START = /(^|\n)[ \t]*([-*+]|\d+[.)])[ \t]+$/
/** 整句算一次行内样式:服务端把一句切成了好几个片段,粗体、代码可能跨片段,每个片段取自己那一截 */
function inlineStyles(text) {
const hide = new Uint8Array(text.length)
const style = new Uint8Array(text.length)
const walk = (from, to, inherited, depth) => {
const part = text.slice(from, to)
const re = new RegExp(INLINE, 'g')
let last = 0
for (let m = re.exec(part); m; m = re.exec(part)) {
for (let i = last; i < m.index; i++) style[from + i] = inherited
const [whole, ticks, code, linkText, , bold1, bold2, strike] = m
const inner = code ?? linkText ?? bold1 ?? bold2 ?? strike ?? ''
const base = from + m.index
const off = code !== undefined ? ticks.length : whole.indexOf(inner)
for (let i = 0; i < whole.length; i++) hide[base + i] = 1
for (let i = 0; i < inner.length; i++) hide[base + off + i] = 0
const bit = code !== undefined ? BIT.code : linkText !== undefined ? BIT.link
: bold1 !== undefined || bold2 !== undefined ? BIT.bold : BIT.strike
if (code !== undefined || linkText !== undefined || depth >= 3) {
for (let i = 0; i < inner.length; i++) style[base + off + i] = inherited | bit
} else {
walk(base + off, base + off + inner.length, inherited | bit, depth + 1)
}
last = m.index + whole.length
}
for (let i = last; i < part.length; i++) style[from + i] = inherited
}
walk(0, text.length, 0, 0)
return { hide, style }
}
function styledSlice(text, styles, from, to) {
const frag = document.createDocumentFragment()
let i = from
while (i < to) {
if (styles.hide[i]) { i++; continue }
const st = styles.style[i]
let j = i + 1
while (j < to && !styles.hide[j] && styles.style[j] === st) j++
let node = document.createTextNode(text.slice(i, j))
if (st & BIT.code) node = h('code', {}, node)
if (st & BIT.bold) node = h('strong', {}, node)
if (st & BIT.strike) node = h('s', {}, node)
if (st & BIT.link) node = h('span', { class: 'link' }, node)
frag.append(node)
i = j
}
return frag
}
const unitText = (unit) => (Array.isArray(unit.segments) ? unit.segments : [])
.filter((s) => isObj(s) && s.kind !== 'structural').map((s) => String(s.text ?? '')).join('')
function renderUnit(unit, ctx) {
const segs = (Array.isArray(unit.segments) ? unit.segments : []).filter((s) => isObj(s) && s.kind !== 'structural')
const text = segs.map((s) => String(s.text ?? '')).join('')
const styles = inlineStyles(text)
const frag = document.createDocumentFragment()
let pos = 0
for (const seg of segs) {
const from = pos
pos += String(seg.text ?? '').length
let body = styledSlice(text, styles, from, pos)
if (seg.strong) body = h('strong', {}, body)
const number = ctx.numbers.get(`${ctx.report.node_id}|${seg.id}`)
frag.append(number ? numberButton(number, body) : h('span', {}, body))
}
return frag
}
const NUMERIC_CELL = /^[-+−]?[\d,]+(?:\.\d+)?%?[^\d\s]{0,4}$/
function renderBlock(block, ctx) {
const units = Array.isArray(block.units) ? block.units.filter(isObj) : []
const inline = (u) => renderUnit(u, ctx)
switch (block.type) {
case 'heading': {
const level = Math.min(Math.max(Number(block.level) || 2, 1), 3)
return h(`h${level + 3}`, { class: `doc-h doc-h${level}` }, units.map(inline))
}
case 'hr':
return h('hr')
case 'quote':
return h('blockquote', {}, units.map(inline))
case 'code':
return h('pre', {}, h('code', {}, units.map(unitText).join('')))
case 'list': {
const items = []
for (const u of units) {
const lead = Array.isArray(u.segments) ? u.segments[0] : null
const starts = !items.length || (isObj(lead) && lead.kind === 'structural' && ITEM_START.test(String(lead.text ?? '')))
if (starts) items.push({ depth: Number(u.depth) || 0, units: [u] })
else items[items.length - 1].units.push(u)
}
const list = h(block.ordered ? 'ol' : 'ul')
if (block.ordered && Number.isInteger(block.start) && block.start !== 1) list.setAttribute('start', String(block.start))
for (const item of items) {
const li = h('li', {}, item.units.map(inline))
if (item.depth) li.style.marginLeft = `${item.depth * 14}px`
list.append(li)
}
return list
}
case 'table': {
// 表头 row = -1;一格可能被分成几句,按位置归拢
const cells = new Map()
const rowIds = []
let cols = 0
for (const u of units) {
const loc = isObj(u.loc) ? u.loc : { row: 0, col: 0 }
push(cells, `${loc.row}:${loc.col}`, u)
cols = Math.max(cols, (Number(loc.col) || 0) + 1)
if (loc.row >= 0 && !rowIds.includes(loc.row)) rowIds.push(loc.row)
}
rowIds.sort((a, b) => a - b)
const textAt = (r, c) => (cells.get(`${r}:${c}`) || []).map(unitText).join('').replace(/[*`]/g, '').trim()
const numeric = Array.from({ length: cols }, (_, c) => {
const texts = rowIds.map((r) => textAt(r, c)).filter(Boolean)
return texts.length > 0 && texts.every((t) => NUMERIC_CELL.test(t))
})
const cell = (tag, r, c) => h(tag, { class: tag === 'td' && numeric[c] ? 'n' : null },
(cells.get(`${r}:${c}`) || []).map(inline))
return h('div', { class: 'table-wrap' }, h('table', {},
h('thead', {}, h('tr', {}, Array.from({ length: cols }, (_, c) => cell('th', -1, c)))),
h('tbody', {}, rowIds.map((r) => h('tr', {}, Array.from({ length: cols }, (_, c) => cell('td', r, c)))))))
}
default:
return h('p', {}, units.map(inline))
}
}
function renderDoc(doc, ctx) {
const blocks = Array.isArray(doc.blocks) ? doc.blocks.filter(isObj) : []
if (!blocks.length) return h('div', { class: 'doc' }, h('pre', {}, String(doc.markdown ?? '')))
return h('div', { class: 'doc' }, blocks.map((b) => renderBlock(b, ctx)))
}
// ------------------------------------------------------------------ 渲染:数字的出处
const view = { data: null, out: null, active: null, back: null }
const nodeLabel = (id) => {
const node = own(view.data.nodes, id)
return isObj(node) && node.label ? String(node.label) : String(id ?? '—')
}
const eventWhat = (ev) => [EVENT_LABEL[ev.type] || ev.type, ev.node_id ? nodeLabel(ev.node_id) : null].filter(Boolean).join(' · ')
function numberButton(number, body) {
const btn = h('button', {
type: 'button', class: 'num', 'data-num': number.key, 'data-result': number.result,
'aria-label': T.numAria(String(number.seg.text ?? ''), RESULT[number.result]),
onclick: () => openNumber(number),
}, body)
number.button = btn
return btn
}
function artifactState(id) {
const item = view.out.artifacts.find((a) => a.a.id === id)
if (!item) return ''
return [' ', badge(item.result), item.reason ? h('span', { class: 'muted' }, ` ${item.reason}`) : null]
}
function detailOf(number) {
const cite = isObj(number.seg.cite) ? number.seg.cite : null
const card = number.card
const rows = [
[T.dResult, [badge(number.result), number.reason ? ` ${number.reason}` : '']],
[T.dSource, number.source],
number.node ? [T.dNode, nodeLabel(number.node)] : null,
card ? [T.dCaliber, [card.caliber || '—', card.caliber_version ? ` ${card.caliber_version}` : ''].join('')] : null,
cite && cite.status === 'resolved' && 'value' in cite ? [T.dCited, valueText(cite.value)] : null,
'truth' in number ? [number.kind === 'metric' ? T.dCard : T.dCell, valueText(number.truth)] : null,
[T.dShown, [String(number.seg.text ?? ''), h('span', { class: 'muted' }, `(${T.dShownNote})`)]],
number.artifact ? [T.dArtifact, [h('code', {}, short(number.artifact)), artifactState(number.artifact)]] : null,
[T.dReport, nodeLabel(number.report.node_id)],
]
const parts = [kv(rows)]
const snap = number.snapshot
if (snap && typeof snap.sql === 'string' && snap.sql) parts.push(h('p', { class: 'sub' }, T.dSql), h('pre', { class: 'sql' }, snap.sql))
const record = snap && Number.isInteger(number.row) && Array.isArray(snap.rows) ? snap.rows[number.row] : undefined
if (Array.isArray(record) || isObj(record)) {
// 被引用的那一行:快照随文件附带(没有遮罩)才走到这里,整行照原样列出,引用的那一格高亮
let columns = (Array.isArray(snap.columns) ? snap.columns : []).map(String)
if (isObj(record) && !columns.length) columns = Object.keys(record)
const cellAt = (c, i) => (isObj(record) ? own(record, c) : record[i])
parts.push(h('p', { class: 'sub' }, T.dRow(number.row)), h('div', { class: 'table-wrap' }, h('table', {},
h('thead', {}, h('tr', {}, columns.map((c) => h('th', { class: c === number.column ? 'hit' : null }, c)))),
h('tbody', {}, h('tr', {}, columns.map((c, i) => h('td', { class: c === number.column ? 'hit' : null },
cellAt(c, i) == null ? '—' : textOf(cellAt(c, i)))))))))
}
return parts
}
function openNumber(number) {
const panel = $('detail')
if (panel.hidden) view.back = document.activeElement
if (view.active) view.active.classList.remove('is-active')
view.active = number.button || null
if (view.active) view.active.classList.add('is-active')
$('detail-title').textContent = String(number.seg.text ?? '')
$('detail-body').replaceChildren(...detailOf(number))
panel.hidden = false
panel.setAttribute('data-open', number.key)
$('detail-close').focus({ preventScroll: true })
// 窄屏上面板从底部抽出:点的数字被盖住时把它滚到面板上方
const sheet = panel.getBoundingClientRect()
const spot = view.active ? view.active.getBoundingClientRect() : null
if (spot && sheet.top > 0 && spot.bottom > sheet.top) window.scrollBy({ top: spot.top - Math.max(16, sheet.top / 3) })
}
function closeDetail() {
const panel = $('detail')
if (panel.hidden) return
panel.hidden = true
panel.removeAttribute('data-open')
if (view.active) view.active.classList.remove('is-active')
if (view.back && typeof view.back.focus === 'function') view.back.focus()
view.active = null
}
// ------------------------------------------------------------------ 渲染:页面各块
function renderHead(data) {
const meta = data.meta
$('title').textContent = String(meta.workflow_name || '')
$('seal-hash').textContent = String(meta.manifest_hash || '')
$('seal-note').textContent = T.sealNote(String(meta.run_id || ''))
let tier = '—'
if (isObj(meta.tier)) {
const ev = data.events.find((e) => e.seq === meta.tier.seq && e.type === 'issuance')
const body = ev ? view.out.checked.get(ev) : undefined
const ok = isObj(body) && body.tier === meta.tier.value
tier = h('span', { 'data-tier': ok ? 'checked' : 'unchecked' }, TIER_LABEL[meta.tier.value] || String(meta.tier.value),
h('span', { class: ok ? 'muted' : 'bad' }, `(${ok ? T.tierChecked(meta.tier.seq) : T.tierUnchecked})`))
}
const runClass = CLASS_LABEL[meta.run_class] || meta.run_class || '—'
const rows = [
[T.metaRun, h('code', {}, String(meta.run_id || ''))],
[T.metaWorkflow, String(meta.workflow_name || '—')],
[T.metaClass, meta.version != null ? `${runClass} v${meta.version}` : runClass],
meta.version_hash ? [T.metaVersionHash, h('code', {}, String(meta.version_hash))] : null,
[T.metaStarted, when(meta.started_at)],
[T.metaFinished, when(meta.finished_at)],
[T.metaRange, `${T.range(meta.manifest_seq, data.events.length)}${meta.legacy ? T.rangeLegacy : ''}`],
[T.metaTier, tier],
[T.metaExported, when(meta.exported_at)],
]
$('meta').replaceChildren(...rows.filter(Boolean).flatMap(([k, v]) => [h('dt', {}, k), h('dd', {}, v)]))
}
function renderReports(out) {
const root = $('reports')
if (!out.docs.length) {
root.append(h('p', { class: 'note' }, T.noReports))
return
}
const numbers = new Map(out.numbers.map((n) => [n.key, n]))
for (const item of out.docs) {
const label = nodeLabel(item.report.node_id)
const link = out.chain.find((c) => c.kind === 'report_doc' && c.report === item.report)
const result = worst([item.art || { result: NA }, link || { result: NA }])
const box = h('article', { class: 'report', 'data-report': item.report.node_id, 'data-result': result },
h('header', { class: 'report-head' }, h('h3', {}, T.reportTitle(label)), badge(result),
h('span', { class: 'mono muted' }, short(item.report.doc_artifact))))
if (!item.trusted) box.append(h('p', { class: 'warn' }, item.doc ? T.docUntrusted : T.docUnreadable))
if (item.doc) box.append(renderDoc(item.doc, { report: item.report, numbers }))
root.append(box)
}
}
function checkBlock(id, title, summary, items, line) {
const result = items.length ? worst(items) : NA
const block = h('section', { class: 'check', 'data-check': id, 'data-result': result },
h('div', { class: 'check-head' }, badge(result), h('h3', {}, title), h('span', { class: 'check-sum' }, summary)))
const list = (rows) => h('ul', { class: 'items' }, rows.map((item) => {
const li = h('li', { 'data-result': item.result }, badge(item.result), h('span', { class: 'what' }, line(item)))
if (item.reason) li.append(h('span', { class: 'why' }, item.reason))
return li
}))
if (items.length <= 20) {
block.append(list(items))
} else {
// 条目多(事件常有几百条):先列没通过的,全部的收起来
block.append(list(items.filter((i) => i.result === FAIL)))
block.append(h('details', {}, h('summary', {}, T.showAll(items.length)), list(items)))
}
return block
}
function renderChecks(out) {
const root = $('checks')
root.append(checkBlock('events', T.checkEvents, T.sumEvents(out.events.length), out.events,
(i) => `${T.chainByEvent(i.ev.seq, eventWhat(i.ev))}`))
const m = out.manifest
root.append(checkBlock('manifest', T.checkManifest, T.sumManifest(m.count), [m],
() => [T.manifestComputed, ' ', h('code', {}, m.computed)]))
root.append(checkBlock('artifacts', T.checkArtifacts, T.sumCount(out.artifacts.length, '份'), out.artifacts,
(i) => [KIND_LABEL[i.a.kind] || i.a.kind, i.a.node_id ? ` · ${nodeLabel(i.a.node_id)}` : '', ' · ',
h('code', {}, short(i.a.id))]))
root.append(checkBlock('chain', T.checkChain, T.sumCount(out.chain.length, '条'), out.chain, (c) => {
if (c.kind === 'report_doc') {
const from = c.ev ? ` ← ${T.chainByEvent(c.ev.seq, eventWhat(c.ev))}` : ''
return [T.chainDoc(nodeLabel(c.report.node_id)), ' ', h('code', {}, short(c.report.doc_artifact)), from]
}
const refs = [...c.docs.map((d) => T.chainByDoc(nodeLabel(d.report.node_id), d.alias)),
...c.events.map((ev) => T.chainByEvent(ev.seq, eventWhat(ev)))]
return [KIND_LABEL[c.kind] || c.kind, ' ', h('code', {}, short(c.art.a.id)),
refs.length ? ` ← ${T.chainFrom}${refs.join('、')}` : '']
}))
root.append(checkBlock('numbers', T.checkNumbers, T.sumCount(out.numbers.length, '个'), out.numbers,
(n) => h('button', { type: 'button', 'data-num-item': n.key, onclick: () => {
if (n.button) n.button.scrollIntoView({ block: 'center' })
openNumber(n)
} }, `${String(n.seg.text ?? '')} · ${n.source}`)))
}
function renderAppended(data) {
const later = Array.isArray(data.appended) ? data.appended.filter(isObj) : []
if (!later.length) return
const root = $('appended')
root.hidden = false
root.append(h('p', { class: 'note' }, T.appendedNote), h('ul', { class: 'items check' }, later.map((ev) =>
h('li', { 'data-appended': ev.seq }, h('span', { class: 'badge', 'data-result': NA }, T.appendedTag),
h('span', { class: 'what' }, `${T.chainByEvent(ev.seq, eventWhat(ev))}${ev.summary ? `:${ev.summary}` : ''}`)))))
}
function renderOverall(out) {
const all = [...out.events, out.manifest, ...out.artifacts, ...out.chain, ...out.numbers]
const count = (r) => all.filter((i) => i.result === r).length
const [pass, fail, na] = [count(PASS), count(FAIL), count(NA)]
const result = fail ? FAIL : PASS
const box = $('overall')
box.setAttribute('data-overall', result)
box.querySelector('.overall-mark').textContent = MARK[result]
$('overall-text').textContent = fail ? T.failed(fail) : na ? T.passWithNa : T.allPass
$('overall-sub').textContent = T.tally(pass, fail, na)
}
function renderFoot(data, kind) {
$('foot').replaceChildren(
h('p', { 'data-sha': kind }, kind === 'subtle' ? T.shaSubtle : T.shaJs),
h('p', {}, T.renderNote),
h('p', {}, T.schema(String(data.schema || ''))))
}
/** 下不了结论:总览写明原因。也标上核对结束(data-verified="fatal"),等着看结果的一方不必干等 */
function fatal(message) {
const box = $('overall')
box.setAttribute('data-overall', FAIL)
box.querySelector('.overall-mark').textContent = MARK.fail
$('overall-text').textContent = message
document.documentElement.setAttribute('data-verified', 'fatal')
}
async function main() {
let data
try {
data = JSON.parse($('bundle-data').textContent)
if (!isObj(data) || !isObj(data.meta) || !Array.isArray(data.events)) throw new Error('bad data')
} catch (e) {
fatal(T.broken)
return
}
view.data = data
const picked = await pickDigest()
if (!picked) {
fatal(T.selfTestFailed)
return
}
view.out = await verify(data, picked.digest)
renderHead(data)
renderReports(view.out)
renderChecks(view.out)
renderAppended(data)
renderFoot(data, picked.kind)
renderOverall(view.out)
$('detail-close').addEventListener('click', closeDetail)
document.addEventListener('keydown', (e) => { if (e.key === 'Escape') closeDetail() })
document.documentElement.setAttribute('data-verified', '')
}
main().catch((e) => {
console.error(e)
fatal(T.broken)
})
})()
+18
View File
@@ -97,6 +97,24 @@ def load(artifact_id: str) -> Any | None:
return json.loads(text)
def read_bytes(artifact_id: str, *, verify: bool = True) -> bytes | None:
"""按 id 取回工件文件的原始字节。文件本身就是 put_json 写下的规范 JSON,字节原样,不解析、不重排。
可验证报告要把工件按原字节嵌进页面,让拿到文件的人自己重算哈希,所以不能像 load 那样解析了再交出去。
默认同样复验哈希、对不上就抛 ValueError(和 load 一样的承诺);verify=False 只给要把字节原样交给别人
去核的调用方——被改过的工件照实交出去,核的那一方就会看到对不上。id 不对或者文件不在了返回 None。
"""
if not artifact_id or not artifact_id.isalnum():
return None
path = _path_of(artifact_id)
if not path.exists():
return None
data = path.read_bytes()
if verify and content_hash(data) != artifact_id:
raise ValueError(f"工件 {artifact_id[:12]}… 内容与哈希不符,疑似被篡改")
return data
def manifest_hash(rows: list[tuple[int, str, str | None, dict[str, Any]]]) -> str:
"""把一次运行的全部事件压成一个清单哈希。
+340
View File
@@ -0,0 +1,340 @@
"""可验证报告的接口:GET /api/runs/{run_id}/bundle.html。
照证据测试的夹具真跑一次:两个虚构的 SQLite 数据源(门店流水没设遮罩,会员库遮了 phone 列),口径卡、
报告引用两份快照里的格子和一个指标。核对:
- 只给已封存且封存核对一致的运行:没封存、封存后被改过都回 409;
- 导出的数据里,事件行按浏览器的拼法重算等于运行记下的 manifest_hash;
- 遮罩库的快照只有 id 没有内容,它的查询事件只有哈希;
- 附件名是「<运行名>-可验证报告.html」,中文名按 RFC 5987 编码;
- 只读:导出前后运行记录、事件一条不多一条不少。
另外核对工件库新加的 read_bytes:原样取回文件字节。
"""
from __future__ import annotations
import asyncio
import base64
import hashlib
import json
import re
import sqlite3
from urllib.parse import quote
import pytest
from httpx import ASGITransport, AsyncClient
from langchain_core.messages import AIMessage
from sqlalchemy import func, select
from app.core.artifact_store import canonical_json, content_hash, put_json, read_bytes
from app.core.config import settings
from app.data.engine import engines
from app.db.base import SessionLocal
from app.db.models import DataSource, Run, RunEvent
from app.engine.runner import run_manager
from app.main import app
SHOP_SQL = "SELECT id, store, amount FROM orders ORDER BY id"
CRM_SQL = "SELECT id, name, phone, spend FROM members ORDER BY id"
TEXT = ("## 门店周报\n\n青石店本周流水 [[v:Q1.r0.amount]],会员消费 [[v:Q2.r1.spend]],"
"口径卡销售额 [[m:gmv]]。[[see:Q1]]")
@pytest.fixture(autouse=True)
async def engine_up():
await run_manager.setup()
yield
await run_manager.shutdown()
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
async def set_source(name, path, options=None):
async with SessionLocal() as session:
row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none()
if row is None:
row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True)
session.add(row)
row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
row.options = dict(options or {})
await session.commit()
return row.id
@pytest.fixture(autouse=True)
async def sources(tmp_path):
shop, crm = tmp_path / "shop.db", tmp_path / "crm.db"
db = sqlite3.connect(shop)
db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, store TEXT, amount REAL);")
db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "青石店", 1280.5), (2, "白沙店", 960.0)])
db.commit()
db.close()
db = sqlite3.connect(crm)
db.executescript("CREATE TABLE members (id INTEGER PRIMARY KEY, name TEXT, phone TEXT, spend REAL);")
db.executemany("INSERT INTO members VALUES (?, ?, ?, ?)", [(1, "林一", "000-0001", 320.0),
(2, "周二", "000-0002", 188.5)])
db.commit()
db.close()
ids = [await set_source("bundle_shop", shop), await set_source("bundle_crm", crm, {"mask_columns": ["phone"]})]
for source_id in ids:
await engines.invalidate(source_id)
yield
for source_id in ids:
await engines.invalidate(source_id)
def node(nid, ntype, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
def weekly(text_nodes=()):
nodes = [
node("start", "input"),
node("fetch", "tool", tool="db_query__bundle_shop", args={"sql": SHOP_SQL}),
node("members", "tool", tool="db_query__bundle_crm", args={"sql": CRM_SQL}),
node("card", "metrics", caliber="周报口径", caliber_version="v2", metrics=[
{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands",
"expression": "cell(nodes.fetch, 0, 'amount')"}]),
node("write", "report", instructions="写周报", on_violation="flag"),
*text_nodes,
node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
]
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
def script(monkeypatch, text=TEXT):
from app.providers import mock_model
monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
async def wait(run_id: str) -> Run:
for _ in range(400):
await asyncio.sleep(0.05)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
return row
raise AssertionError(f"run {run_id} 没跑完:{row.status}")
async def sealed_run(monkeypatch, text=TEXT) -> Run:
script(monkeypatch, text)
row = await wait((await run_manager.start(graph=weekly(), input_payload={})).id)
assert row.status == "succeeded", row.error
return row
def data_of(html: str) -> dict:
m = re.search(r'<script type="application/json" id="bundle-data">(.*?)</script>', html, re.S)
assert m, "页面里没有数据块"
return json.loads(m.group(1))
async def query_artifact(run_id: str, node_id: str) -> str:
async with SessionLocal() as session:
[end] = (await session.execute(select(RunEvent).where(
RunEvent.run_id == run_id, RunEvent.type == "tool.end", RunEvent.node_id == node_id))).scalars()
return end.data["query_artifact"]
def file_of(artifact: str):
return settings.data_dir / "artifacts" / artifact[:2] / f"{artifact}.json"
# --------------------------------------------------------------------------
# 导出
# --------------------------------------------------------------------------
async def test_export_rebuilds_the_runners_manifest(client, monkeypatch):
row = await sealed_run(monkeypatch)
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 200, resp.text
assert resp.headers["content-type"].startswith("text/html")
bundle = data_of(resp.text)
meta = bundle["meta"]
assert meta["run_id"] == row.id and meta["manifest_hash"] == row.manifest_hash
assert meta["manifest_seq"] == row.manifest_seq and meta["legacy"] is False
lines = [f"{e['seq']}|{e['type']}|{e['node_id'] or ''}|{e['data_sha256']}"
for e in sorted(bundle["events"], key=lambda e: e["seq"])]
assert hashlib.sha256("\n".join(lines).encode()).hexdigest() == row.manifest_hash
assert max(e["seq"] for e in bundle["events"]) == row.manifest_seq
async with SessionLocal() as session:
rows = {r.seq: r for r in (await session.execute(
select(RunEvent).where(RunEvent.run_id == row.id))).scalars()}
for item in bundle["events"]:
if "canonical" in item:
assert item["canonical"] == canonical_json(rows[item["seq"]].data)
assert content_hash(item["canonical"]) == item["data_sha256"]
types = {e["type"] for e in bundle["events"] if "canonical" in e}
assert {"run.started", "report.checked", "tool.end", "node.finished"} <= types
assert not types & {"llm.start", "llm.end", "node.started", "tool.start", "log"}
async def test_the_masked_source_ships_only_hashes(client, monkeypatch):
row = await sealed_run(monkeypatch)
shop, crm = await query_artifact(row.id, "fetch"), await query_artifact(row.id, "members")
bundle = data_of((await client.get(f"/api/runs/{row.id}/bundle.html")).text)
by_id = {a["id"]: a for a in bundle["artifacts"]}
assert by_id[crm] == {"id": crm, "kind": "query", "node_id": "members", "masked": True}
assert base64.b64decode(by_id[shop]["b64"]) == file_of(shop).read_bytes()
report = next(a for a in bundle["artifacts"] if a["kind"] == "report_doc")
assert json.loads(base64.b64decode(report["b64"]))["run_id"] == row.id
assert {a["kind"] for a in bundle["artifacts"]} == {"report_doc", "metric_set", "query"}
ends = {e["node_id"]: e for e in bundle["events"] if e["type"] == "tool.end"}
assert "canonical" in ends["fetch"] and "canonical" not in ends["members"]
# 运行里有带遮罩的快照:成果节点、运行结束事件的预览可能带出查询结果,只给哈希
out = next(e for e in bundle["events"] if e["type"] == "node.finished" and e["node_id"] == "out")
finished = next(e for e in bundle["events"] if e["type"] == "run.finished")
assert "canonical" not in out and "canonical" not in finished
assert "000-0001" not in json.dumps(bundle, ensure_ascii=False)
async def test_the_page_is_named_after_the_run(client, monkeypatch):
row = await sealed_run(monkeypatch)
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
header = resp.headers["content-disposition"]
assert header.startswith("attachment;")
assert f"filename*=UTF-8''{quote('未保存的工作流-可验证报告.html', safe='')}" in header
assert resp.headers.get("cache-control") == "no-store"
async with SessionLocal() as session:
run = await session.get(Run, row.id)
run.workflow_name = "门店周报"
await session.commit()
header = (await client.get(f"/api/runs/{row.id}/bundle.html")).headers["content-disposition"]
assert f"filename*=UTF-8''{quote('门店周报-可验证报告.html', safe='')}" in header
async def test_report_text_cannot_close_the_data_block(client, monkeypatch):
row = await sealed_run(monkeypatch, text="结论见下:</script><script>alert(1)</script> 流水 [[v:Q1.r0.amount]]。")
html = (await client.get(f"/api/runs/{row.id}/bundle.html")).text
assert html.count("</script>") == 2
bundle = data_of(html)
write = next(e for e in bundle["events"] if e["type"] == "node.finished" and e["node_id"] == "write")
assert "</script><script>alert(1)</script>" in write["canonical"]
async def test_events_after_the_seal_are_listed_apart(client, monkeypatch):
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
session.add(RunEvent(run_id=row.id, seq=row.manifest_seq + 3, type="evidence.judged", node_id="write",
ts=0.0, data={"verdicts": {"u1": {"verdict": "supported"}}, "model": "judge-x",
"post_seal": True}))
await session.commit()
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 200, resp.text
bundle = data_of(resp.text)
assert max(e["seq"] for e in bundle["events"]) == row.manifest_seq
[later] = bundle["appended"]
assert later["seq"] == row.manifest_seq + 3 and later["type"] == "evidence.judged"
# --------------------------------------------------------------------------
# 拒绝、只读
# --------------------------------------------------------------------------
async def test_an_unsealed_run_is_refused(client):
async with SessionLocal() as session:
run = Run(workflow_name="门店周报", status="running", graph={"nodes": [], "edges": []})
session.add(run)
await session.commit()
run_id = run.id
resp = await client.get(f"/api/runs/{run_id}/bundle.html")
assert resp.status_code == 409
assert resp.json() == {"detail": "运行尚未结束封存,结束后才能导出", "code": "bundle_unsealed"}
@pytest.mark.parametrize("status", ["running", "interrupted", "queued"])
async def test_a_run_going_again_on_an_old_seal_is_refused(client, monkeypatch, status):
"""失败后接着跑的运行留着上一次的封存,跑完会重新封存、换掉清单哈希;早年的运行停在审批时也封存过。
这时导出去的文件,封存哈希和系统里之后记的对不上,页面上「去系统里对照」那一步就作废了:一样回 409。"""
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
run = await session.get(Run, row.id)
run.status = status
await session.commit()
assert (await client.get(f"/api/runs/{row.id}/verify")).json()["ok"] is True
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 409
assert resp.json() == {"detail": "运行尚未结束封存,结束后才能导出", "code": "bundle_unsealed"}
async def test_a_failed_run_is_exportable(client, monkeypatch):
"""失败、取消的运行同样封存了:报告已经写出来的,照样导得出去。"""
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
run = await session.get(Run, row.id)
run.status = "failed"
await session.commit()
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 200, resp.text
assert data_of(resp.text)["meta"]["status"] == "failed"
async def test_a_broken_seal_is_refused(client, monkeypatch):
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
event = (await session.execute(select(RunEvent).where(
RunEvent.run_id == row.id, RunEvent.type == "report.checked"))).scalar_one()
event.data = {**event.data, "ok": not event.data.get("ok")}
await session.commit()
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 409
body = resp.json()
assert body["code"] == "bundle_seal_broken" and "封存核对不一致" in body["detail"], body
async def test_an_unknown_run_is_404(client):
resp = await client.get("/api/runs/nope/bundle.html")
assert resp.status_code == 404
async def test_exporting_appends_nothing(client, monkeypatch):
row = await sealed_run(monkeypatch)
async def snapshot():
async with SessionLocal() as session:
run = await session.get(Run, row.id)
count = (await session.execute(
select(func.count()).select_from(RunEvent).where(RunEvent.run_id == row.id))).scalar_one()
return run.last_seq, count, run.manifest_hash, run.manifest_seq, run.status
before = await snapshot()
for _ in range(2):
assert (await client.get(f"/api/runs/{row.id}/bundle.html")).status_code == 200
assert await snapshot() == before
assert (await client.get(f"/api/runs/{row.id}/verify")).json()["ok"] is True
# --------------------------------------------------------------------------
# 工件库:read_bytes
# --------------------------------------------------------------------------
async def test_read_bytes_returns_the_file_as_stored():
artifact = await put_json({"门店": "青石店", "流水": 1280.5}, kind="test")
assert read_bytes(artifact) == file_of(artifact).read_bytes()
assert read_bytes(artifact) == canonical_json({"门店": "青石店", "流水": 1280.5}).encode()
assert read_bytes("0" * 64) is None and read_bytes("../etc/passwd") is None and read_bytes("") is None
async def test_read_bytes_checks_the_hash_unless_told_not_to():
artifact = await put_json({"门店": "白沙店", "流水": 960.0}, kind="test")
path = file_of(artifact)
original = path.read_bytes()
path.write_bytes(original.replace(b"960.0", b"999.0"))
try:
with pytest.raises(ValueError):
read_bytes(artifact)
assert read_bytes(artifact, verify=False) == original.replace(b"960.0", b"999.0")
finally:
path.write_bytes(original)
+351
View File
@@ -0,0 +1,351 @@
"""可验证报告的构建(bundle/build.py):纯函数,不碰数据库。
拿一份手搭的已封存运行(事件、报告文档、口径卡、两份查询快照,其中一份带遮罩列)核对:
- 每条封存范围内的事件都带 data_sha256,按浏览器的拼法(seq|type|node_id|data_sha256 按 seq 排、换行连接)
重算出来的清单哈希等于 runner 记下的那个;
- 只有白名单里的事件带 canonical(逐字等于 canonical_json(data)),其余只给哈希;运行里有带遮罩的快照时,
成果节点和运行结束事件的预览可能带出查询结果,也只给哈希;
- 报告文档、它引用到的口径卡和查询快照按原字节嵌入;带遮罩的快照只给 id;没被引用的工件不嵌;
- 封存之后追加的事件单列,不进清单;
- 页面零外链,数据里的 </script> 截不断数据块。
"""
from __future__ import annotations
import base64
import hashlib
import json
import re
from urllib.parse import quote
import pytest
from app.bundle.build import (
SCHEMA, attachment_name, build_bundle, bundle_json, content_disposition, display_name, render_html,
)
from app.core.artifact_store import canonical_json, content_hash, manifest_hash
def stored(obj) -> tuple[str, bytes]:
"""照工件库的样子落一份:规范 JSON 的字节,id 是它的 sha256。"""
raw = canonical_json(obj).encode()
return content_hash(raw), raw
SNAP_OPEN = {"columns": ["store", "amount"], "rows": [["青石店", 1280.5], ["白沙店", "960.00"]],
"column_types": {"store": "text", "amount": "number"}, "source": "shop",
"sql": "SELECT store, amount FROM orders", "truncated": False}
SNAP_MASKED = {"columns": ["name", "phone", "spend"], "rows": [["林一", "000-0001", 320.0]],
"source": "crm", "sql": "SELECT name, phone, spend FROM members", "mask_columns": ["phone"]}
SNAP_UNCITED = {"columns": ["n"], "rows": [[7]], "source": "shop", "sql": "SELECT 7 AS n"}
RETRIEVAL = {"hits": [{"content": "退款按发生周统计"}]}
OPEN_ID, OPEN_RAW = stored(SNAP_OPEN)
MASKED_ID, MASKED_RAW = stored(SNAP_MASKED)
UNCITED_ID, UNCITED_RAW = stored(SNAP_UNCITED)
RETRIEVAL_ID, RETRIEVAL_RAW = stored(RETRIEVAL)
CARD = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v2",
"metrics": [{"id": "gmv", "name": "销售额", "value": 1280.5, "unit": "元", "decimals": 1}]}
CARD_ID, CARD_RAW = stored(CARD)
def cite(alias, kind, value, **locator):
return {"alias": alias, "kind": kind, "locator": locator, "role": "value", "status": "resolved",
"value": value, "rendered": str(value)}
def doc_of(run_id="r1", node_id="write", extra_catalog=None):
return {
"schema": "agentlab.report/1", "run_id": run_id, "node_id": node_id, "markdown": "…",
"catalog": {
"m:gmv": {"alias": "m:gmv", "kind": "metric", "artifact": CARD_ID, "node_id": "card",
"locator": {"metric": "gmv"}, "value": 1280.5},
"Q1": {"alias": "Q1", "kind": "query", "artifact": OPEN_ID, "node_id": "fetch"},
"Q2": {"alias": "Q2", "kind": "query", "artifact": MASKED_ID, "node_id": "members"},
"Q3": {"alias": "Q3", "kind": "query", "artifact": UNCITED_ID, "node_id": "fetch"},
"K1": {"alias": "K1", "kind": "retrieval", "artifact": RETRIEVAL_ID, "node_id": "docs"},
"i:week": {"alias": "i:week", "kind": "input", "value": "2026-W37", "locator": {"field": "week"}},
**(extra_catalog or {}),
},
"blocks": [{"id": "b0", "type": "paragraph", "units": [{
"id": "u0", "kind": "claim", "cites": ["m:gmv", "Q1", "Q2", "K1"],
"see": [{"alias": "K1", "kind": "retrieval", "role": "support", "status": "resolved"}],
"segments": [
{"id": "s0", "kind": "number", "text": "1,280.5元", "state": "deterministic",
"cite": cite("m:gmv", "metric", 1280.5, metric="gmv")},
{"id": "s1", "kind": "number", "text": "960", "state": "deterministic",
"cite": cite("Q1", "cell", 960, row=1, column="amount")},
{"id": "s2", "kind": "number", "text": "320", "state": "deterministic",
"cite": cite("Q2", "cell", 320.0, row=0, column="spend")},
{"id": "s3", "kind": "number", "text": "42", "state": "none", "issue": "uncited_number"},
]}]}],
}
DOC = doc_of()
DOC_ID, DOC_RAW = stored(DOC)
STORE = {DOC_ID: DOC_RAW, CARD_ID: CARD_RAW, OPEN_ID: OPEN_RAW, MASKED_ID: MASKED_RAW,
UNCITED_ID: UNCITED_RAW, RETRIEVAL_ID: RETRIEVAL_RAW}
RUN = {"id": "r1", "workflow_id": "wf1", "workflow_name": "门店周报", "version": 3, "version_hash": "ab" * 32,
"run_class": "formal", "status": "succeeded",
"started_at": "2026-09-07T01:00:00+00:00", "finished_at": "2026-09-07T01:02:03+00:00"}
GRAPH = {"nodes": [{"id": n, "type": t, "data": {"label": label}} for n, t, label in [
("start", "input", "开始"), ("fetch", "tool", "门店流水"), ("members", "tool", "会员消费"),
("docs", "retrieve", "口径说明"), ("card", "metrics", "周报口径"), ("write", "report", "写周报"),
("out", "output", "成果")]]}
def sample_events(doc_id=DOC_ID):
return [
(1, "run.started", None, {"nodes": 7, "memory_scope": "default"}),
(2, "node.started", "start", {"label": "开始"}),
(3, "node.finished", "start", {"preview": {"week": "2026-W37"}, "artifact": "f" * 64}),
(4, "tool.end", "fetch", {"tool": "db_query__shop", "preview": "[[\"青石店\", 1280.5]]",
"artifact": "e" * 64, "query_artifact": OPEN_ID}),
(5, "node.finished", "fetch", {"preview": "{\"rows\": …}", "evidence": [
{"kind": "query", "node_id": "fetch", "artifact": OPEN_ID}]}),
(6, "tool.end", "members", {"tool": "db_query__crm", "preview": "[[\"林一\", \"000-0001\"]]",
"query_artifact": MASKED_ID}),
(7, "tool.end", "docs", {"tool": "web_search", "preview": "外部结果"}),
(8, "node.finished", "card", {"preview": {"kind": "metric_set"}, "evidence": [
{"kind": "metric_set", "node_id": "card", "artifact": CARD_ID}]}),
(9, "caliber.upgrade", "card", {"caliber_node": "card", "pinned": "v2", "latest": "v3", "policy": "pin"}),
(10, "llm.end", "write", {"model": "mock", "output": "系统提示和草稿"}),
(11, "report.checked", "write", {"doc_artifact": doc_id, "ok": True}),
(12, "node.finished", "write", {"preview": {"text": "青石店本周流水…", "doc_artifact": doc_id}}),
(13, "issuance", "out", {"tier": "formal", "mode": "citations"}),
(14, "node.finished", "out", {"preview": {"周报": "…", "明细": "[[\"林一\", \"000-0001\"]]"}}),
(15, "run.finished", None, {"output": {"周报": "…"}}),
# 封存之后追加:按需裁判的判定
(16, "evidence.judged", "write", {"verdicts": {"u0": {"verdict": "supported"}}, "model": "judge-x",
"post_seal": True}),
]
def build(events=None, store=None, source_masks=None, **kw):
data = store if store is not None else STORE
return build_bundle(RUN, events if events is not None else sample_events(), manifest_seq=kw.pop("seq", 15),
manifest_hash=kw.pop("manifest", "0" * 64), graph=GRAPH, source_masks=source_masks or {},
read=data.get, exported_at="2026-10-05T08:00:00+00:00", **kw)
def browser_manifest(events: list[dict]) -> str:
"""照浏览器里 verifier.js 的拼法重算:每行 seq|type|node_id|data_sha256,node_id 为空写空串,按 seq 排序。"""
lines = [f"{e['seq']}|{e['type']}|{e['node_id'] or ''}|{e['data_sha256']}"
for e in sorted(events, key=lambda e: e["seq"])]
return hashlib.sha256("\n".join(lines).encode()).hexdigest()
# --------------------------------------------------------------------------
# 事件:哈希、清单、规范字符串
# --------------------------------------------------------------------------
def test_event_hashes_rebuild_the_runners_manifest():
events = sample_events()
bundle = build()
assert bundle["schema"] == SCHEMA
sealed = [e for e in events if e[0] <= 15]
assert [e["seq"] for e in bundle["events"]] == [e[0] for e in sealed]
for item, (seq, etype, node_id, data) in zip(bundle["events"], sealed):
assert (item["seq"], item["type"], item["node_id"]) == (seq, etype, node_id)
assert item["data_sha256"] == content_hash(canonical_json(data))
assert browser_manifest(bundle["events"]) == manifest_hash(sealed)
def test_only_whitelisted_events_carry_their_canonical_text():
events = {e[0]: e for e in sample_events()}
bundle = build()
shown = {e["seq"]: e["canonical"] for e in bundle["events"] if "canonical" in e}
# run.started、无遮罩的查询 tool.end、口径卡和报告的 node.finished、升版处置、报告自查、出具判定
assert sorted(shown) == [1, 4, 8, 9, 11, 12, 13], sorted(shown)
for seq, text in shown.items():
assert text == canonical_json(events[seq][3])
assert hashlib.sha256(text.encode()).hexdigest() == next(
e["data_sha256"] for e in bundle["events"] if e["seq"] == seq)
def test_output_and_finish_show_content_only_when_nothing_is_masked():
"""成果字段可以直接引用查询结果:运行里有带遮罩的快照时,成果节点和运行结束事件只给哈希。"""
unmasked = {**SNAP_MASKED}
unmasked.pop("mask_columns")
plain_id, plain_raw = stored(unmasked)
doc = doc_of(extra_catalog={"Q2": {"alias": "Q2", "kind": "query", "artifact": plain_id, "node_id": "members"}})
doc_id, doc_raw = stored(doc)
events = [(s, t, n, {**d, "query_artifact": plain_id} if s == 6 else d) for s, t, n, d in sample_events(doc_id)]
store = {**STORE, plain_id: plain_raw, doc_id: doc_raw}
bundle = build(events, store)
shown = sorted(e["seq"] for e in bundle["events"] if "canonical" in e)
assert shown == [1, 4, 6, 8, 9, 11, 12, 13, 14, 15], shown
def test_masks_set_on_the_source_now_count_too():
"""快照查询当时没记遮罩、数据源事后设了遮罩:照样算带遮罩——证据面板也按两者之和遮。"""
bundle = build(source_masks={"shop": ["store"]})
assert 4 not in {e["seq"] for e in bundle["events"] if "canonical" in e}
entry = next(a for a in bundle["artifacts"] if a["id"] == OPEN_ID)
assert entry["masked"] is True and "b64" not in entry
def test_an_unreadable_snapshot_counts_as_masked():
"""读不出来的快照不知道带没带遮罩:它的查询事件只给哈希,成果和运行结束事件也一样。"""
store = {k: v for k, v in STORE.items() if k != OPEN_ID}
bundle = build(store=store)
shown = {e["seq"] for e in bundle["events"] if "canonical" in e}
assert 4 not in shown and 14 not in shown and 15 not in shown
entry = next(a for a in bundle["artifacts"] if a["id"] == OPEN_ID)
assert entry["missing"] is True and "b64" not in entry
# --------------------------------------------------------------------------
# 工件:只嵌报告文档和它引用到的口径卡、查询快照
# --------------------------------------------------------------------------
def test_report_doc_and_what_it_cites_are_embedded_byte_for_byte():
bundle = build()
by_id = {a["id"]: a for a in bundle["artifacts"]}
assert [a["id"] for a in bundle["artifacts"]] == [DOC_ID, CARD_ID, OPEN_ID, MASKED_ID]
assert by_id[DOC_ID]["kind"] == "report_doc" and by_id[DOC_ID]["node_id"] == "write"
assert by_id[CARD_ID]["kind"] == "metric_set" and by_id[CARD_ID]["node_id"] == "card"
assert by_id[OPEN_ID]["kind"] == "query" and by_id[OPEN_ID]["node_id"] == "fetch"
for art in (DOC_ID, CARD_ID, OPEN_ID):
assert base64.b64decode(by_id[art]["b64"]) == STORE[art]
assert bundle["reports"] == [{"node_id": "write", "doc_artifact": DOC_ID, "seq": 11}]
def test_a_masked_snapshot_ships_only_its_id():
entry = next(a for a in build()["artifacts"] if a["id"] == MASKED_ID)
assert entry == {"id": MASKED_ID, "kind": "query", "node_id": "members", "masked": True}
assert "000-0001" not in render_html(build())
def test_artifacts_the_report_does_not_cite_stay_out():
"""目录里有、正文没引用的查询(Q3),检索快照,节点产出、工具快照:都不嵌。"""
ids = {a["id"] for a in build()["artifacts"]}
assert UNCITED_ID not in ids and RETRIEVAL_ID not in ids
assert "f" * 64 not in ids and "e" * 64 not in ids
def test_the_latest_report_checked_of_a_node_wins():
"""同一个报告节点自查了两次(重写过):和证据接口一样认最后一次。"""
old_id, old_raw = stored({**DOC, "markdown": "旧稿"})
events = sample_events()
events.insert(10, (10, "report.checked", "write", {"doc_artifact": old_id, "ok": False}))
events = [(i + 1, t, n, d) for i, (_, t, n, d) in enumerate(events)]
bundle = build(events, {**STORE, old_id: old_raw}, seq=16)
assert bundle["reports"] == [{"node_id": "write", "doc_artifact": DOC_ID, "seq": 12}]
assert old_id not in {a["id"] for a in bundle["artifacts"]}
def test_a_doc_from_another_run_is_embedded_but_not_followed():
"""文档记的运行不是这一次:照实嵌进去让浏览器标出来,但它的目录不可信,不顺着它找工件。"""
stray = doc_of(run_id="other")
stray_id, stray_raw = stored(stray)
bundle = build(sample_events(stray_id), {**STORE, stray_id: stray_raw})
assert [a["id"] for a in bundle["artifacts"]] == [stray_id]
def test_a_missing_or_changed_artifact_is_listed_for_the_browser_to_judge():
"""服务端的文件不在了:只给 id 并标明;被改过的:原样嵌进去,浏览器自己算出对不上。"""
changed = CARD_RAW.replace(b"1280.5", b"1999.5")
bundle = build(store={**STORE, CARD_ID: changed, OPEN_ID: None})
by_id = {a["id"]: a for a in bundle["artifacts"]}
assert base64.b64decode(by_id[CARD_ID]["b64"]) == changed
assert by_id[OPEN_ID] == {"id": OPEN_ID, "kind": "query", "node_id": "fetch", "missing": True}
# --------------------------------------------------------------------------
# 元信息、封存之后追加的事件
# --------------------------------------------------------------------------
def test_meta_names_the_seal_and_the_run():
bundle = build(manifest="ab" * 32)
meta = bundle["meta"]
assert meta["run_id"] == "r1" and meta["workflow_name"] == "门店周报" and meta["workflow_id"] == "wf1"
assert meta["version"] == 3 and meta["run_class"] == "formal"
assert meta["manifest_seq"] == 15 and meta["manifest_hash"] == "ab" * 32 and meta["legacy"] is False
assert meta["started_at"] == RUN["started_at"] and meta["finished_at"] == RUN["finished_at"]
assert meta["exported_at"] == "2026-10-05T08:00:00+00:00"
# 出具档位取封存范围内最后一条出具判定事件,带上序号,浏览器拿已核对的事件对照
assert meta["tier"] == {"value": "formal", "seq": 13}
assert bundle["nodes"]["write"] == {"label": "写周报", "type": "report"}
def test_events_after_the_seal_are_listed_apart():
bundle = build()
assert 16 not in {e["seq"] for e in bundle["events"]}
[later] = bundle["appended"]
assert later["seq"] == 16 and later["type"] == "evidence.judged" and later["node_id"] == "write"
assert "1 句" in later["summary"] and "judge-x" in later["summary"]
assert "canonical" not in later and "data_sha256" not in later
def test_display_name_and_attachment_name():
assert display_name({"workflow_id": None, "workflow_name": "临时图"}) == "未保存的工作流"
assert display_name({"workflow_id": None, "workflow_name": ""}) == "未保存的工作流"
assert display_name({"workflow_id": "w", "workflow_name": "门店周报"}) == "门店周报"
assert attachment_name({"workflow_id": "w", "workflow_name": "门店/周报: v2?"}) == "门店_周报_ v2_-可验证报告.html"
assert attachment_name({"workflow_id": None, "workflow_name": ""}) == "未保存的工作流-可验证报告.html"
header = content_disposition("门店周报-可验证报告.html", run_id="0123456789abcdef")
assert header.startswith("attachment; ")
assert f"filename*=UTF-8''{quote('门店周报-可验证报告.html', safe='')}" in header
# 不认 filename* 的老客户端拿到一个纯 ASCII 的名字
fallback = re.search(r'filename="([^"]+)"', header).group(1)
assert fallback.isascii() and "01234567" in fallback
# --------------------------------------------------------------------------
# 页面:自包含、数据块截不断
# --------------------------------------------------------------------------
def data_block(html: str) -> str:
m = re.search(r'<script type="application/json" id="bundle-data">(.*?)</script>', html, re.S)
assert m, "页面里没有数据块"
return m.group(1)
def test_the_page_has_no_external_links():
html = render_html(build())
assert html.lstrip().lower().startswith("<!doctype html>")
assert not re.search(r"https?://", html), re.search(r".{0,40}https?://.{0,40}", html).group(0)
assert not re.search(r"<script[^>]*\ssrc\s*=", html, re.I)
assert not re.search(r"<(?:link|img|iframe|object|embed|audio|video|source)\b", html, re.I)
assert not re.search(r"url\s*\(|@import", html, re.I)
# 内容安全策略:默认什么都不许连,脚本只认内联的那一段核对脚本(按哈希)
csp = re.search(r'<meta http-equiv="Content-Security-Policy" content="([^"]+)"', html).group(1)
assert "default-src 'none'" in csp
script = re.search(r"<script>(.*?)</script>", html, re.S).group(1)
digest = base64.b64encode(hashlib.sha256(script.encode()).digest()).decode()
assert f"script-src 'sha256-{digest}'" in csp
def test_data_cannot_break_out_of_its_script_tag():
nasty = "</script><script>alert(1)</script><!-- & \u2028\u2029 >"
bundle = build()
bundle["meta"]["workflow_name"] = nasty
bundle["events"][0]["canonical"] = nasty
text = bundle_json(bundle)
assert not re.search("[<>&\u2028\u2029]", text)
assert json.loads(text) == bundle
html = render_html(bundle)
assert html.count("</script>") == 2 # 数据块和核对脚本各一个,再没有别的
assert json.loads(data_block(html)) == bundle
title = re.search(r"<title>(.*?)</title>", html, re.S).group(1)
assert "<script>" not in title and "&lt;/script&gt;" in title
def test_the_verifier_script_is_inline_and_cannot_close_early():
html = render_html(build())
script = re.search(r"<script>(.*?)</script>", html, re.S).group(1)
assert "</script" not in script.lower() and "<!--" not in script
assert "bundle-data" in script and "crypto.subtle" in script.replace("window.", "")
@pytest.mark.parametrize("legacy", [True, False])
def test_legacy_seals_are_flagged(legacy):
assert build(legacy=legacy)["meta"]["legacy"] is legacy