mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 02:21:58 +00:00
合入期 5 前置修复 L2：可验证报告里设了遮罩的数据源上、能证明没取遮罩列的查询快照照常随文件附带，引用它的数字可以核对；解析不了、可能展开出遮罩列、整行引用的照旧只给哈希，说明写清原因
This commit is contained in:
ThearchyHelios
+468
-20
@@ -8,11 +8,12 @@
|
||||
浏览器先核它的哈希，再从里面读出报告文档、口径卡、查询快照的 id。其余事件只给哈希：工具返回、节点产出的
|
||||
预览里可能有遮罩列的原值和提示词。
|
||||
- 工件：每份报告文档（各报告节点最后一次自查记下的 doc_artifact）、它引用到的口径卡和查询快照，按原字节
|
||||
base64 嵌入，浏览器重算 sha256 和 id 比；带遮罩的快照不嵌内容、只给 id。口径卡上记着每个输入的值和代入式，
|
||||
有输入取自遮罩列（cell() 的格、Agent 字段、按位置取的 vars.res.rows[0][2]）的整张卡同样只给 id，认法和证据
|
||||
面板、推演同一个（recompute.input_origin）；服务端文件被改过的卡照样按卡上记的输入认，解析不出一张卡的整张只给
|
||||
id。只给 id 的工件，服务端的文件和 id 对不上的另标 mismatch、解析不出来的另标 unparsable，核对页照实说（对不上的
|
||||
判不通过）。没被报告引用的工件不嵌。
|
||||
base64 嵌入，浏览器重算 sha256 和 id 比。数据源设了遮罩的查询快照，能证明这条查询没碰遮罩列的（_untouched）照常
|
||||
嵌入，别的不嵌内容、只给 id。口径卡上记着每个输入的值和代入式，有输入取自遮罩列（cell() 的格、Agent 字段、按位置
|
||||
取的 vars.res.rows[0][2]，SQL 里改了名字的遮罩列也算，SQL 认不出结果里哪一列取自哪一列的整份快照的列都算）的整张卡
|
||||
同样只给 id，认法和证据面板、推演同一个（recompute.input_origin）；服务端文件被改过的卡照样按卡上记的输入认，解析
|
||||
不出一张卡的整张只给 id。只给 id 的工件，服务端的文件和 id 对不上的另标 mismatch、解析不出来的另标 unparsable，
|
||||
核对页照实说（对不上的判不通过）。没被报告引用的工件不嵌。
|
||||
- 封存之后追加的事件（按需裁判、审批质询这类批注）单列：类型、序号、内容摘要，不进清单。
|
||||
|
||||
它只证明「内部一致」：数字、文档、运行记录彼此对得上，而且没被改过。真伪要拿页面上的封存哈希和系统里这次
|
||||
@@ -26,6 +27,7 @@ import base64
|
||||
import hashlib
|
||||
import html
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||||
from datetime import datetime
|
||||
@@ -42,6 +44,8 @@ from app.engine.recompute import input_origin
|
||||
from app.engine.schema import NodeType
|
||||
from app.engine.toolcalls import QUERY_PREFIX
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SCHEMA = "agentlab.bundle/1"
|
||||
|
||||
#: 不论哪个节点都带规范字符串的事件：运行开始（实际生效的运行参数）、报告自查（报告文档的 id）、
|
||||
@@ -71,14 +75,19 @@ Reader = Callable[[str], "bytes | None"]
|
||||
|
||||
|
||||
class _Store:
|
||||
"""这一次构建里读过的工件：原字节、解析出来的内容、查询快照带不带遮罩。每份只读一次盘。"""
|
||||
"""这一次构建里读过的工件：原字节、解析出来的内容、查询快照带不带遮罩。每份只读一次盘。events 是封存范围内的
|
||||
事件（认这次运行里查到过哪些遮罩值，seen）。"""
|
||||
|
||||
def __init__(self, read: Reader, source_masks: Mapping[str, Any]) -> None:
|
||||
def __init__(self, read: Reader, source_masks: Mapping[str, Any],
|
||||
events: Sequence[tuple[int, str, str | None, Any]] = ()) -> None:
|
||||
self._read = read
|
||||
self._source_masks = source_masks
|
||||
self._events = events
|
||||
self._raw: dict[str, bytes | None] = {}
|
||||
self._json: dict[str, Any] = {}
|
||||
self._masked: dict[str, bool] = {}
|
||||
self._hidden: dict[str, set[str]] = {}
|
||||
self._seen: Any = None
|
||||
|
||||
def raw(self, artifact: str) -> bytes | None:
|
||||
if artifact not in self._raw:
|
||||
@@ -104,21 +113,52 @@ class _Store:
|
||||
return raw is not None and content_hash(bytes(raw)) == artifact
|
||||
|
||||
def masked(self, artifact: str) -> bool:
|
||||
"""这份查询快照算不算带遮罩：查询当时记下了遮罩列（mask_columns），或者它的数据源现在设了遮罩
|
||||
（合并查询看它合并的那几个数据源）。比证据面板粗：面板只遮那几列，这里整份不随文件附带——文件会
|
||||
离开系统，遮罩只减少暴露，宁可多挡。读不出来的快照不知道带没带遮罩，按带遮罩算。"""
|
||||
"""这份查询快照算不算带遮罩（整份不随文件附带、只给 id）：查询当时记下了遮罩列（mask_columns），或者它的数据源
|
||||
现在设了遮罩（合并查询看它合并的那几个数据源），而且证明不了这条查询没碰遮罩列（_untouched）。比证据面板粗：
|
||||
面板只遮那几列，这里要么原字节整份附带、要么只给 id——拿掉几格哈希就对不上了；文件会离开系统，遮罩只减少暴露，
|
||||
宁可多挡。读不出来的快照不知道带没带遮罩，按带遮罩算；服务端的文件和 id 对不上的，里面的 SQL、列名都不可信，
|
||||
也按带遮罩算。"""
|
||||
if artifact not in self._masked:
|
||||
self._masked[artifact] = _snapshot_masked(self.content(artifact), self._source_masks)
|
||||
self._masked[artifact] = _snapshot_masked(self.content(artifact), self._source_masks,
|
||||
intact=self.intact(artifact), seen=self.seen)
|
||||
return self._masked[artifact]
|
||||
|
||||
def hidden(self, artifact: str) -> set[str]:
|
||||
"""一份读得出来的查询快照里值取自遮罩列的列（小写）：查询当时记下的、数据源现在设的遮罩列，加上 SQL 里给它们
|
||||
改了名字的；SQL 认不出结果里哪一列取自哪一列的（解析不了的也算），整份的列（_derived_columns）。"""
|
||||
if artifact not in self._hidden:
|
||||
content = self.content(artifact)
|
||||
base = _snapshot_hidden(content, self._source_masks) if isinstance(content, dict) else set()
|
||||
self._hidden[artifact] = _derived_columns(content, base) if base else set()
|
||||
return self._hidden[artifact]
|
||||
|
||||
def seen(self) -> Any:
|
||||
"""这次运行里查到过的遮罩值（inquiry._Secrets，只用按值认得出的写法：够长的数、两个字以上的文字）：封存范围内
|
||||
每份查询快照里值取自遮罩列的那几列（hidden）的格子。模型拿在遮罩查询里看到的号码去查别的表时，SQL 里没写遮罩列，
|
||||
字面量却就是号码（_untouched）。认不出哪一列取自哪一列的快照（整行摊开的画像查询）整份的格子都算：同一次运行里
|
||||
别的查询的字面量碰上其中哪个值，也只给哈希（宁可多挡）。这次运行里没查到过的（提问人写进问题的、别的运行里看到的）
|
||||
认不出。只收一次。"""
|
||||
if self._seen is None:
|
||||
from app.engine.inquiry import _Secrets # inquiry 导入了 engine.nodes 里的模块：用到时再导入
|
||||
|
||||
seen = _Secrets()
|
||||
for artifact in _run_queries(self._events, self):
|
||||
content = self.content(artifact)
|
||||
if isinstance(content, dict) and (hidden := self.hidden(artifact)):
|
||||
seen.add(content, hidden)
|
||||
self._seen = seen
|
||||
return self._seen
|
||||
|
||||
def card_masked(self, artifact: str, events: Sequence[tuple[int, str, str | None, Any]]) -> bool:
|
||||
"""这张口径卡算不算带遮罩：卡上有一个输入取自遮罩列，或者遮没遮核对不了。卡上记着每个输入的值（cell() 取的格、
|
||||
Agent 字段、按位置取的 vars.res.rows[0][2]）和代入式，整张卡不随文件附带——没法只拿掉几个值，拿掉了哈希就对不上。
|
||||
|
||||
认哪个输入取自哪份查询结果的哪几列，和证据面板、推演同一个认法（recompute.input_origin）；按位置取的一格顺着
|
||||
取值链从写变量的节点在口径卡算这一次之前的产出起走。核对不了的宁可多挡：快照读不出来、取值链上的产出读不出来、
|
||||
那份结果没存快照只带着数据源名、定位里没写列。认得出、可结果既没带快照的 id 也没带数据源名的（数据整形改过形状）
|
||||
认不出出处，不算。
|
||||
取值链从写变量的节点在口径卡算这一次之前的产出起走。那几列算不算遮罩列，除了列名对上的，SQL 里给遮罩列起了别名、
|
||||
套了函数、经子查询或 WITH 转了一道的也算（hidden，同质询、双盲复算的认法）；SQL 认不出结果里哪一列取自哪一列的
|
||||
（解析不了、列名表、整行引用这类，快照本身因此只给哈希），整份快照的列都算，两条路的判断一致。核对不了的宁可多挡：
|
||||
快照读不出来、取值链上的产出读不出来、那份结果没存快照只带着数据源名、定位里没写列。认得出、可结果既没带快照的 id
|
||||
也没带数据源名的（数据整形改过形状）认不出出处，不算。
|
||||
|
||||
卡本身按解析出来的内容认，不看哈希（同 content）：服务端的文件被改过，嵌进页面的照样是这份原字节，卡上记着的号码
|
||||
也照样在里面。读得出字节、解析不出一张卡的（截断的 JSON、不是对象）认不出有哪些输入，整张按带遮罩算，和读不出来的
|
||||
@@ -153,18 +193,20 @@ class _Store:
|
||||
return True
|
||||
if not origin.columns:
|
||||
return self.masked(origin.artifact)
|
||||
hidden = _snapshot_hidden(content, self._source_masks)
|
||||
hidden = self.hidden(origin.artifact)
|
||||
return any(c.lower() in hidden for c in origin.columns)
|
||||
|
||||
|
||||
def _snapshot_masked(content: Any, source_masks: Mapping[str, Any]) -> bool:
|
||||
def _snapshot_masked(content: Any, source_masks: Mapping[str, Any], *, intact: bool,
|
||||
seen: Callable[[], Any] | None = None) -> 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 [])
|
||||
if not (isinstance(recorded, list) and any(isinstance(c, str) and c.strip() for c in recorded)) \
|
||||
and not any(source_masks.get(str(name or "")) for name in sources or []):
|
||||
return False
|
||||
return not (intact and _untouched(content, _snapshot_hidden(content, source_masks), seen))
|
||||
|
||||
|
||||
def _snapshot_hidden(content: dict[str, Any], source_masks: Mapping[str, Any]) -> set[str]:
|
||||
@@ -178,6 +220,400 @@ def _snapshot_hidden(content: dict[str, Any], source_masks: Mapping[str, Any]) -
|
||||
return hidden
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 设了遮罩的数据源上的查询：能不能证明没碰遮罩列
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
#: 数据源查询工具（tools/datasource.py）落的查询快照里会有的键。有别的键的（比如核查 agent 在查询时就把遮罩列换成
|
||||
#: 「已遮罩」、另记下了遮了哪些列）不知道它记着什么，照现有规则只给 id
|
||||
_QUERY_KEYS = frozenset({"columns", "rows", "row_count", "truncated", "elapsed_ms", "sql", "column_types", "source",
|
||||
"checks", "schema_artifact", "mask_columns", "data_version"})
|
||||
#: 导出时只知道数据源设的遮罩、不知道是哪种库：SQL 按每种支持的库的读法各解析一遍（None 是 sqlglot 的通用读法），
|
||||
#: 至少一种读得通，读得通的每一种都要过关
|
||||
_READINGS = (None, "sqlite", "mysql", "postgres", "oracle")
|
||||
#: 语法树和数据库实际执行的对不上的写法：MySQL、MariaDB 会执行的注释（/*! … */、/*M! … */，解析器当注释扔掉），
|
||||
#: Postgres 按 Unicode 转义写的名字（U&"…"，原文里看不出是哪一列）
|
||||
_OPAQUE_SQL = re.compile(r'/\*m?!|u&"', re.I)
|
||||
#: 不点列名也读得出表里内容的写法。SQL 原文里出现就按碰了遮罩列算，写在字符串里的也算（宁可多挡）。函数要跟着左括号
|
||||
#: 才算（列名碰巧叫 crosstab 的不算），表、视图、schema 按名字认
|
||||
_INDIRECT_READS = re.compile(r"(?<![0-9A-Za-z_$])(?:(?:" + "|".join((
|
||||
# 表名、SQL 写在字符串里交给函数执行或整表导出：Postgres 的 XML 一族、ts_stat、tablefunc 的 crosstab / connectby、
|
||||
# dblink，Oracle 的 XMLTYPE、XQuery（XMLQUERY、XMLTABLE）
|
||||
r"(?:query|table|cursor|schema|database)_to_xml\w*", r"ts_stat", r"crosstab\d*", r"connectby", r"dblink\w*",
|
||||
r"xmltype", r"xmlquery", r"xmltable",
|
||||
# 按页、按文件、按日志读出的原始数据：Postgres 的 pageinspect（按索引名读 B 树页的 bt_page_items 也算）、服务器文件、
|
||||
# 逻辑解码、pg_walinspect 读出的 WAL 块，MySQL 的 LOAD_FILE，Oracle 的 BFILENAME，SQLite 的 readfile
|
||||
r"get_raw_page", r"bt_page_items", r"pg_read_(?:binary_)?file", r"pg_logical_slot_\w+", r"pg_get_wal_block_info",
|
||||
r"load_file", r"bfilename", r"readfile",
|
||||
# 别的会话执行过的 SQL 原文：Postgres 的 pg_stat_activity 背后的函数和按会话取的那一个，MySQL 企业版的审计日志
|
||||
r"pg_stat_get_(?:backend_)?activity", r"audit_log_read",
|
||||
)) + r")[\"`]?\s*\(|(?:" + "|".join((
|
||||
# Oracle 的 DBMS_XMLGEN 这一族（包名后面跟着点），XQuery 里的 ora:view 和 oradb: 表
|
||||
r"dbms_xml\w*", r"ora:view", r"oradb",
|
||||
# 统计信息里的采样值、全文索引里的词：Postgres 的 pg_stats、pg_statistic 一族，SQLite 的 sqlite_stat*，MySQL 的
|
||||
# column_statistics、innodb_ft_*，MariaDB 的 column_stats，Oracle 的直方图、列统计和记着最小最大值的 *_tab_columns，
|
||||
# 以及它们的底表 histgrm$、hist_head$
|
||||
r"pg_stat(?:s|istic)\w*", r"sqlite_stat\d", r"column_statistics", r"innodb_ft_\w+", r"column_stats",
|
||||
r"(?:all|dba|user|cdb)_(?:\w+_)?(?:histograms|col_statistics|tab_cols|tab_columns)", r"(?:histgrm|hist_head)\$",
|
||||
# 别的会话执行过的 SQL 原文（写入语句里就是值）：Postgres 的 pg_stat_activity、pg_stat_statements，本连接预备过的
|
||||
# 语句（连接池里的连接留着前几次查询的原文）；MySQL 的 general_log、slow_log，sys 库里建在 performance_schema 上的
|
||||
# 会话、锁等待视图（带 x$ 的是不排版的那一份；只认这几张，Oracle 常写的 sys.dual 不算）；Oracle 的 V$SQL 一族、
|
||||
# 打开着的游标、审计记录
|
||||
r"pg_stat_activity", r"pg_stat_statements", r"pg_prepared_statements", r"general_log", r"slow_log",
|
||||
r"sys[\"`]?\s*\.\s*[\"`]?(?:x\$)?(?:session|processlist|innodb_lock_waits|schema_table_lock_waits)",
|
||||
r"g?v\$sql\w*", r"(?:dba|cdb)_hist_sql\w*", r"g?v\$open_cursor",
|
||||
r"(?:dba|cdb|user)_(?:\w+_)?audit_trail", r"unified_audit_trail", r"aud\$\w*", r"fga_log\$",
|
||||
# SQLite 按页读出的原始数据
|
||||
r"sqlite_dbpage",
|
||||
# 系统 schema：哪几张表采样了列值、记着别的会话的 SQL，一张张数靠不住，整个不认
|
||||
r"pg_catalog", r"information_schema", r"performance_schema",
|
||||
)) + r")(?![0-9A-Za-z_$]))", re.I)
|
||||
#: Postgres 的函数记法（m.to_jsonb 就是 to_jsonb(m)）下能把一整行交出来的内置函数（_whole_row）：只收一个参数、收得下
|
||||
#: 整行（record、anyelement、"any" 这类），交回的值里带着整行。汇总函数也能这么写：any_value、*_agg 交回整行，max、min
|
||||
#: 交回最大、最小的那一整行。照 PG 18 的 pg_proc 逐个对过：只交回个数、大小、类型、哈希的不算，这么写会报错的（伪类型的
|
||||
#: 输出函数、要成对参数的 json_build_object、要 WITHIN GROUP 的 mode）也不算；扩展里的只认 hstore，自己建的函数这里不知道
|
||||
_ROW_FUNCTIONS = frozenset({"to_json", "to_jsonb", "row_to_json", "hstore", "quote_literal", "quote_nullable",
|
||||
"concat", "json_build_array", "jsonb_build_array", "record_out", "record_send",
|
||||
"array_agg", "json_agg", "jsonb_agg", "json_agg_strict", "jsonb_agg_strict", "any_value",
|
||||
"max", "min"})
|
||||
|
||||
|
||||
def _untouched(snapshot: dict[str, Any], hidden: set[str], seen: Callable[[], Any] | None = None) -> bool:
|
||||
"""数据源设了遮罩的一份查询快照，能不能证明这条查询没碰遮罩列（快照里不会有遮罩值）。全部满足才算，能证明的照常
|
||||
随文件附带：
|
||||
- 认得这份快照：数据源查询工具落的那种（不是合并查询，没有不认识的键），SQL、列名、行都在；
|
||||
- 结果的列名对不上遮罩列（按数据库交回的列名，不论大小写）；
|
||||
- SQL 原文里没写到遮罩列，条件里、COUNT 里也不行：SQL 随快照一起附带，WHERE phone = '…' 的字面量就是遮罩值；
|
||||
FROM 里的表函数吃进遮罩列时（unnest(ARRAY[phone])），它交出的列取自哪一列也认不出。这一条比「按 SQL 认不出结果列
|
||||
取自遮罩列」更严，有意的：COUNT(DISTINCT phone) 这类结果里没有号码的也只给哈希。连字符串、注释里写到的也算（宁可
|
||||
多挡）；
|
||||
- 没有语法树认不准的写法（_OPAQUE_SQL），也没有不点列名也读得出表里内容的写法（_INDIRECT_READS）；
|
||||
- SQL 解析得了；字面量里没写着这次运行里查到过的遮罩值（seen，_quotes）；每种读得通的读法下，没有认不出结果里哪一列
|
||||
取自哪一列的写法（_stray_columns：列名表、整行引用、可能展开出遮罩列的 *），按 SQL 认遮罩列的那一套
|
||||
（inquiry._derived_hidden，双盲复算、审批质询用的同一个）也认不出哪个结果列取自遮罩列。SQL 里没写到遮罩列时它本来
|
||||
就认不出，这一条是同一套认法的第二道关，那一套以后认得更多时跟着收紧。
|
||||
判断过程出错按碰了算。和证据面板一样只按列名认遮罩：视图、别的表里换了名字存着的遮罩列，自己建的函数，都认不出
|
||||
（遮罩只减少暴露，不是安全边界）。hidden 是要遮的列（小写，_snapshot_hidden）；seen 交回这次运行里查到过的遮罩值
|
||||
（_Store.seen）。"""
|
||||
names = {name.strip() for name in hidden if name.strip()}
|
||||
sql, columns = snapshot.get("sql"), snapshot.get("columns")
|
||||
if not names or snapshot.get("source") == MERGE_SOURCE or not set(snapshot) <= _QUERY_KEYS \
|
||||
or not isinstance(sql, str) or not sql.strip() or not isinstance(snapshot.get("rows"), list) \
|
||||
or not isinstance(columns, list) or not all(isinstance(c, str) for c in columns):
|
||||
return False
|
||||
result = {c.strip().lower() for c in columns}
|
||||
if result & names or _writes_any(sql, names) or _OPAQUE_SQL.search(sql) or _INDIRECT_READS.search(sql):
|
||||
return False
|
||||
try:
|
||||
from app.data.sql_semantics import _parse, _read
|
||||
from app.engine.inquiry import _derived_hidden # inquiry 导入了 engine.nodes 里的模块：用到时再导入
|
||||
|
||||
readings = [(kind, tree) for kind in _READINGS for tree in [_parse(sql, _read(kind))[0]] if tree is not None]
|
||||
if not readings or (seen is not None and _quotes([tree for _, tree in readings], seen())):
|
||||
return False
|
||||
return not any(
|
||||
_stray_columns(tree) or result & {c.lower() for c in _derived_hidden(dict(snapshot), set(names), kind)[0]}
|
||||
for kind, tree in readings)
|
||||
except Exception: # noqa: BLE001 - 解析器、认遮罩列的那一套出了意外：按碰了遮罩列算（宁可多挡）
|
||||
logger.warning("可验证报告：判断查询快照有没有碰遮罩列时出错，按碰了处理", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
def _derived_columns(snapshot: dict[str, Any], base: set[str]) -> set[str]:
|
||||
"""一份快照里值取自遮罩列的结果列（小写）：遮罩列本身（base），加上 SQL 里给它们起别名、套函数、经子查询或 WITH
|
||||
转一道得来的（inquiry._derived_hidden，质询、双盲复算同一个认法）。SQL 按每种读得通的读法各认一遍、合起来。认不出
|
||||
结果里哪一列取自哪一列的，整份的列都算：每种读法都解析不了、没有 SQL 的，语法树认不准、不点列名也读得出表里内容的
|
||||
写法（_OPAQUE_SQL、_INDIRECT_READS），以及哪一种读法下有列名表、整行引用、不照原名展开的 *、吃进遮罩列的表函数的
|
||||
（_renamed）——快照本身因此只给哈希，口径卡上取自它的值也一样。解析不了的不退回那一套：它解析不了时只看 SQL 里写没
|
||||
写着遮罩列，没写就只认遮罩列本身，卡上取自整行的值认不出。合并查询照那一套的办法不看 SQL（它记着自己的来历）。
|
||||
出错时整份的列都算（宁可多挡）。"""
|
||||
columns = snapshot.get("columns")
|
||||
every = set(base) | ({str(c).lower() for c in columns} if isinstance(columns, list) else set())
|
||||
try:
|
||||
from app.data.sql_semantics import _parse, _read
|
||||
from app.engine.inquiry import _derived_hidden
|
||||
|
||||
if snapshot.get("source") == MERGE_SOURCE:
|
||||
return set(base) | {c.lower() for c in _derived_hidden(dict(snapshot), set(base), None)[0]}
|
||||
sql = snapshot.get("sql")
|
||||
if not isinstance(sql, str) or _OPAQUE_SQL.search(sql) or _INDIRECT_READS.search(sql):
|
||||
return every
|
||||
readings = [(kind, tree) for kind in _READINGS for tree in [_parse(sql, _read(kind))[0]] if tree is not None]
|
||||
if not readings:
|
||||
return every
|
||||
out = set(base)
|
||||
for kind, tree in readings:
|
||||
found, names = _derived_hidden(dict(snapshot), set(base), kind)
|
||||
if _renamed(tree, {n.lower() for n in names}):
|
||||
return every
|
||||
out |= {c.lower() for c in found}
|
||||
return out
|
||||
except Exception: # noqa: BLE001 - 认遮罩列的那一套出了意外：整份的列都按遮罩算
|
||||
logger.warning("可验证报告：按 SQL 认改了名字的遮罩列时出错，整份快照的列都按遮罩处理", exc_info=True)
|
||||
return every
|
||||
|
||||
|
||||
def _writes_any(sql: str, names: Iterable[str]) -> bool:
|
||||
"""SQL 原文里有没有写到这些名字：不论大小写，前后不是字母、数字、下划线（引号、括号、点都算隔开）。"""
|
||||
return any(re.search(rf"(?<![0-9A-Za-z_]){re.escape(n)}(?![0-9A-Za-z_])", sql, re.I) for n in names if n)
|
||||
|
||||
|
||||
def _quotes(trees: Iterable[Any], secrets: Any) -> bool:
|
||||
"""SQL 的字面量（'…'、数、$$…$$、E'…'、N'…'、U&'…'，MySQL 读法下的 "…"）里有没有写着按值认得出的遮罩值（secrets
|
||||
是 inquiry._Secrets）。只看字面量：守卫已经去掉了注释；表名、列名碰巧和哪个值一样不算（整行摊开的画像查询把列名
|
||||
当值交了出来，那些列名会出现在别的 SQL 里）。trees 是各种读法解析出来的语法树，字面量合起来看。"""
|
||||
from sqlglot import exp
|
||||
|
||||
kinds = (exp.Literal, exp.RawString, exp.National, exp.UnicodeString, exp.ByteString)
|
||||
texts = {node.this for tree in trees for node in tree.walk()
|
||||
if isinstance(node, kinds) and isinstance(node.this, str)}
|
||||
for text in texts:
|
||||
pattern = secrets.pattern(text)
|
||||
if pattern is not None and pattern.search(text):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _stray_columns(tree: Any) -> bool:
|
||||
"""语法树里有没有认不出结果里哪一列取自哪一列的写法。有的话，就算 SQL 里没写到遮罩列、结果的列名也对不上，结果里
|
||||
照样可能是遮罩列的值：
|
||||
- 列名表、整行引用（_listed_or_whole）；
|
||||
- 可能展开出遮罩列的 *：展开了数据源里的表（隔着几层派生来源也算，_star_reaches），又不是只按名字往外传的那几种
|
||||
（_contained）。COUNT(*) 只数个数，不算。"""
|
||||
if _listed_or_whole(tree):
|
||||
return True
|
||||
memo: dict[int, bool] = {}
|
||||
return any(_star_reaches(node, memo) and not _contained(node) for node in _stars(tree))
|
||||
|
||||
|
||||
def _renamed(tree: Any, names: set[str]) -> bool:
|
||||
"""结果里可能有遮罩列的值、列名却和 SQL 里认得出的名字对不上的写法（口径卡按列名认遮罩列，这几种认不出）：
|
||||
- 列名表、整行引用（_listed_or_whole）；
|
||||
- 不照原名展开的 *：UNION 这类里按位置对上第一支列名的，套在表达式、标量子查询里装进一列的。最外层直接写的 *、
|
||||
只按名字往外传的（_contained）照原名展开，遮罩列在结果里还叫原来的名字；
|
||||
- FROM、JOIN 里的表函数（UNNEST、ROWS FROM 这类，_table_function）吃进了遮罩列（names：SQL 里指代遮罩列的名字，
|
||||
小写）：它交出的列叫什么、装着什么，SQL 里看不出。"""
|
||||
from sqlglot import exp
|
||||
|
||||
if _listed_or_whole(tree) or any(not (_contained(node) or _outermost(node)) for node in _stars(tree)):
|
||||
return True
|
||||
return any(_table_function(source) and names & {c.name.lower() for c in source.find_all(exp.Column)}
|
||||
for select in tree.find_all(exp.Select) for source in _sources(select))
|
||||
|
||||
|
||||
def _listed_or_whole(tree: Any) -> bool:
|
||||
"""有没有列名表（members AS m(a, b, c)、(…) AS t(a, b)、WITH t(a, b)：按位置给列改了名），或者整行引用
|
||||
（Postgres 的 SELECT m、to_json(m)，_whole_row：一整行装进一列）。"""
|
||||
from sqlglot import exp
|
||||
|
||||
return any(alias.columns for alias in tree.find_all(exp.TableAlias)) \
|
||||
or any(_whole_row(column) for column in tree.find_all(exp.Column))
|
||||
|
||||
|
||||
def _sources(select: Any) -> list[Any]:
|
||||
"""一条 SELECT 的 FROM、JOIN 里的来源节点。括号括起来的一组连接（(members m JOIN orders o ON …)）摊开，里面每个
|
||||
来源也算：组里的表函数看得见组里前面的来源，外层也看得见组里的名字（组写了别名时 Postgres 外层只认别名，这里宁可
|
||||
多认）。"""
|
||||
from_ = select.args.get("from_")
|
||||
pending = ([from_.this] if from_ is not None and from_.this is not None else []) \
|
||||
+ [join.this for join in select.args.get("joins") or [] if join.this is not None]
|
||||
out: list[Any] = []
|
||||
while pending:
|
||||
source = pending.pop(0)
|
||||
out.append(source)
|
||||
head = _join_group(source)
|
||||
if head is not None:
|
||||
if head is not source:
|
||||
out.append(head)
|
||||
pending[:0] = [join.this for join in head.args["joins"] if join.this is not None]
|
||||
return out
|
||||
|
||||
|
||||
def _join_group(source: Any) -> Any:
|
||||
"""括号括起来的一组连接里打头的那个来源（语法树里是一层括号包着一张带 joins 的表）；别的来源是 None。"""
|
||||
from sqlglot import exp
|
||||
|
||||
node = source
|
||||
while isinstance(node, exp.Subquery):
|
||||
node = node.this
|
||||
return node if isinstance(node, exp.Table) and node.args.get("joins") else None
|
||||
|
||||
|
||||
def _derived(source: Any) -> Any:
|
||||
"""FROM、JOIN 里的一个来源是派生来源（子查询、LATERAL、指向 WITH 里一段的名字）时，它的那块查询；数据源里的表、表函数、
|
||||
VALUES 这类是 None（有哪些列、装着什么，SQL 里看不出）。"""
|
||||
from sqlglot import exp
|
||||
|
||||
from app.data.sql_semantics import _visible_ctes
|
||||
|
||||
node = source.this if isinstance(source, exp.Lateral) else source
|
||||
if isinstance(node, exp.Table):
|
||||
if node.args.get("db") or node.args.get("catalog") or not isinstance(node.this, exp.Identifier):
|
||||
return None
|
||||
cte = _visible_ctes(node).get(node.name.lower())
|
||||
if cte is None:
|
||||
return None
|
||||
node = cte.this
|
||||
while isinstance(node, exp.Subquery):
|
||||
node = node.this
|
||||
return node if isinstance(node, (exp.Select, exp.SetOperation)) else None
|
||||
|
||||
|
||||
def _table_function(source: Any) -> bool:
|
||||
"""FROM、JOIN 里的一个来源是不是表函数这类（函数调用、UNNEST、ROWS FROM、VALUES，写没写 LATERAL 都一样）：既不是
|
||||
数据源里的表，也不是派生来源。"""
|
||||
from sqlglot import exp
|
||||
|
||||
if (isinstance(source, exp.Table) and isinstance(source.this, exp.Identifier)) or _join_group(source) is not None:
|
||||
return False
|
||||
return _derived(source) is None
|
||||
|
||||
|
||||
def _leaves(block: Any) -> list[Any]:
|
||||
"""一块查询的每一支，按先后：UNION 这类摊开，套在外面的括号去掉。"""
|
||||
from sqlglot import exp
|
||||
|
||||
if isinstance(block, exp.Subquery):
|
||||
return _leaves(block.this)
|
||||
if isinstance(block, exp.SetOperation):
|
||||
return _leaves(block.this) + _leaves(block.expression)
|
||||
return [block]
|
||||
|
||||
|
||||
def _whole_row(column: Any) -> bool:
|
||||
"""一个名字是不是整行引用（Postgres 的 SELECT m、to_json(m)，函数记法的 m.to_jsonb：一整行装进一列）。
|
||||
|
||||
没写表名的，Postgres 先按列认：所在 SELECT 的派生来源明写出了叫这个名字的列，就是那一列；不然对上它看得见的某个来源
|
||||
的名字（别名、表名、WITH 里的一段）就算。名字写在 FROM、JOIN 里的表函数的参数里时，只有排在这个表函数前面的派生
|
||||
来源算数：它看不见后面的来源，(SELECT 1 AS m) s 排在后面，m 还是整行（_holder）。写了表名的，列名是 _ROW_FUNCTIONS
|
||||
里的、表名对上看得见的来源才算，同样先按列认：那个来源是明写出了这一列的派生来源，t.max 就是那一列，不是 max(t)。
|
||||
看得见的是所在 SELECT 和外层 SELECT 的 FROM、JOIN 里的来源。FROM、JOIN 里的表函数（UNNEST、ROWS FROM 这类）和
|
||||
LATERAL 看得见同一层的来源：Postgres 的函数不写 LATERAL 也看得见（逗号、CROSS JOIN、JOIN … ON TRUE 都一样），整行
|
||||
交给 jsonb_each_text 这类就一列列竖着摊开了；括号括起来的一组连接按同一层算（_sources 摊开）。FROM、JOIN 里的子查询
|
||||
不写 LATERAL 看不见，WITH 里的一段也看不见。数据源里的表有哪些列这里不知道：碰巧和表、别名同名的列，碰巧叫 max、min
|
||||
这类名字的列也算（宁可多挡）。"""
|
||||
from sqlglot import exp
|
||||
|
||||
select = column.find_ancestor(exp.Select)
|
||||
if isinstance(column.this, exp.Star) or not column.name or select is None:
|
||||
return False
|
||||
if column.table:
|
||||
if column.name.lower() not in _ROW_FUNCTIONS:
|
||||
return False
|
||||
name, field = column.table.lower(), column.name.lower()
|
||||
else:
|
||||
name, field = column.name.lower(), None
|
||||
sources = _sources(select)
|
||||
if any(_names_column(_derived(source), name) for source in sources[:_holder(column, select, sources)]):
|
||||
return False
|
||||
child, node = column, column.parent
|
||||
while node is not None:
|
||||
# FROM、JOIN 里真的子查询（括号里是一条查询，不是一组连接）看不见同一层，WITH 里的一段也看不见
|
||||
subquery = isinstance(child.this, exp.Subquery) and _derived(child.this) is not None
|
||||
sees = child.arg_key != "with_" and not (child.arg_key in ("from_", "joins") and subquery)
|
||||
found = [source for source in _sources(node) if (source.alias_or_name or "").lower() == name] \
|
||||
if isinstance(node, exp.Select) and sees else []
|
||||
if found:
|
||||
return field is None or not any(_names_column(_derived(source), field) for source in found)
|
||||
child, node = node, node.parent
|
||||
return False
|
||||
|
||||
|
||||
def _holder(column: Any, select: Any, sources: list[Any]) -> int | None:
|
||||
"""名字写在所在 SELECT 的哪一个表函数（_table_function）的参数里：交回它在 sources 里的位置；写在选出项、条件、
|
||||
ON 里的是 None（同一层的来源都看得见）。"""
|
||||
places = {id(source): place for place, source in enumerate(sources) if _table_function(source)}
|
||||
node = column.parent
|
||||
while node is not None and node is not select:
|
||||
if id(node) in places:
|
||||
return places[id(node)]
|
||||
node = node.parent
|
||||
return None
|
||||
|
||||
|
||||
def _names_column(block: Any, name: str) -> bool:
|
||||
"""一个派生来源（_derived 交回的那块查询，None 是数据源里的表这类）明写出了叫这个名字的列：第一支的选出项里起了
|
||||
这个别名、或者直接写着这一列（* 不算，展开出哪些列这里看不出）。"""
|
||||
from sqlglot import exp
|
||||
|
||||
first = _leaves(block)[0] if block is not None else None
|
||||
return isinstance(first, exp.Select) and any(
|
||||
isinstance(item, (exp.Alias, exp.Column)) and not isinstance(item.this, exp.Star)
|
||||
and item.alias_or_name.lower() == name for item in first.expressions)
|
||||
|
||||
|
||||
def _stars(tree: Any) -> list[Any]:
|
||||
"""语法树里的 *（t.* 取包着它的 Column），COUNT(*)、COUNT(t.*) 不算：只数个数。"""
|
||||
from sqlglot import exp
|
||||
|
||||
nodes = [star.parent if isinstance(star.parent, exp.Column) else star for star in tree.find_all(exp.Star)]
|
||||
return [node for node in nodes if not isinstance(node.parent, exp.Count)]
|
||||
|
||||
|
||||
def _star_reaches(node: Any, memo: dict[int, bool]) -> bool:
|
||||
"""这个 *（t.* 是 Column 包着的 *）展开了数据源里的表没有：t.* 看 FROM、JOIN 里叫 t 的那一个来源，单独的 * 看所在
|
||||
SELECT 的全部来源。数据源里的表、表函数、VALUES 这类来源都算展开了；派生来源看它每一支选出项里的 * 展开了没有
|
||||
（隔几层都一样）。t 对不上任何来源（外层的别名）、找不到所在的 SELECT，按展开了算。"""
|
||||
from sqlglot import exp
|
||||
|
||||
select = node.find_ancestor(exp.Select)
|
||||
if select is None:
|
||||
return True
|
||||
sources = _sources(select)
|
||||
if isinstance(node, exp.Column):
|
||||
sources = [s for s in sources if (s.alias_or_name or "").lower() == node.table.lower()]
|
||||
if not sources:
|
||||
return True
|
||||
for source in sources:
|
||||
block = _derived(source)
|
||||
if block is None:
|
||||
return True
|
||||
if id(block) not in memo:
|
||||
# 先记成展开了：递归的 WITH 绕回自己时按展开了算
|
||||
memo[id(block)] = True
|
||||
memo[id(block)] = any(not isinstance(leaf, exp.Select) or any(
|
||||
_star_reaches(star, memo) for item in leaf.expressions for star in _stars(item))
|
||||
for leaf in _leaves(block))
|
||||
if memo[id(block)]:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _contained(node: Any) -> bool:
|
||||
"""展开出来的列只按名字往外传的 *：它就是一条 SELECT 的选出项本身（不套在表达式里），这条 SELECT 不在 UNION 这类
|
||||
里，而且是 FROM / JOIN 里的派生来源、LATERAL、WITH 里的一段，或者 EXISTS / IN 里的子查询（只决定有没有、在不在）。
|
||||
派生来源的列外层只能按名字取（取到遮罩列，名字就写在 SQL 里），或者再用一个 * 展开（那个 * 照样要过这一关），
|
||||
或者整行交出去（整行引用另认，_whole_row）。最外层的 *（结果就是它展开的列）、UNION 里的、标量子查询里的都不算。"""
|
||||
from sqlglot import exp
|
||||
|
||||
select = node.parent
|
||||
if not isinstance(select, exp.Select) or node.arg_key != "expressions":
|
||||
return False
|
||||
outer = select
|
||||
while isinstance(outer.parent, exp.Subquery):
|
||||
outer = outer.parent
|
||||
parent = outer.parent
|
||||
if isinstance(parent, (exp.CTE, exp.Lateral, exp.Exists)):
|
||||
return True
|
||||
if isinstance(parent, (exp.From, exp.Join)):
|
||||
return outer is not select
|
||||
return isinstance(parent, exp.In) and outer.arg_key == "query"
|
||||
|
||||
|
||||
def _outermost(node: Any) -> bool:
|
||||
"""这个 * 是不是最外层那条 SELECT 的选出项本身（不套在表达式里，这条 SELECT 也不在 UNION 这类里）：展开的列照原来
|
||||
的名字落进结果。"""
|
||||
from sqlglot import exp
|
||||
|
||||
select = node.parent
|
||||
if not isinstance(select, exp.Select) or node.arg_key != "expressions":
|
||||
return False
|
||||
outer = select
|
||||
while isinstance(outer.parent, exp.Subquery):
|
||||
outer = outer.parent
|
||||
return outer.parent is None
|
||||
|
||||
|
||||
def _card_outputs(events: Sequence[tuple[int, str, str | None, Any]], card: str) -> dict[str, str]:
|
||||
"""算出这份指标集的那一次口径卡执行开始之前，每个节点最后一次的产出工件：卡上 vars.*、nodes.* 取到的就是它们
|
||||
（和证据接口 api/evidence._card_outputs 同一个认法）。认不出是哪一次执行时取封存范围内每个节点最后一次的产出。"""
|
||||
@@ -254,6 +690,18 @@ def _query_returned(data: Mapping[str, Any], store: _Store) -> tuple[str, str |
|
||||
return "unknown", None
|
||||
|
||||
|
||||
def _run_queries(events: Sequence[tuple[int, str, str | None, Any]], store: _Store) -> list[str]:
|
||||
"""封存范围内查到过的查询快照：_query_artifacts 之外，升级前的运行从工具快照、预览里认出来的（_query_returned）。"""
|
||||
out = _query_artifacts(events)
|
||||
for _, etype, _, data in events:
|
||||
if etype == EventType.TOOL_END and isinstance(data, dict) \
|
||||
and str(data.get("tool") or "").startswith(QUERY_PREFIX):
|
||||
artifact = _query_returned(data, store)[1]
|
||||
if artifact:
|
||||
out.append(artifact)
|
||||
return list(dict.fromkeys(out))
|
||||
|
||||
|
||||
def _masked_run(events: Sequence[tuple[int, str, str | None, Any]], store: _Store, cited: Iterable[str],
|
||||
source_masks: Mapping[str, Any]) -> bool:
|
||||
"""运行里有没有哪次查询的结果可能带着遮罩列。有的话，成果节点和运行结束事件的预览可能直接带出查询结果，只给哈希。
|
||||
@@ -468,7 +916,7 @@ def build_bundle(
|
||||
|
||||
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 {})
|
||||
store = _Store(read, source_masks or {}, sealed)
|
||||
|
||||
artifacts = _Artifacts(store, sealed)
|
||||
reports = _report_events(sealed)
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
* engine/expressions.py 的 cell_value / same_value；数字怎么显示是服务端渲染好的，这里不重新排版；
|
||||
* ⑥ 运行信息：运行结果、运行类别、出具档位拿封存范围内已核对的事件对照。工作流名、版本、起止时间、节点名称
|
||||
* 绑不进封存，页面上写明它们取自运行记录。
|
||||
* 每项给出通过 / 不通过 / 无法核对和原因。服务端导出时不会出现的状态（被引用的工件不在清单里、口径卡标着遮罩）
|
||||
* 每项给出通过 / 不通过 / 无法核对和原因。服务端导出时不会出现的状态（被引用的工件不在清单里、报告文档标着遮罩）
|
||||
* 一律算不通过：只可能是文件被改过。这些只证明内部一致，真伪要拿封存哈希去系统里对照，页面上写明。
|
||||
*
|
||||
* SHA-256 优先用浏览器的 crypto.subtle；拿不到（非安全上下文、某些浏览器打开本地文件时）用下面的纯 JS 实现，
|
||||
@@ -105,6 +105,9 @@
|
||||
artifactMismatch: '内容与工件 id（哈希）不一致',
|
||||
artifactUndecodable: '内容无法解码',
|
||||
artifactUnparsable: '哈希一致，但内容无法解析',
|
||||
/** 查询快照只给哈希：服务端证明不了这条查询没碰设了遮罩的列（没碰的照常随文件附带） */
|
||||
maskedQuery: '这条查询用到或可能用到设了遮罩的列，快照未随文件附带，只能核对到哈希',
|
||||
/** 口径卡只给哈希：卡上有输入取自遮罩列（卡上记着输入的值和代入式） */
|
||||
masked: '含遮罩列，未随文件附带，只能核对到哈希',
|
||||
maskedMismatch: '导出时服务端的这份文件与工件 id（哈希）不一致，可能被改过；按含遮罩列处理，未随文件附带',
|
||||
maskedUnparsable: '导出时服务端的这份文件无法解析，认不出是否含遮罩列，按含遮罩列处理，未随文件附带',
|
||||
@@ -487,7 +490,8 @@
|
||||
}
|
||||
}
|
||||
} else if (a.masked) {
|
||||
// 查询快照带遮罩列；口径卡上有输入取自遮罩列（卡上记着输入的值和代入式）。两种都只给 id，报告文档不会。
|
||||
// 查询快照：这条查询用到或可能用到设了遮罩的列；口径卡：卡上有输入取自遮罩列（卡上记着输入的值和代入式）。
|
||||
// 两种都只给 id，报告文档不会。
|
||||
// 内容不随文件附带，浏览器算不出它对不对得上：导出时服务端看到自己的文件和 id 对不上（mismatch）、解析不出来
|
||||
// （unparsable）的照实说，对不上的判不通过——不然被改过的文件看起来和正常的遮罩一样，篡改的证据就丢了
|
||||
if (a.kind !== 'query' && a.kind !== 'metric_set') {
|
||||
@@ -497,7 +501,7 @@
|
||||
item.result = FAIL
|
||||
item.reason = T.maskedMismatch
|
||||
} else {
|
||||
item.reason = a.unparsable ? T.maskedUnparsable : T.masked
|
||||
item.reason = a.unparsable ? T.maskedUnparsable : a.kind === 'query' ? T.maskedQuery : T.masked
|
||||
}
|
||||
} else if (a.missing) {
|
||||
item.reason = T.artifactMissing
|
||||
@@ -980,7 +984,7 @@
|
||||
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])
|
||||
|
||||
@@ -19,6 +19,8 @@
|
||||
- missing-doc / missing-card：导出时服务端已没有报告文档 / 口径卡的文件
|
||||
- masked-card：口径卡上一个指标的输入取自会员消费的号码列（设了遮罩），卡上记着号码原值：整张卡只给 id，引用它的
|
||||
数字无法核对
|
||||
- masked-source：会员库设了号码遮罩（查询当时记下了，数据源现在也设着）。报告引用会员按地区汇总的查询（没碰遮罩列：
|
||||
快照随文件附带，数字核对得过）和选出了号码的那份会员消费（只给哈希）
|
||||
- nan：运行开始事件和门店流水快照里有 json.dumps 写出的裸 NaN（快照里那一列报告没有引用）
|
||||
- stray：报告自查记下的文档是另一次运行写的
|
||||
"""
|
||||
@@ -39,8 +41,8 @@ RUN = {"id": RUN_ID, "workflow_id": "wf-sample-weekly", "workflow_name": "门店
|
||||
"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", "会员消费"),
|
||||
("card", "metrics", "周报口径"), ("write", "report", "写周报"), ("monthly", "report", "写月报"),
|
||||
("out", "output", "成果")]]}
|
||||
("regions", "tool", "会员地区汇总"), ("card", "metrics", "周报口径"), ("write", "report", "写周报"),
|
||||
("monthly", "report", "写月报"), ("out", "output", "成果")]]}
|
||||
EXPORTED_AT = "2026-10-05T08:00:00+00:00"
|
||||
|
||||
STORES = [["青石店", 1280.5], ["白沙店", "960.00"], ["临川店", 732], ["望江店", 1105.25], ["枫桥店", 418],
|
||||
@@ -48,6 +50,11 @@ STORES = [["青石店", 1280.5], ["白沙店", "960.00"], ["临川店", 732], ["
|
||||
CRM = {"columns": ["name", "phone", "spend"], "column_types": {"name": "text", "phone": "text", "spend": "number"},
|
||||
"rows": [["林一", "000-0001", 320.0], ["周二", "000-0002", 188.5]], "source": "crm",
|
||||
"sql": "SELECT name, phone, spend FROM members ORDER BY id", "truncated": False, "mask_columns": ["phone"]}
|
||||
#: 会员按地区汇总：同一个设了号码遮罩的会员库，SQL 没碰号码列
|
||||
REGIONS = {"columns": ["region", "spend"], "column_types": {"region": "text", "spend": "number"},
|
||||
"rows": [["华东", 320.0], ["华南", 188.5]], "source": "crm",
|
||||
"sql": "SELECT region, SUM(spend) AS spend FROM members GROUP BY region ORDER BY region", "truncated": False,
|
||||
"mask_columns": ["phone"]}
|
||||
CARD = {"kind": "metric_set", "caliber": "周报口径", "caliber_version": "v2",
|
||||
"metrics": [{"id": "gmv", "name": "销售额", "value": 8381.25, "unit": "元", "decimals": 2, "format": "thousands"}]}
|
||||
|
||||
@@ -55,6 +62,9 @@ WEEKLY = ("## 门店周报\n\n本周销售额 [[m:gmv]]，*按发生周统计*
|
||||
"**会员消费** [[v:Q2.r0.spend]]，另有 42 笔退款待入账。[[see:Q1]]\n\n- 云岭店流水最高\n- 枫桥店流水最低\n\n"
|
||||
"[[table:Q1 cols=store,amount rows=0-7]]")
|
||||
MONTHLY = "## 门店月报\n\n本月销售额 [[m:gmv]]，青石店流水 [[v:Q1.r0.amount]]。[[see:Q1]]"
|
||||
#: 会员周报：Q3 是会员按地区汇总（没碰遮罩列），Q2 是选出了号码的会员消费
|
||||
MEMBERS = ("## 会员周报\n\n华东会员消费 [[v:Q3.r0.spend]]，华南 [[v:Q3.r1.spend]]。[[see:Q3]]\n\n"
|
||||
"会员林一消费 [[v:Q2.r0.spend]]。")
|
||||
|
||||
|
||||
def stored(obj: Any) -> tuple[str, bytes]:
|
||||
@@ -83,21 +93,29 @@ def masked_card(crm_id: str) -> dict[str, Any]:
|
||||
|
||||
|
||||
class Sample:
|
||||
"""一次虚构的已封存运行：工件（原字节）、证据目录、报告文档、事件。masked=True 时口径卡带着取自遮罩列的输入。"""
|
||||
"""一次虚构的已封存运行：工件（原字节）、证据目录、报告文档、事件。masked=True 时口径卡带着取自遮罩列的输入；
|
||||
regions=True 时会员库另有一次按地区汇总的查询（目录里排在 Q3）。"""
|
||||
|
||||
def __init__(self, *, nan: bool = False, masked: bool = False) -> None:
|
||||
def __init__(self, *, nan: bool = False, masked: bool = False, regions: bool = False) -> None:
|
||||
self.files: dict[str, bytes] = {}
|
||||
self.shop_id = self.put(shop(nan))
|
||||
self.crm_id = self.put(CRM)
|
||||
self.card = masked_card(self.crm_id) if masked else CARD
|
||||
self.card_id = self.put(self.card)
|
||||
self.ledger = [
|
||||
{"kind": "query", "node_id": "fetch", "exec": 1, "artifact": self.shop_id, "tool": "db_query__shop",
|
||||
"source": "shop", "columns": ["store", "amount", "ratio"], "rows": len(STORES), "truncated": False},
|
||||
{"kind": "query", "node_id": "members", "exec": 1, "artifact": self.crm_id, "tool": "db_query__crm",
|
||||
"source": "crm", "columns": CRM["columns"], "rows": len(CRM["rows"]), "truncated": False},
|
||||
{"kind": "metric_set", "node_id": "card", "artifact": self.card_id},
|
||||
]
|
||||
self.entries = {
|
||||
"fetch": {"kind": "query", "node_id": "fetch", "exec": 1, "artifact": self.shop_id,
|
||||
"tool": "db_query__shop", "source": "shop", "columns": ["store", "amount", "ratio"],
|
||||
"rows": len(STORES), "truncated": False},
|
||||
"members": {"kind": "query", "node_id": "members", "exec": 1, "artifact": self.crm_id,
|
||||
"tool": "db_query__crm", "source": "crm", "columns": CRM["columns"], "rows": len(CRM["rows"]),
|
||||
"truncated": False},
|
||||
}
|
||||
if regions:
|
||||
self.entries["regions"] = {"kind": "query", "node_id": "regions", "exec": 1,
|
||||
"artifact": self.put(REGIONS), "tool": "db_query__crm", "source": "crm",
|
||||
"columns": REGIONS["columns"], "rows": len(REGIONS["rows"]), "truncated": False}
|
||||
self.entries["card"] = {"kind": "metric_set", "node_id": "card", "artifact": self.card_id}
|
||||
self.ledger = list(self.entries.values())
|
||||
self.catalog = build_catalog(nodes={"card": self.card}, ledger=self.ledger)
|
||||
|
||||
def put(self, obj: Any) -> str:
|
||||
@@ -131,12 +149,17 @@ class Sample:
|
||||
"preview": json.dumps({"columns": ["store", "amount"], "rows": STORES[:2]},
|
||||
ensure_ascii=False)})
|
||||
add("node.finished", "fetch", {"duration_ms": 14, "attempt": 1, "preview": "{\"rows\": …}",
|
||||
"evidence": [self.ledger[0]]})
|
||||
"evidence": [self.entries["fetch"]]})
|
||||
add("tool.end", "members", {"tool": "db_query__crm", "duration_ms": 9, "query_artifact": self.crm_id,
|
||||
"preview": json.dumps(CRM["rows"][:1], ensure_ascii=False)})
|
||||
add("node.finished", "members", {"duration_ms": 10, "attempt": 1, "evidence": [self.ledger[1]]})
|
||||
add("node.finished", "members", {"duration_ms": 10, "attempt": 1, "evidence": [self.entries["members"]]})
|
||||
if "regions" in self.entries:
|
||||
add("tool.end", "regions", {"tool": "db_query__crm", "duration_ms": 7,
|
||||
"query_artifact": self.entries["regions"]["artifact"],
|
||||
"preview": json.dumps(REGIONS["rows"], ensure_ascii=False)})
|
||||
add("node.finished", "regions", {"duration_ms": 8, "attempt": 1, "evidence": [self.entries["regions"]]})
|
||||
add("node.finished", "card", {"duration_ms": 3, "attempt": 1, "preview": {"kind": "metric_set"},
|
||||
"evidence": [self.ledger[2]]})
|
||||
"evidence": [self.entries["card"]]})
|
||||
for node_id, doc_id in reports.items():
|
||||
add("llm.end", node_id, {"model": "mock", "output": "（写作草稿）"})
|
||||
add("report.checked", node_id, {"doc_artifact": doc_id, "ok": True})
|
||||
@@ -167,12 +190,12 @@ class Sample:
|
||||
|
||||
|
||||
def render(sample: Sample, events: list[tuple[int, str, str | None, Any]], bound: int, *, run: dict[str, Any] | None = None,
|
||||
legacy: bool = False, drop: tuple[str, ...] = ()) -> str:
|
||||
"""真实封存哈希的页面。drop：导出时服务端已没有文件的工件。"""
|
||||
legacy: bool = False, drop: tuple[str, ...] = (), source_masks: dict[str, list[str]] | None = None) -> str:
|
||||
"""真实封存哈希的页面。drop：导出时服务端已没有文件的工件；source_masks：数据源现在设的遮罩。"""
|
||||
sealed = [e for e in events if e[0] <= bound]
|
||||
files = {k: v for k, v in sample.files.items() if k not in drop}
|
||||
bundle = build_bundle(run or RUN, events, manifest_seq=bound, manifest_hash=manifest_hash(sealed), read=files.get,
|
||||
exported_at=EXPORTED_AT, legacy=legacy, graph=GRAPH, source_masks={})
|
||||
exported_at=EXPORTED_AT, legacy=legacy, graph=GRAPH, source_masks=source_masks or {})
|
||||
return render_html(bundle)
|
||||
|
||||
|
||||
@@ -215,6 +238,10 @@ def pages() -> dict[str, str]:
|
||||
m = Sample(masked=True)
|
||||
events, bound = m.events({"write": m.doc(WEEKLY, "write")})
|
||||
out["masked-card"] = render(m, events, bound)
|
||||
|
||||
r = Sample(regions=True)
|
||||
events, bound = r.events({"write": r.doc(MEMBERS, "write")})
|
||||
out["masked-source"] = render(r, events, bound, source_masks={"crm": ["phone"]})
|
||||
return out
|
||||
|
||||
|
||||
|
||||
@@ -5,7 +5,8 @@
|
||||
|
||||
- 只给已封存且封存核对一致的运行：没封存、封存后被改过都回 409；
|
||||
- 导出的数据里，事件行按浏览器的拼法重算等于运行记下的 manifest_hash；
|
||||
- 遮罩库的快照只有 id 没有内容，它的查询事件只有哈希；运行之后才设的遮罩同样生效；
|
||||
- 遮罩库上选出了号码的快照只有 id 没有内容，它的查询事件只有哈希；运行之后才设的遮罩同样生效；同一个库上只汇总消费额
|
||||
（没碰号码列）的那次查询，快照照常随文件附带；
|
||||
- 附件名是「<运行名>-可验证报告.html」，中文名按 RFC 5987 编码；
|
||||
- 只读：导出前后运行记录、事件一条不多一条不少。
|
||||
|
||||
@@ -91,11 +92,12 @@ def node(nid, ntype, **config):
|
||||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||||
|
||||
|
||||
def weekly(text_nodes=()):
|
||||
def weekly(text_nodes=(), queries=()):
|
||||
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}),
|
||||
*queries,
|
||||
node("card", "metrics", caliber="周报口径", caliber_version="v2", metrics=[
|
||||
{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands",
|
||||
"expression": "cell(nodes.fetch, 0, 'amount')"}]),
|
||||
@@ -197,6 +199,33 @@ async def test_the_masked_source_ships_only_hashes(client, monkeypatch):
|
||||
assert "000-0001" not in json.dumps(bundle, ensure_ascii=False)
|
||||
|
||||
|
||||
async def test_an_aggregate_on_the_masked_source_ships_its_snapshot(client, monkeypatch):
|
||||
"""会员库设了号码遮罩，另一次查询只汇总了消费额（没碰号码列）：数据源查询工具真落的快照随文件附带，原字节和服务端的
|
||||
文件一致，查询事件带规范字符串，报告里引用它的数字和快照里那一格对得上；选出了号码的那份照旧只给 id。"""
|
||||
script(monkeypatch, "会员消费合计 [[v:Q3.r0.spend]]，会员林一 [[v:Q2.r0.spend]]。[[see:Q3]]")
|
||||
total = node("total", "tool", tool="db_query__bundle_crm",
|
||||
args={"sql": "SELECT COUNT(*) AS members, SUM(spend) AS spend FROM members"})
|
||||
row = await wait((await run_manager.start(graph=weekly(queries=[total]), input_payload={})).id)
|
||||
assert row.status == "succeeded", row.error
|
||||
crm, summed = await query_artifact(row.id, "members"), await query_artifact(row.id, "total")
|
||||
assert json.loads(file_of(summed).read_bytes())["mask_columns"] == ["phone"]
|
||||
|
||||
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 base64.b64decode(by_id[summed]["b64"]) == file_of(summed).read_bytes()
|
||||
assert by_id[crm] == {"id": crm, "kind": "query", "node_id": "members", "masked": True}
|
||||
ends = {e["node_id"]: e for e in bundle["events"] if e["type"] == "tool.end"}
|
||||
assert "canonical" in ends["total"] and "canonical" not in ends["members"]
|
||||
|
||||
doc = json.loads(base64.b64decode(next(a["b64"] for a in bundle["artifacts"] if a["kind"] == "report_doc")))
|
||||
assert doc["catalog"]["Q3"]["artifact"] == summed
|
||||
[seg] = [s for b in doc["blocks"] for u in b["units"] for s in u["segments"]
|
||||
if (s.get("cite") or {}).get("alias") == "Q3"]
|
||||
snapshot = json.loads(base64.b64decode(by_id[summed]["b64"]))
|
||||
assert seg["cite"]["status"] == "resolved" and seg["cite"]["value"] == snapshot["rows"][0][1] == 508.5
|
||||
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")
|
||||
|
||||
@@ -461,9 +461,10 @@ def test_the_sample_pages_for_the_browser_check_are_sealed_for_real():
|
||||
本身：封存哈希是对事件真算的（浏览器的拼法重算得出），各页的形状是 check-bundle 的断言预期的那样。"""
|
||||
from tests.bundle_samples import pages
|
||||
|
||||
got = {name: json.loads(data_block(page)) for name, page in pages().items()}
|
||||
rendered = pages()
|
||||
got = {name: json.loads(data_block(page)) for name, page in rendered.items()}
|
||||
assert set(got) == {"base", "two-reports", "legacy", "failed", "continued", "degraded", "missing-doc",
|
||||
"missing-card", "nan", "stray", "masked-card"}
|
||||
"missing-card", "nan", "stray", "masked-card", "masked-source"}
|
||||
for name, bundle in got.items():
|
||||
assert browser_manifest(bundle["events"]) == bundle["meta"]["manifest_hash"], name
|
||||
|
||||
@@ -502,6 +503,13 @@ def test_the_sample_pages_for_the_browser_check_are_sealed_for_real():
|
||||
# 口径卡上有输入取自遮罩列：整张卡只给 id，页面里找不到那个号码
|
||||
card = next(a for a in got["masked-card"]["artifacts"] if a["kind"] == "metric_set")
|
||||
assert card.get("masked") is True and "b64" not in card
|
||||
# 会员库设了号码遮罩：按地区汇总的那份没碰号码列，随文件附带；选出了号码的那份只给 id，页面里找不到号码
|
||||
source = got["masked-source"]
|
||||
shapes = {(a["kind"], a.get("node_id")): {k for k in a if k in ("b64", "masked", "missing")}
|
||||
for a in source["artifacts"]}
|
||||
assert shapes == {("report_doc", "write"): {"b64"}, ("query", "regions"): {"b64"}, ("query", "members"): {"masked"}}
|
||||
assert [n for n in segments(source) if n.get("cite", {}).get("alias") == "Q3"]
|
||||
assert not any(phone in rendered["masked-source"] for phone in ("000-0001", "000-0002"))
|
||||
nan = got["nan"]
|
||||
assert "NaN" in next(e["canonical"] for e in nan["events"] if e["type"] == "run.started")
|
||||
assert b"NaN" in base64.b64decode(next(a["b64"] for a in nan["artifacts"] if a.get("node_id") == "fetch"))
|
||||
|
||||
@@ -2,7 +2,8 @@
|
||||
|
||||
以前只有查询快照会因遮罩不随文件附带，口径卡一律原样嵌入：证据面板上已经遮住的号码（cell() 取的遮罩格、按位置取的
|
||||
vars.res.rows[0][1]），在这份要发给别人的文件里照样看得到。认哪个输入取自哪份查询结果的哪几列，和证据面板、推演同一个
|
||||
认法（recompute.input_origin）；核对不了的宁可多挡。
|
||||
认法（recompute.input_origin）；核对不了的宁可多挡。那几列算不算遮罩列，除了列名对上的，SQL 里给遮罩列起了别名、套了函数、
|
||||
经子查询或 WITH 转了一道的也算（同质询、双盲复算的 inquiry._derived_hidden）：卡上取的 tel 可能就是 phone AS tel。
|
||||
|
||||
服务端的卡文件被改过（哈希对不上）也照卡上记的输入认：嵌进去的是原字节，号码照样在里面。读得出字节、却解析不出一张卡
|
||||
（截断的 JSON、不是对象）的，认不出输入，整张按带遮罩算、不嵌：和读不出来的快照按带遮罩算是同一个道理。
|
||||
@@ -89,6 +90,106 @@ def test_a_card_reading_only_unmasked_columns_is_embedded():
|
||||
assert "masked" not in entry and base64.b64decode(entry["b64"]) == stored(card)[1]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql", [
|
||||
"SELECT name, phone AS tel, spend FROM members",
|
||||
"SELECT name, tel, spend FROM (SELECT name, phone AS tel, spend FROM members) t",
|
||||
"WITH c AS (SELECT name, SUBSTR(phone, 1) AS tel, spend FROM members) SELECT name, tel, spend FROM c",
|
||||
])
|
||||
def test_a_card_reading_a_renamed_masked_column_ships_only_its_id(sql):
|
||||
"""查询给号码起了别名、经子查询或 WITH 转了一道：卡上取的 tel 那一格就是号码。改了名字的遮罩列按 SQL 认（同质询、
|
||||
双盲复算的 inquiry._derived_hidden），整张卡只给 id。以前只按列名认，号码原样进了文件。"""
|
||||
snap = {**SNAP_MASKED, "columns": ["name", "tel", "spend"], "sql": sql}
|
||||
snap_id, raw = stored(snap)
|
||||
item = {**cell_input("tel", PHONE), "artifact": snap_id, "path": "cell(nodes.members, 0, 'tel')"}
|
||||
bundle, entry = bundle_of(card_with([item], expression="len(cell(nodes.members, 0, 'tel'))"),
|
||||
files_changed={snap_id: raw})
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
assert PHONE not in render_html(bundle)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql", [
|
||||
# 列名表按位置改了名
|
||||
"SELECT name, c AS tel, spend FROM members AS m(a, b, c)",
|
||||
# 整行引用：FROM 里的表函数不写 LATERAL 也看得见前面的来源，整行一列列摊开；直接把整行转成文字
|
||||
"SELECT name, j.value AS tel, spend FROM members m, jsonb_each_text(to_jsonb(m)) j",
|
||||
"SELECT name, to_json(m)::text AS tel, spend FROM members m",
|
||||
# UNION 一支里的 * 按位置对上第一支的列名
|
||||
"SELECT name, tel, spend FROM (SELECT 'x' AS name, 'y' AS tel, 0 AS spend UNION ALL SELECT * FROM members) t",
|
||||
# FROM 里的表函数吃进遮罩列，交出的列叫别的名字
|
||||
"SELECT name, value AS tel, spend FROM members m, json_each_text(json_build_object('p', phone)) j",
|
||||
# 不点列名也读得出表里内容；语法树认不准数据库实际执行什么
|
||||
"SELECT name, table_to_xml('members', true, false, '')::text AS tel, spend FROM members",
|
||||
"SELECT name, /*!50000 phone AS */ tel, spend FROM members",
|
||||
# 函数记法交出整行的文字写法；表函数看不见排在它后面的派生来源里叫 m 的列，m 还是整行
|
||||
"SELECT name, m.record_out AS tel, spend FROM members m",
|
||||
"SELECT name, value AS tel, spend FROM members m, jsonb_each_text(to_jsonb(m)), (SELECT 1 AS m) s",
|
||||
])
|
||||
def test_a_card_reading_a_column_whose_origin_the_sql_hides_ships_only_its_id(sql):
|
||||
"""快照的 SQL 认不出结果里哪一列取自哪一列（快照本身因此只给哈希）：卡上取的 tel 可能就是号码，整份快照的列都按
|
||||
遮罩列算，整张卡只给 id。以前这条路只按列名和改了名字的遮罩列认，快照只给哈希，卡上记着的号码却原样进了文件。"""
|
||||
snap = {**SNAP_MASKED, "columns": ["name", "tel", "spend"], "sql": sql}
|
||||
snap_id, raw = stored(snap)
|
||||
item = {**cell_input("tel", PHONE), "artifact": snap_id, "path": "cell(nodes.members, 0, 'tel')"}
|
||||
bundle, entry = bundle_of(card_with([item], expression="len(cell(nodes.members, 0, 'tel'))"),
|
||||
files_changed={snap_id: raw})
|
||||
assert entry.get("masked") is True and "b64" not in entry, sql
|
||||
assert PHONE not in render_html(bundle)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql", [
|
||||
# 每种读法都解析不了：PG17 起的 JSON_QUERY … WITH WRAPPER（这里把整行装进了 tel）、ORDER BY … USING
|
||||
"SELECT name, JSON_QUERY(to_jsonb(m), '$.*' WITH WRAPPER) AS tel, spend FROM members m",
|
||||
"SELECT name, value AS tel, spend FROM members m, jsonb_each_text(to_jsonb(m)) ORDER BY value USING <",
|
||||
"SELECT name, tel, spend FROM (SELECT * FROM members) AS x(name, tel, spend) ORDER BY spend USING <",
|
||||
# 没有 SQL（数据源查询工具落的快照都带着）、SQL 是空的
|
||||
None,
|
||||
"",
|
||||
])
|
||||
def test_a_card_reading_a_snapshot_whose_sql_does_not_parse_ships_only_its_id(sql):
|
||||
"""快照的 SQL 每种读法都解析不了（或者根本没有）：快照本身只给哈希（_untouched），认不出结果里哪一列取自哪一列，
|
||||
卡上取自这份快照任一列的输入都按遮罩列算，整张卡只给 id——和快照同一个结论。以前这条路退回按 SQL 认遮罩列的那一套，
|
||||
解析不了时它只看 SQL 里写没写着 phone：没写就只认 phone 本身，卡上取的 tel 是整行、号码原样进了文件。"""
|
||||
snap = {**SNAP_MASKED, "columns": ["name", "tel", "spend"], "sql": sql}
|
||||
if sql is None:
|
||||
del snap["sql"]
|
||||
snap_id, raw = stored(snap)
|
||||
for column, value in (("tel", PHONE), ("spend", 320.0)):
|
||||
item = {**cell_input(column, value), "artifact": snap_id, "path": f"cell(nodes.members, 0, '{column}')"}
|
||||
bundle, entry = bundle_of(card_with([item], substituted=repr(value),
|
||||
expression=f"len(cell(nodes.members, 0, '{column}'))"),
|
||||
files_changed={snap_id: raw})
|
||||
assert entry.get("masked") is True and "b64" not in entry, (sql, column)
|
||||
assert PHONE not in render_html(bundle)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns", [
|
||||
("SELECT name, phone AS tel, spend FROM members", ["name", "tel", "spend"]),
|
||||
# 最外层的 * 照原来的名字展开：号码还在 phone 那一列
|
||||
("SELECT * FROM members", ["name", "phone", "spend"]),
|
||||
])
|
||||
def test_a_card_reading_an_unrenamed_column_of_that_query_is_embedded(sql, columns):
|
||||
"""同一条查询，卡上取的是消费额（不取自号码）：照常嵌入。"""
|
||||
snap = {**SNAP_MASKED, "columns": columns, "sql": sql}
|
||||
snap_id, raw = stored(snap)
|
||||
item = {**cell_input("spend", 320.0), "artifact": snap_id}
|
||||
card = card_with([item], substituted="320.0", expression="cell(nodes.members, 0, 'spend')")
|
||||
_, entry = bundle_of(card, files_changed={snap_id: raw})
|
||||
assert "masked" not in entry and base64.b64decode(entry["b64"]) == stored(card)[1]
|
||||
|
||||
|
||||
def test_an_error_while_finding_renamed_columns_masks_the_whole_snapshot(monkeypatch):
|
||||
"""按 SQL 认改了名字的遮罩列时出了意外：认不出哪几列取自遮罩列，整份快照的列都按遮罩算，卡上取消费额也只给 id。"""
|
||||
from app.engine import inquiry
|
||||
|
||||
def broken(*_args, **_kwargs):
|
||||
raise RuntimeError("解析器坏了")
|
||||
|
||||
monkeypatch.setattr(inquiry, "_derived_hidden", broken)
|
||||
card = card_with([cell_input("spend", 320.0)], substituted="320.0", expression="cell(nodes.members, 0, 'spend')")
|
||||
_, entry = bundle_of(card)
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
|
||||
|
||||
def test_a_mask_set_on_the_source_now_counts_too():
|
||||
card = card_with([cell_input("spend", 320.0)], substituted="320.0", expression="cell(nodes.members, 0, 'spend')")
|
||||
_, entry = bundle_of(card, source_masks={"crm": ["spend"]})
|
||||
|
||||
@@ -0,0 +1,460 @@
|
||||
"""可验证报告里数据源设了遮罩的查询快照（期 5 前置 L2）：能证明这条查询没碰遮罩列的，快照照常随文件附带。
|
||||
|
||||
以前只要查询当时记下了数据源的遮罩列（mask_columns），或者数据源现在设了遮罩，整份快照就只给哈希——哪怕这条查询只按
|
||||
地区汇总了金额。数据源里只要有一列设了遮罩（会员手机号），报告里引用查询格子的数字就全成了「无法核对」。
|
||||
|
||||
现在全部满足才嵌入（build._untouched）：认得这份快照（数据源查询工具落的那种、哈希对得上）；结果的列名对不上遮罩列；
|
||||
SQL 原文里没写到遮罩列（连条件里也不行：SQL 随快照一起附带，WHERE phone = '…' 的字面量就是遮罩值），也没写着这次运行
|
||||
别的查询里查到过的遮罩值；没有不点列名也读得出表里内容的写法（字符串里的 SQL 交给函数执行、统计信息里的采样值、
|
||||
系统 schema）；SQL 解析得了，没有认不出取自哪一列的写法（列名表、整行引用——FROM 里的表函数不写 LATERAL 也看得见前面的
|
||||
来源、可能展开出遮罩列的 *），按 SQL 认遮罩列的那一套（inquiry._derived_hidden）也认不出哪个结果列取自遮罩列。任何一条
|
||||
不满足、判断过程出错，照旧只给哈希。
|
||||
|
||||
工件、事件都是手搭的（会员、号码、地区、金额都是编的），照 tests/test_bundle_build.py 的样子。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from app.bundle.build import build_bundle, render_html
|
||||
from app.engine.expressions import cell_value, column_kind, same_value
|
||||
from app.engine.merge_query import MERGE_SOURCE
|
||||
from tests.test_bundle_build import GRAPH, RUN, stored
|
||||
|
||||
#: 会员库的遮罩：手机号
|
||||
MASKS = {"crm": ["phone"]}
|
||||
PHONES = ("000-0001", "000-0002")
|
||||
AGGREGATE = "SELECT region, SUM(spend) AS spend FROM members GROUP BY region ORDER BY region"
|
||||
|
||||
|
||||
def snapshot(sql: str, columns: list[str], rows: list[list], **extra) -> dict:
|
||||
"""会员库（crm）的一份查询快照，照数据源查询工具落的样子：查询当时记下了遮罩列。"""
|
||||
return {"columns": columns, "rows": rows, "row_count": len(rows), "truncated": False, "elapsed_ms": 3,
|
||||
"sql": sql, "source": "crm", "mask_columns": ["phone"], **extra}
|
||||
|
||||
|
||||
REGIONS = snapshot(AGGREGATE, ["region", "spend"], [["华东", 508.5], ["华南", 300.0]],
|
||||
column_types={"region": "text", "spend": "number"})
|
||||
|
||||
|
||||
def cite(value, row: int, column: str) -> dict:
|
||||
return {"alias": "Q1", "kind": "cell", "locator": {"row": row, "column": column}, "role": "value",
|
||||
"status": "resolved", "value": value, "rendered": str(value)}
|
||||
|
||||
|
||||
def bundle_of(snap: dict, *, value=508.5, row: int = 0, column: str = "spend", masks=MASKS, raw: bytes | None = None,
|
||||
artifact: str | None = None, others: tuple[dict, ...] = ()):
|
||||
"""一次手搭的运行：会员库一次查询 → 报告引用快照里的一格（默认第 0 行的 spend）。raw 给了就把服务端的快照文件换成
|
||||
这份字节；artifact 给了就用它当快照的 id（文件被改过、哈希对不上）。others 是同一次运行里另外几次会员库的查询
|
||||
（门店流水节点查的，报告没引用），排在这次查询之后。"""
|
||||
snap_id, snap_raw = stored(snap)
|
||||
snap_id, snap_raw = artifact or snap_id, raw if raw is not None else snap_raw
|
||||
doc = {"schema": "agentlab.report/1", "run_id": RUN["id"], "node_id": "write", "markdown": "…",
|
||||
"catalog": {"Q1": {"alias": "Q1", "kind": "query", "artifact": snap_id, "node_id": "members"}},
|
||||
"blocks": [{"id": "b0", "type": "paragraph", "units": [{
|
||||
"id": "u0", "kind": "claim", "cites": ["Q1"],
|
||||
"segments": [{"id": "s0", "kind": "number", "text": str(value), "state": "deterministic",
|
||||
"cite": cite(value, row, column)}]}]}]}
|
||||
doc_id, doc_raw = stored(doc)
|
||||
files = {snap_id: snap_raw, doc_id: doc_raw}
|
||||
later = []
|
||||
for other in others:
|
||||
other_id, other_raw = stored(other)
|
||||
files[other_id] = other_raw
|
||||
later += [("tool.end", "fetch", {"tool": "db_query__crm", "query_artifact": other_id,
|
||||
"preview": json.dumps(other["rows"][:1], ensure_ascii=False)}),
|
||||
("node.finished", "fetch",
|
||||
{"evidence": [{"kind": "query", "node_id": "fetch", "artifact": other_id}]})]
|
||||
events = [(seq, *event) for seq, event in enumerate([
|
||||
("run.started", None, {"nodes": 3}),
|
||||
("tool.end", "members", {"tool": "db_query__crm", "query_artifact": snap_id,
|
||||
"preview": json.dumps(snap["rows"][:1], ensure_ascii=False)}),
|
||||
("node.finished", "members", {"evidence": [{"kind": "query", "node_id": "members", "artifact": snap_id}]}),
|
||||
*later,
|
||||
("report.checked", "write", {"doc_artifact": doc_id, "ok": True}),
|
||||
("node.finished", "write", {"preview": {"text": "…", "doc_artifact": doc_id}}),
|
||||
("run.finished", None, {"output": {}}),
|
||||
], start=1)]
|
||||
bundle = build_bundle(RUN, events, manifest_seq=len(events), manifest_hash="0" * 64, graph=GRAPH,
|
||||
source_masks=masks, read=files.get, exported_at="2026-10-06T08:00:00+00:00")
|
||||
return bundle, next(a for a in bundle["artifacts"] if a["id"] == snap_id)
|
||||
|
||||
|
||||
def browser_would_pass(bundle: dict, entry: dict) -> bool:
|
||||
"""照浏览器 ⑤ 的办法核一遍：引用值和嵌进来的快照里那一格相等（cell_value / same_value，verifier.js 移植的就是它们）。"""
|
||||
snap = json.loads(base64.b64decode(entry["b64"]))
|
||||
doc_entry = next(a for a in bundle["artifacts"] if a["kind"] == "report_doc")
|
||||
doc = json.loads(base64.b64decode(doc_entry["b64"]))
|
||||
seg = doc["blocks"][0]["units"][0]["segments"][0]
|
||||
loc = seg["cite"]["locator"]
|
||||
index = snap["columns"].index(loc["column"])
|
||||
truth = cell_value(snap["rows"][loc["row"]][index], column_kind(snap, loc["column"]))
|
||||
return same_value(seg["cite"]["value"], truth)
|
||||
|
||||
|
||||
def shown(bundle: dict, seq: int) -> bool:
|
||||
return "canonical" in next(e for e in bundle["events"] if e["seq"] == seq)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 没碰遮罩列：照常附带
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("masks, recorded", [
|
||||
(MASKS, True), # 查询当时记下了遮罩，数据源现在也设着
|
||||
({}, True), # 只有查询当时记下的（数据源事后取消了遮罩、改了名）
|
||||
(MASKS, False), # 查询之后才给数据源设的遮罩
|
||||
])
|
||||
def test_an_aggregate_by_region_on_a_masked_source_is_embedded(masks, recorded):
|
||||
"""数据源设了手机号遮罩、查询按地区汇总金额：快照原字节随文件附带，引用它的数字核对得过；查询事件也带规范字符串。"""
|
||||
snap = REGIONS if recorded else {k: v for k, v in REGIONS.items() if k != "mask_columns"}
|
||||
bundle, entry = bundle_of(snap, masks=masks)
|
||||
assert "masked" not in entry and base64.b64decode(entry["b64"]) == stored(snap)[1]
|
||||
assert browser_would_pass(bundle, entry)
|
||||
assert shown(bundle, 2)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, rows", [
|
||||
# 先在 WITH 里把一张表整个选出来（* 展开的列只按名字往外传），外层按名字取
|
||||
("WITH paid AS (SELECT * FROM members WHERE spend > 0) SELECT region, SUM(spend) AS spend FROM paid "
|
||||
"GROUP BY region ORDER BY region", ["region", "spend"], [["华东", 508.5]]),
|
||||
# FROM 里的子查询、LATERAL 里的 *：同上
|
||||
("SELECT t.region, SUM(t.spend) AS spend FROM (SELECT * FROM members) AS t GROUP BY t.region",
|
||||
["region", "spend"], [["华东", 508.5]]),
|
||||
("SELECT m.region, SUM(o.amount) AS spend FROM members m JOIN LATERAL "
|
||||
"(SELECT * FROM orders o2 WHERE o2.member_id = m.id) o ON TRUE GROUP BY m.region", ["region", "spend"],
|
||||
[["华东", 508.5]]),
|
||||
# EXISTS 里的 * 只决定有没有，带不出值
|
||||
("SELECT region, SUM(spend) AS spend FROM members m WHERE NOT EXISTS (SELECT * FROM blocked b WHERE b.id = m.id) "
|
||||
"GROUP BY region", ["region", "spend"], [["华东", 508.5]]),
|
||||
# COUNT(*) 只数个数
|
||||
("SELECT region, COUNT(*) AS spend FROM members GROUP BY region", ["region", "spend"], [["华东", 508.5]]),
|
||||
# * 展开的是写明了列的派生来源：列名都看得见
|
||||
("WITH s AS (SELECT region, SUM(spend) AS spend FROM members GROUP BY region) SELECT * FROM s",
|
||||
["region", "spend"], [["华东", 508.5]]),
|
||||
# MySQL、SQLite 的反引号：通用读法解析不了，SQLite / MySQL 的读法读得通就行
|
||||
("SELECT `region`, SUM(`spend`) AS `spend` FROM `members` GROUP BY `region`", ["region", "spend"],
|
||||
[["华东", 508.5]]),
|
||||
# 列和 WITH 里的那一段同名：那一段明写出了这一列，是列不是整行引用
|
||||
("WITH spend AS (SELECT region, SUM(spend) AS spend FROM members GROUP BY region) "
|
||||
"SELECT region, spend FROM spend ORDER BY spend DESC", ["region", "spend"], [["华东", 508.5]]),
|
||||
# 列名碰巧像统计视图、导出函数的名字（不跟左括号的不是函数），FROM 里的表函数吃的是没设遮罩的列
|
||||
("SELECT region, SUM(spend) AS spend FROM members WHERE pg_status = 'paid' AND crosstab IS NULL GROUP BY region",
|
||||
["region", "spend"], [["华东", 508.5]]),
|
||||
("SELECT m.region, SUM(m.spend) AS spend FROM members m, json_each_text(m.extra) j WHERE j.key = 'vip' "
|
||||
"GROUP BY m.region", ["region", "spend"], [["华东", 508.5]]),
|
||||
# 括号括起来的一组连接，按名字取列
|
||||
("SELECT m.region, SUM(o.amount) AS spend FROM (members m JOIN orders o ON o.member_id = m.id) GROUP BY m.region",
|
||||
["region", "spend"], [["华东", 508.5]]),
|
||||
# 派生来源明写出了叫 max 的列：t.max 是那一列，不是函数记法的 max(t)
|
||||
("SELECT t.region, t.max AS spend FROM (SELECT region, MAX(spend) AS max FROM members GROUP BY region) t",
|
||||
["region", "spend"], [["华东", 508.5]]),
|
||||
# 表函数的参数没写表名、和它自己的别名同名：排在它前面的派生来源明写出了这一列，是列不是整行引用
|
||||
("SELECT p.region, SUM(p.spend) AS spend FROM (SELECT region, spend, attrs FROM members) p, "
|
||||
"jsonb_each_text(attrs) AS attrs WHERE attrs.key = 'vip' GROUP BY p.region", ["region", "spend"],
|
||||
[["华东", 508.5]]),
|
||||
# Oracle 的 sys.dual 不是 MySQL sys 库里记着别的会话 SQL 的视图
|
||||
("SELECT region, SUM(spend) AS spend FROM members CROSS JOIN sys.dual GROUP BY region", ["region", "spend"],
|
||||
[["华东", 508.5]]),
|
||||
])
|
||||
def test_queries_that_only_read_unmasked_columns_are_embedded(sql, columns, rows):
|
||||
bundle, entry = bundle_of(snapshot(sql, columns, rows))
|
||||
assert "masked" not in entry and "b64" in entry, sql
|
||||
assert browser_would_pass(bundle, entry)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 用到或可能用到遮罩列：只给哈希
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
MASKED_SHAPES = [
|
||||
# 直接选出遮罩列
|
||||
("SELECT name, phone, spend FROM members", ["name", "phone", "spend"]),
|
||||
# 起别名
|
||||
("SELECT name, phone AS tel, spend FROM members", ["name", "tel", "spend"]),
|
||||
# 经子查询转一道
|
||||
("SELECT t.x, t.spend FROM (SELECT phone AS x, spend FROM members) t", ["x", "spend"]),
|
||||
# 经 WITH 转一道
|
||||
("WITH c AS (SELECT phone AS p, spend FROM members) SELECT p AS contact, spend FROM c", ["contact", "spend"]),
|
||||
# SELECT *：结果里有遮罩列
|
||||
("SELECT * FROM members", ["id", "name", "phone", "region", "spend"]),
|
||||
# 函数包着遮罩列，起不起别名都一样
|
||||
("SELECT SUBSTR(phone, 5) AS tail, spend FROM members", ["tail", "spend"]),
|
||||
("SELECT UPPER(phone), spend FROM members", ["UPPER(phone)", "spend"]),
|
||||
# 遮罩列只在条件里：SQL 随快照一起附带，条件里的号码就是遮罩值
|
||||
("SELECT region, SUM(spend) AS spend FROM members WHERE phone = '000-0001' GROUP BY region", ["region", "spend"]),
|
||||
# 汇总里数遮罩列的个数、只在 IS NOT NULL 里：结果里没有号码，也算碰了。有意比「按 SQL 认不出结果列取自遮罩列」
|
||||
# 更严：FROM 里的表函数吃进遮罩列（unnest(ARRAY[phone])）时，它交出的列取自哪一列认不出，只能看 SQL 写没写到
|
||||
("SELECT region, COUNT(DISTINCT phone) AS spend FROM members GROUP BY region", ["region", "spend"]),
|
||||
("SELECT unnest AS spend FROM members, unnest(ARRAY[phone])", ["spend"]),
|
||||
# 换个大小写、加引号照样认得出
|
||||
('SELECT region, SUM(spend) AS spend FROM members WHERE "PHONE" IS NOT NULL GROUP BY region', ["region", "spend"]),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns", MASKED_SHAPES)
|
||||
def test_queries_that_use_a_masked_column_ship_only_their_id(sql, columns):
|
||||
rows = [[f"v{i}" if c != "spend" else 508.5 for i, c in enumerate(columns)]]
|
||||
if "phone" in columns:
|
||||
rows[0][columns.index("phone")] = PHONES[0]
|
||||
bundle, entry = bundle_of(snapshot(sql, columns, rows))
|
||||
assert entry == {"id": entry["id"], "kind": "query", "node_id": "members", "masked": True}, sql
|
||||
assert not shown(bundle, 2)
|
||||
assert PHONES[0] not in render_html(bundle)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns", [
|
||||
# * 展开了数据源里的表：不知道它展开了哪些列，宁可多挡（哪怕结果的列名对不上遮罩列）
|
||||
("SELECT * FROM orders", ["id", "region", "spend"]),
|
||||
("SELECT o.* FROM orders o", ["id", "region", "spend"]),
|
||||
("SELECT * FROM (SELECT * FROM orders WHERE spend > 0) t", ["id", "region", "spend"]),
|
||||
# * 落在 UNION 的一支里：按位置对上第一支的列名，列名认不出取自哪一列
|
||||
("SELECT region, spend FROM (SELECT 'x' AS region, 0 AS spend UNION ALL SELECT * FROM members2) t",
|
||||
["region", "spend"]),
|
||||
# 标量子查询、套在函数里的 *：值装进了另一个名字的列
|
||||
("SELECT (SELECT * FROM members2 LIMIT 1) AS spend", ["spend"]),
|
||||
("SELECT to_json(m.*) AS spend FROM members m", ["spend"]),
|
||||
# 整行引用（Postgres）：一整行装进一列；子查询里引用外层的别名、没起别名的表名也算
|
||||
("SELECT to_json(m) AS spend FROM members m", ["spend"]),
|
||||
("SELECT m AS spend FROM members m", ["spend"]),
|
||||
("SELECT (SELECT to_json(m) FROM orders o WHERE o.member_id = m.id) AS spend FROM members m", ["spend"]),
|
||||
("SELECT row_to_json(members) AS spend FROM members", ["spend"]),
|
||||
# FROM、JOIN 里的表函数（Postgres）不写 LATERAL 也看得见前面的来源：整行交给它，一列列竖着摊开（逐列做数据画像的
|
||||
# 常见写法，结果的 key、value 里就是 phone 和号码）。逗号、CROSS JOIN、JOIN … ON TRUE、LEFT JOIN、ROWS FROM、
|
||||
# 不带别名的 UNNEST、套在 FROM 的子查询里都一样
|
||||
("SELECT key, value, COUNT(*) AS n FROM members m, jsonb_each_text(to_jsonb(m)) GROUP BY key, value "
|
||||
"ORDER BY n DESC LIMIT 20", ["key", "value", "n"]),
|
||||
("SELECT j.key AS k, j.value AS spend FROM members m CROSS JOIN jsonb_each_text(to_jsonb(m)) j", ["k", "spend"]),
|
||||
("SELECT j.value AS spend FROM members m JOIN jsonb_each_text(row_to_json(m)::jsonb) j ON TRUE", ["spend"]),
|
||||
("SELECT r.value AS spend FROM members m LEFT JOIN jsonb_each_text(to_jsonb(m)) r ON TRUE", ["spend"]),
|
||||
("SELECT value AS spend FROM members m, ROWS FROM (jsonb_each_text(to_jsonb(m)))", ["spend"]),
|
||||
("SELECT value AS spend FROM members m, each(hstore(m))", ["spend"]),
|
||||
("SELECT value AS spend FROM members, json_each_text(row_to_json(members))", ["spend"]),
|
||||
("SELECT unnest AS spend FROM members m, unnest(ARRAY[to_jsonb(m)])", ["spend"]),
|
||||
("SELECT v AS spend FROM (SELECT value AS v FROM members m, jsonb_each_text(to_jsonb(m))) t", ["spend"]),
|
||||
# 括号括起来的一组连接：组里的表函数看得见组里前面的来源，外层也看得见组里的名字
|
||||
("SELECT value AS spend FROM (members m CROSS JOIN jsonb_each_text(to_jsonb(m))) AS x", ["spend"]),
|
||||
("SELECT value AS spend FROM members m JOIN (orders o CROSS JOIN jsonb_each_text(to_jsonb(m))) ON TRUE", ["spend"]),
|
||||
("SELECT to_json(m) AS spend FROM (members m JOIN orders o ON o.member_id = m.id)", ["spend"]),
|
||||
# WITH 里的 * 只按名字往外传，可外层把那一段整行交给了表函数
|
||||
("WITH c AS (SELECT * FROM members) SELECT region, value AS spend FROM c, json_each_text(row_to_json(c))",
|
||||
["region", "spend"]),
|
||||
# 函数记法：m.to_jsonb 就是 to_jsonb(m)。整行的文字、二进制写法（record_out、record_send），汇总函数也能这么写
|
||||
# （any_value、*_agg_strict，max、min 交回最大、最小的那一整行）；WITH 里的 * 只按名字往外传，可外层整行交了出去
|
||||
("SELECT m.to_jsonb AS spend FROM members m", ["spend"]),
|
||||
("SELECT m.row_to_json AS spend FROM members m", ["spend"]),
|
||||
("SELECT m.record_out AS spend FROM members m", ["spend"]),
|
||||
("SELECT m.record_send AS spend FROM members m", ["spend"]),
|
||||
("SELECT m.any_value AS spend FROM members m", ["spend"]),
|
||||
("SELECT m.json_agg_strict AS spend FROM members m", ["spend"]),
|
||||
("SELECT m.jsonb_agg_strict AS spend FROM members m", ["spend"]),
|
||||
("SELECT m.max AS spend FROM members m", ["spend"]),
|
||||
("SELECT region, m.min AS spend FROM members m GROUP BY region", ["region", "spend"]),
|
||||
("WITH c AS (SELECT * FROM members) SELECT region, c.record_out AS spend FROM c", ["region", "spend"]),
|
||||
# 表函数只看得见排在它前面的来源：后面那个派生来源里叫 m 的列它看不见，m 还是整行
|
||||
("SELECT value AS spend FROM members m, jsonb_each_text(to_jsonb(m)), (SELECT 1 AS m) s", ["spend"]),
|
||||
("SELECT value AS spend FROM (members m CROSS JOIN jsonb_each_text(to_jsonb(m))) AS x, (SELECT 1 AS m) s",
|
||||
["spend"]),
|
||||
# 列名表按位置给列改名（Postgres 的表别名、子查询、WITH 都能写）
|
||||
("SELECT c AS spend FROM members AS m(a, b, c)", ["spend"]),
|
||||
("WITH t(a, b) AS (SELECT * FROM members2) SELECT b AS spend FROM t", ["spend"]),
|
||||
# 语法树认不准数据库实际执行什么：MySQL、MariaDB 会执行的注释，Postgres 的 Unicode 转义名字
|
||||
("SELECT region /*! , SUM(spend) */ AS spend FROM members GROUP BY region", ["region", "spend"]),
|
||||
("SELECT region, c AS spend FROM (SELECT 1 AS region, 2 AS c /*M! UNION ALL SELECT * FROM contacts */) t",
|
||||
["region", "spend"]),
|
||||
('SELECT U&"\\0073pend" AS spend FROM members', ["spend"]),
|
||||
])
|
||||
def test_shapes_that_could_hide_a_masked_column_ship_only_their_id(sql, columns):
|
||||
"""SQL 里没写到遮罩列、结果的列名也对不上，可认不出结果里哪一列取自哪一列：照样只给哈希，查询事件也只给哈希。"""
|
||||
bundle, entry = bundle_of(snapshot(sql, columns, [[508.5] * len(columns)]), column=columns[-1])
|
||||
assert entry.get("masked") is True and "b64" not in entry, sql
|
||||
assert not shown(bundle, 2)
|
||||
|
||||
|
||||
def test_profiling_every_column_of_a_masked_table_ships_only_its_id():
|
||||
"""逐列做数据画像（整行摊成键值、按值数个数）：结果里就是 phone 和号码。快照、查询事件都只给哈希，页面里找不到号码。
|
||||
以前 FROM 里的表函数按看不见前面的来源算，整行引用认不出，快照和事件的预览原样带出了号码。"""
|
||||
sql = ("SELECT key, value, COUNT(*) AS n FROM members m, jsonb_each_text(to_jsonb(m)) GROUP BY key, value "
|
||||
"ORDER BY n DESC LIMIT 20")
|
||||
snap = snapshot(sql, ["key", "value", "n"], [["phone", PHONES[0], 3], ["region", "华东", 2]])
|
||||
bundle, entry = bundle_of(snap, value=3, column="n")
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
assert not shown(bundle, 2)
|
||||
assert PHONES[0] not in render_html(bundle)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns", [
|
||||
# 表名、SQL 写在字符串里交给函数执行、整表导出：语法树里看不到 *，也看不到遮罩列（Postgres 的 XML 一族、ts_stat，
|
||||
# Oracle 的 DBMS_XMLGEN）
|
||||
("SELECT table_to_xml('members', true, false, '') AS x, 3 AS spend", ["x", "spend"]),
|
||||
("SELECT query_to_xml('select * from members', true, false, '') AS spend", ["spend"]),
|
||||
("SELECT DBMS_XMLGEN.GETXML('SELECT * FROM members') AS spend FROM dual", ["spend"]),
|
||||
("SELECT word AS region, ndoc AS spend FROM ts_stat('SELECT search_text FROM members')", ["region", "spend"]),
|
||||
# 统计信息里的采样值（Postgres 的 pg_stats、SQLite 的 sqlite_stat4、MySQL 的 column_statistics、Oracle 的直方图和
|
||||
# 列统计）
|
||||
("SELECT attname AS region, most_common_vals::text AS vals, n_distinct AS spend FROM pg_stats "
|
||||
"WHERE tablename = 'members'", ["region", "vals", "spend"]),
|
||||
("SELECT tbl AS region, sample AS spend FROM sqlite_stat4", ["region", "spend"]),
|
||||
("SELECT histogram AS spend FROM information_schema.column_statistics WHERE table_name = 'members'", ["spend"]),
|
||||
("SELECT endpoint_actual_value AS spend FROM all_tab_histograms WHERE table_name = 'MEMBERS'", ["spend"]),
|
||||
("SELECT low_value AS spend FROM user_tab_columns WHERE table_name = 'MEMBERS'", ["spend"]),
|
||||
# 按页、按文件、按日志读出的原始数据；别的会话执行过的 SQL 原文（写入语句里就是值）
|
||||
("SELECT get_raw_page('members', 0) AS spend", ["spend"]),
|
||||
("SELECT pg_read_file('base/16384/16385') AS spend", ["spend"]),
|
||||
("SELECT LOAD_FILE('/var/lib/mysql/crm/members.ibd') AS spend", ["spend"]),
|
||||
("SELECT data AS spend FROM pg_logical_slot_peek_changes('cdc', NULL, NULL)", ["spend"]),
|
||||
("SELECT argument AS spend FROM mysql.general_log", ["spend"]),
|
||||
# 别的会话执行过的 SQL 原文的另几种写法：Postgres 里 pg_stat_activity 背后的函数和按会话取的那一个、
|
||||
# pg_stat_statements、本连接预备过的语句（连接池里的连接留着前几次查询的原文）；MySQL sys 库建在
|
||||
# performance_schema 上的视图（带 x$ 的是不排版的那一份）、企业版的审计日志；Oracle 打开着的游标和审计记录
|
||||
("SELECT query AS spend FROM pg_stat_get_activity(NULL)", ["spend"]),
|
||||
("SELECT pg_stat_get_backend_activity(1) AS spend", ["spend"]),
|
||||
("SELECT query AS spend FROM pg_stat_statements", ["spend"]),
|
||||
("SELECT statement AS spend FROM pg_prepared_statements", ["spend"]),
|
||||
("SELECT last_statement AS spend FROM sys.session", ["spend"]),
|
||||
("SELECT current_statement AS spend FROM sys.x$processlist", ["spend"]),
|
||||
("SELECT `last_statement` AS spend FROM `sys`.`x$session`", ["spend"]),
|
||||
("SELECT waiting_query AS spend FROM sys.innodb_lock_waits", ["spend"]),
|
||||
("SELECT waiting_query AS spend FROM sys.x$schema_table_lock_waits", ["spend"]),
|
||||
("SELECT audit_log_read() AS spend", ["spend"]),
|
||||
("SELECT sql_text AS spend FROM v$open_cursor", ["spend"]),
|
||||
("SELECT sql_text AS spend FROM dba_audit_trail", ["spend"]),
|
||||
("SELECT sql_text AS spend FROM unified_audit_trail", ["spend"]),
|
||||
("SELECT lsqltext AS spend FROM sys.fga_log$", ["spend"]),
|
||||
# 按页、按日志、按文件读的另几种写法：pageinspect 按索引名读 B 树页，pg_walinspect 读 WAL 里的块，SQLite 的
|
||||
# readfile；统计信息的另几处：Oracle 直方图和列统计的底表，MariaDB 的 column_stats
|
||||
("SELECT data AS spend FROM bt_page_items('members_phone_idx', 1)", ["spend"]),
|
||||
("SELECT block_data AS spend FROM pg_get_wal_block_info('0/0', '0/FFFFFFFF')", ["spend"]),
|
||||
("SELECT readfile('crm.db') AS spend", ["spend"]),
|
||||
("SELECT epvalue AS spend FROM sys.histgrm$", ["spend"]),
|
||||
("SELECT lowval AS spend FROM sys.hist_head$", ["spend"]),
|
||||
("SELECT min_value AS spend FROM mysql.column_stats WHERE table_name = 'members'", ["spend"]),
|
||||
# 系统 schema
|
||||
("SELECT relname AS region, reltuples AS spend FROM pg_catalog.pg_class", ["region", "spend"]),
|
||||
])
|
||||
def test_reads_that_name_no_column_ship_only_their_id(sql, columns):
|
||||
"""不点列名也读得出表里内容的写法：SQL 里没写到遮罩列、结果的列名也对不上、语法树里也没有 * 和整行引用，结果里照样
|
||||
可能是号码。照样只给哈希。"""
|
||||
bundle, entry = bundle_of(snapshot(sql, columns, [[508.5] * len(columns)]), column=columns[-1])
|
||||
assert entry.get("masked") is True and "b64" not in entry, sql
|
||||
assert not shown(bundle, 2)
|
||||
|
||||
|
||||
#: 同一次运行里先查过号码的那份快照（模型在这里看到了号码）
|
||||
LOOKUP = snapshot("SELECT name, phone FROM members ORDER BY id", ["name", "phone"],
|
||||
[["林一", PHONES[0]], ["周二", PHONES[1]]])
|
||||
|
||||
|
||||
def test_a_query_quoting_a_masked_value_seen_in_the_run_ships_only_its_id():
|
||||
"""模型拿在遮罩查询里看到的号码去查别的表：SQL 里没写遮罩列，可字面量就是号码，SQL 随快照附带就带出了它。这次运行里
|
||||
查到过的遮罩值写在 SQL 里的，只给哈希，查询事件也只给哈希。"""
|
||||
sql = (f"SELECT region, SUM(spend) AS spend FROM members WHERE id IN "
|
||||
f"(SELECT member_id FROM tickets WHERE caller = '{PHONES[1]}') GROUP BY region")
|
||||
bundle, entry = bundle_of(snapshot(sql, ["region", "spend"], [["华东", 508.5]]), others=(LOOKUP,))
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
assert not shown(bundle, 2)
|
||||
assert PHONES[1] not in render_html(bundle)
|
||||
|
||||
|
||||
def test_a_literal_that_is_not_a_masked_value_is_still_embedded():
|
||||
"""同一次运行里查过号码，可这条 SQL 的字面量不是其中哪一个（地区名、别的号码）：照常附带。"""
|
||||
sql = ("SELECT region, SUM(spend) AS spend FROM members WHERE id IN "
|
||||
"(SELECT member_id FROM tickets WHERE caller = '000-0009') AND region = '华东' GROUP BY region")
|
||||
bundle, entry = bundle_of(snapshot(sql, ["region", "spend"], [["华东", 508.5]]), others=(LOOKUP,))
|
||||
assert "masked" not in entry and "b64" in entry
|
||||
assert browser_would_pass(bundle, entry)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("parsed", [True, False])
|
||||
def test_result_columns_named_like_a_masked_column_ship_only_their_id(parsed, monkeypatch):
|
||||
"""结果的列名按数据库交回的认（大小写不论）：SQL 里看不出来，列名对上遮罩列也只给哈希。这一条不靠按 SQL 认遮罩列的
|
||||
那一套（它认出的名字里本来就有遮罩列本身）：那一套什么都没认出来时照样拦得住。"""
|
||||
if not parsed:
|
||||
from app.engine import inquiry
|
||||
|
||||
monkeypatch.setattr(inquiry, "_derived_hidden", lambda *_args: (set(), set()))
|
||||
snap = snapshot("SELECT region, contact FROM member_view", ["region", "Phone"], [["华东", 508.5]])
|
||||
_, entry = bundle_of(snap, column="Phone")
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql", [
|
||||
"SELECT region, SUM(spend) AS spend FROM members GROUP BY region HAVING",
|
||||
"SELEC region, spend FROM members",
|
||||
"",
|
||||
])
|
||||
def test_sql_that_does_not_parse_ships_only_its_id(sql):
|
||||
bundle, entry = bundle_of(snapshot(sql, ["region", "spend"], [["华东", 508.5]]))
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
assert not shown(bundle, 2)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 改过的、认不得的、查询时就遮过的：照现有规则
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_a_changed_aggregate_snapshot_ships_only_its_id():
|
||||
"""服务端的快照文件被改过（哈希对不上）：里面的 SQL、列名都不可信，照旧只给 id、另标 mismatch。"""
|
||||
original = stored(REGIONS)[0]
|
||||
changed = stored(REGIONS)[1].replace("华东".encode(), "华北".encode())
|
||||
_, entry = bundle_of(REGIONS, raw=changed, artifact=original)
|
||||
assert entry == {"id": original, "kind": "query", "node_id": "members", "masked": True, "mismatch": True}
|
||||
|
||||
|
||||
def test_an_unreadable_snapshot_still_counts_as_masked():
|
||||
_, entry = bundle_of(REGIONS, raw=b"{not json")
|
||||
assert entry.get("masked") is True and entry.get("unparsable") is True and "b64" not in entry
|
||||
|
||||
|
||||
def test_a_snapshot_with_keys_the_bundle_does_not_know_ships_only_its_id():
|
||||
"""快照里有不认识的键（比如以后核查 agent 在查询时就遮过、另记下了遮了哪些列）：不知道它记着什么，照旧只给哈希。"""
|
||||
snap = {**REGIONS, "masked_at_query": ["phone"]}
|
||||
_, entry = bundle_of(snap)
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
|
||||
|
||||
@pytest.mark.parametrize("merged", [
|
||||
# 合并查询节点落的样子：另记着各输入和来历（inputs、sources 这几个键本身就不在认得的范围里）
|
||||
{"columns": ["region", "spend"], "rows": [["华东", 508.5]], "source": MERGE_SOURCE, "sources": ["crm"],
|
||||
"sql": AGGREGATE, "mask_columns": ["phone"]},
|
||||
# 只剩数据源查询工具也会落的键：单凭来源是合并查询就不认
|
||||
{"columns": ["region", "spend"], "rows": [["华东", 508.5]], "row_count": 1, "truncated": False,
|
||||
"source": MERGE_SOURCE, "sql": AGGREGATE, "mask_columns": ["phone"]},
|
||||
])
|
||||
def test_a_merged_snapshot_from_a_masked_source_ships_only_its_id(merged):
|
||||
"""合并查询的 SQL 查的是各输入的结果在内存里落成的表，遮罩要按各输入的来历算，这里不认：照旧只给哈希。"""
|
||||
_, entry = bundle_of(merged)
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
|
||||
|
||||
def test_an_error_while_judging_ships_only_the_id(monkeypatch):
|
||||
"""判断过程出错（解析器、认遮罩列的那一套抛了意外）：按碰了遮罩列算，只给哈希。"""
|
||||
from app.engine import inquiry
|
||||
|
||||
def broken(*_args, **_kwargs):
|
||||
raise RuntimeError("解析器坏了")
|
||||
|
||||
monkeypatch.setattr(inquiry, "_derived_hidden", broken)
|
||||
_, entry = bundle_of(REGIONS)
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
|
||||
|
||||
def test_derived_masked_columns_are_found_the_way_inquiry_finds_them(monkeypatch):
|
||||
"""按 SQL 认遮罩列用的是 inquiry._derived_hidden（双盲复算、质询同一套）：它认出结果列取自遮罩列，就只给哈希。"""
|
||||
from app.engine import inquiry
|
||||
|
||||
calls = []
|
||||
|
||||
def flags_spend(snap, hidden, kind):
|
||||
calls.append(kind)
|
||||
return set(hidden) | {"spend"}, set(hidden)
|
||||
|
||||
monkeypatch.setattr(inquiry, "_derived_hidden", flags_spend)
|
||||
_, entry = bundle_of(REGIONS)
|
||||
assert entry.get("masked") is True and "b64" not in entry
|
||||
assert calls
|
||||
@@ -15,9 +15,11 @@
|
||||
// - 深浅色、窄屏：390 宽不横向溢出，点开的数字面板在窄屏上从底部抽出，深色下底是深的、字是浅的；
|
||||
// - 样本页：tests/bundle_samples.py 用虚构数据真实生成的几份页面（遮罩快照、两份报告、早期运行、失败结束、先失败后
|
||||
// 经继续运行跑到成功的早期运行、两条出具
|
||||
// 判定、服务端缺文件、裸 NaN、别的运行的文档、口径卡有输入取自遮罩列），逐份断言各项结果和原因文案；在上面篡改 ④
|
||||
// 判定、服务端缺文件、裸 NaN、别的运行的文档、口径卡有输入取自遮罩列、设了遮罩的数据源上按地区汇总的查询），逐份
|
||||
// 断言各项结果和原因文案；在上面篡改 ④
|
||||
// 引用链（换文档、拿掉报告、多一份没人引用的工件）、⑥ 运行信息（运行结果、类别、出具档位）、拿掉被引用的口径卡、把
|
||||
// 报告文档标成遮罩、把遮罩卡标成服务端的文件与 id 不一致或解析不出来（服务端对被改过、解析不了的遮罩卡就这么标）、
|
||||
// 遮罩源上随文件附带的汇总快照改一个字节、它的数字改引用值并重算全部哈希、
|
||||
// 只改页面顶部的封存哈希；宽屏点开表格里右对齐的数字不被面板盖住；不执行脚本时封存哈希、运行 ID 照样看得到，状态写
|
||||
// 「尚未核对」；
|
||||
// - 记录页的导出按钮（预览用例 ?syn=bundle-provenance-*，page.route 假装导出接口）：已封存时可点、下载的文件名同服务端；
|
||||
@@ -145,6 +147,8 @@ const stateOf = (page) => page.evaluate(() => ({
|
||||
}))
|
||||
|
||||
const describe = (s) => Object.entries(s.checks).map(([k, v]) => `${k}=${v}`).join(' ')
|
||||
/** 查询快照只给哈希时核对页写的原因（verifier.js 的 T.maskedQuery） */
|
||||
const MASKED_QUERY = '这条查询用到或可能用到设了遮罩的列，快照未随文件附带，只能核对到哈希'
|
||||
/** 有通过、没有不通过：「部分无法核对」也算（遮罩快照、没有出处的数字是正常的无法核对） */
|
||||
const noFail = (r) => r === 'pass' || r === 'partial'
|
||||
const hasItem = (s, check, result, text) => s.items.some((i) => i.check === check && i.result === result
|
||||
@@ -489,16 +493,15 @@ await section('样本页：各种形状', async () => {
|
||||
check('生成样本页（tests/bundle_samples.py）', false, String(e?.message ?? e))
|
||||
return
|
||||
}
|
||||
check('生成样本页（tests/bundle_samples.py）', Object.keys(samples).length === 11, Object.keys(samples).join('、'))
|
||||
check('生成样本页（tests/bundle_samples.py）', Object.keys(samples).length === 12, Object.keys(samples).join('、'))
|
||||
|
||||
await sample('base', async (s, page) => {
|
||||
check('base：总览通过，有无法核对的项时写「可核对的项全部通过」', s.overall === 'pass' && s.text === '可核对的项全部通过',
|
||||
`${s.overall}：${s.text}`)
|
||||
check('base：①②④⑥ 通过，③⑤ 部分无法核对', ['events', 'manifest', 'chain', 'meta'].every((k) => s.checks[k] === 'pass')
|
||||
&& s.checks.artifacts === 'partial' && s.checks.numbers === 'partial', describe(s))
|
||||
check('base：遮罩快照在 ③ 无法核对，写明「含遮罩列，未随文件附带，只能核对到哈希」',
|
||||
hasItem(s, 'artifacts', 'na', '含遮罩列，未随文件附带，只能核对到哈希'))
|
||||
check('base：引用遮罩快照的数字无法核对，原因同上', hasItem(s, 'numbers', 'na', '含遮罩列，未随文件附带，只能核对到哈希'))
|
||||
check(`base：选出了遮罩列的快照在 ③ 无法核对，写明「${MASKED_QUERY}」`, hasItem(s, 'artifacts', 'na', MASKED_QUERY))
|
||||
check('base：引用这份快照的数字无法核对，原因同上', hasItem(s, 'numbers', 'na', MASKED_QUERY))
|
||||
check('base：没有出处的数字写明未使用引用标记', hasItem(s, 'numbers', 'na', '该数字未使用引用标记，无法核对其来源'))
|
||||
const outside = s.items.filter((i) => i.check === 'numbers' && !i.folded)
|
||||
check('base：⑤ 超过 20 项收起来时，无法核对的列在外面', outside.length === 2 && outside.every((i) => i.result === 'na')
|
||||
@@ -616,6 +619,35 @@ await section('样本页：各种形状', async () => {
|
||||
const html = await page.content()
|
||||
check('masked-card：页面里找不到卡上那个遮罩列的号码', !html.includes('000-0001') && !html.includes('000-0002'))
|
||||
})
|
||||
// 会员库设了号码遮罩（查询当时记下了，数据源现在也设着）：按地区汇总的查询没碰号码列，快照随文件附带、数字核对得过；
|
||||
// 选出了号码的那份只给哈希，原因写明这条查询用到或可能用到设了遮罩的列
|
||||
await sample('masked-source', async (s, page) => {
|
||||
const regions = s.items.find((i) => i.check === 'artifacts' && i.text.includes('会员地区汇总'))
|
||||
check('masked-source：按地区汇总的快照随文件附带，③ 通过', regions?.result === 'pass', regions?.text)
|
||||
const cited = s.items.filter((i) => i.check === 'numbers' && i.text.includes('查询 Q3'))
|
||||
check('masked-source：引用地区汇总的 2 个数字都核对通过', cited.length === 2 && cited.every((i) => i.result === 'pass'),
|
||||
cited.map((i) => `${i.result}：${i.text}`).join(' | '))
|
||||
const members = s.items.find((i) => i.check === 'artifacts' && i.text.includes('会员消费'))
|
||||
check('masked-source：选出了号码的快照只给哈希，写明这条查询用到或可能用到设了遮罩的列',
|
||||
members?.result === 'na' && members.text.includes(MASKED_QUERY), members?.text)
|
||||
const phoneCited = s.items.find((i) => i.check === 'numbers' && i.text.includes('查询 Q2'))
|
||||
check('masked-source：引用它的数字无法核对，原因同上', phoneCited?.result === 'na' && phoneCited.text.includes(MASKED_QUERY),
|
||||
phoneCited?.text)
|
||||
check('masked-source：总览通过（可核对的项全部通过），没有一项不通过', s.overall === 'pass'
|
||||
&& s.text === '可核对的项全部通过' && !s.items.some((i) => i.result === 'fail'), `${s.overall}：${s.text}`)
|
||||
// 点开地区汇总的第一个数字：出处面板里有这条查询的 SQL 和被引用的那一行。先看有没有核对通过的数字，没有就照实记一笔、
|
||||
// 不去点：点空了要等满时限才抛错，同一节后面的样本（连同下面这一项）就都不跑了
|
||||
const passed = page.locator('button.num[data-result="pass"]')
|
||||
let detail = '页面上没有核对通过的数字'
|
||||
if (await passed.count()) {
|
||||
await passed.first().click({ timeout: 5000 })
|
||||
detail = await appears(page.locator('#detail')) ? await page.locator('#detail').innerText() : '点开后没有出处面板'
|
||||
}
|
||||
check('masked-source：点开数字看得到汇总查询的 SQL 和被引用的那一行', detail.includes('SUM(spend)')
|
||||
&& detail.includes('快照第 1 行') && detail.includes('华东'), detail.replace(/\s+/g, ' ').slice(0, 120))
|
||||
const html = await page.content()
|
||||
check('masked-source：页面里找不到号码', !html.includes('000-0001') && !html.includes('000-0002'))
|
||||
})
|
||||
await sample('nan', async (s) => {
|
||||
check('nan：内容里有裸 NaN 的事件和工件照常通过', s.overall === 'pass' && s.checks.events === 'pass'
|
||||
&& !s.items.some((i) => i.result === 'fail'), describe(s))
|
||||
@@ -694,6 +726,13 @@ await section('样本页：篡改', async () => {
|
||||
&& metricNumber.text.includes('引用的工件没有通过核对'), metricNumber?.text],
|
||||
]
|
||||
})
|
||||
// 遮罩源上随文件附带的汇总快照照样逐字节核对：改一个字节 ③ 变红；改它的数字的引用值、哈希全部重算，⑤ 拿快照对出来变红
|
||||
await tamper(samples['masked-source'], '遮罩源上汇总快照的一个字节', (data) => {
|
||||
const art = data.artifacts.find((a) => a.kind === 'query' && typeof a.b64 === 'string')
|
||||
return art ? flipByte(art) : null
|
||||
}, turnsRed('artifacts', ['events', 'manifest']))
|
||||
await tamper(samples['masked-source'], '遮罩源上汇总查询的数字的引用值并重算全部哈希', forge,
|
||||
turnsRed('numbers', ['events', 'manifest', 'artifacts', 'chain']), { seal: true })
|
||||
// 服务端标了解析不出来、哈希又对得上的遮罩卡：照实说认不出是否含遮罩列，是正常的无法核对（同哈希一致却解析不了的
|
||||
// 嵌入工件），不说成「含遮罩列」
|
||||
await tamper(samples['masked-card'], '遮罩卡标成服务端解析不出来', (data) => {
|
||||
|
||||
Reference in New Issue
Block a user