mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 01:11:59 +00:00
K3b 复核修正（第 2 轮）：VALUES、行构造器里的 (表达式).* 和交回复合类型的表值函数整份遮；记着别的会话 SQL 的视图、不只读表结构的 SHOW 和 EXPLAIN 整份遮
- VALUES 里的 (表达式).*（基线就有）：Postgres 在 VALUES 里同样把复合类型、记录展开成好几列，LATERAL VALUES 按表值 函数认时一个元组项只算一列，column2 起的号码原样交给模型、落库。现在 (表达式).* 不只看选出项：树上任何位置的（VALUES、 行构造器、函数参数里的），以及 VALUES 里的 s.*、(s.*)，读到遮罩列就整份遮。(表达式).* 的值只按那个表达式认，不再把 * 当成展开 FROM 的来源（没设遮罩的复合类型 (p.address).* 以前因此被整列遮） - 同一类、复核没点到的：Postgres 里 FROM 中交回复合类型的表值函数。unnest 的数组元素是复合类型时展开成各个字段 （unnest(ARRAY[p.card]) 出 addr、tel，写了列名表也只改前几列的名字），不认得的函数只写了列名、没写类型的列名表 （jsonb_populate_record(NULL::contact, …) AS r(a) 后面的字段照原名出），读到遮罩列的整份遮（_spreads_record）；拆普通 数组的 unnest 也一样（元素类型 SQL 里看不出，多遮）。写了带类型列定义的照写的认 - Oracle 的 JSON 点号、对象类型的属性（t.profile.tel，解析器读成库、表、列）：前面几段也算列，以前只认 tel，号码原样交出 - 记着别的会话 SQL 原文、被锁那一行的值的来源补进 INDIRECT_NAMES（工具边界和可验证报告共用一份）：MySQL 的 INNODB_TRX、 OPTIMIZER_TRACE，MariaDB 的 INNODB_LOCKS、QUERY_CACHE_INFO，Oracle 的 V$ / GV$ / X$ 一律算、AWR 历史、SQL 调优集和 调优顾问、固定的执行计划和 SQL Profile、闪回事务，DBMS_XPLAN、DBMS_SQLTUNE 这类出报告的包 - 不是查询的语句（_reads_beyond）：只读表结构的 SHOW（表、列、索引、库、表状态、建库语句）、DESCRIBE / EXPLAIN 一张表 照常，别的 SHOW（ENGINE INNODB STATUS、BINLOG / RELAYLOG EVENTS、REPLICA STATUS、WARNINGS、PROFILES、CREATE TABLE / VIEW、TRIGGERS……）、EXPLAIN 一条语句（MySQL 把只有一行的表的值代进计划里的条件，8.0 实测）不逐个认，一律整份遮。 SHOW CREATE TABLE 不算只读表结构：对视图交回的是视图定义（WHERE 里写死的号码，8.0 实测），以前因为写着 TABLE x 本来 就整份遮，照旧；SHOW TABLE STATUS 以前同样因此整份遮，现在照常 - 质询模块说明的已知局限同步；补一条：视图、函数、约束定义里写死的字面量（information_schema.views、pg_views、 pg_get_viewdef()……）不认，照常交给 agent 补测试：VALUES × LATERAL / JOIN LATERAL / CTE / FROM / 匿名 ROW / 整行 / 表和子查询的 *，函数参数里的 (表达式).*， 没碰遮罩列的对照；交回复合类型的表值函数九种写法和三种照旧按名字认的对照；Oracle 点号四种和两种对照；十九种视图和包、 十三种 SHOW、九种 EXPLAIN、十二种只读表结构的对照；核查 agent 真实查询工具上的端到端（Postgres 的 VALUES 展开、MySQL 的 INNODB_TRX 和 SHOW ENGINE INNODB STATUS）。逐项去掉实现（(表达式).* 的取值、树上的 Dot、VALUES 里的 *、新名字、 _reads_beyond、只读表结构的豁免、SHOW CREATE TABLE、unnest、只写列名的列名表、三段名），对应的测试都变红。另在 Postgres 18、MySQL 8.0 临时实例上经核查 agent 的真实查询工具复跑复核的复现和上面这些写法（44 条），交给模型的、落库的、 通用工件接口给的都没有号码；169c500 上其中 23 条漏号码。
This commit is contained in:
ThearchyHelios
@@ -63,8 +63,13 @@
|
||||
IS NOT NULL THEN amount ELSE 0 END)）同写在 WHERE 里，不算
|
||||
- FROM、JOIN、LATERAL 里的表值函数（json_each、json_tree、unnest、jsonb_each_text、JSON_TABLE、LATERAL VALUES）拆开
|
||||
遮罩列的，按出列名认它出的列（常见函数的出列名、写的列名表、别名、函数名，_function_columns），同名的别的列一并遮
|
||||
（多遮）；不认得的函数又没写列名表的，整份快照都遮。函数参数里没写表名的 col->'path'（JSON 箭头，解析器当成以 col
|
||||
为参数的 lambda）按 col 认，真的 lambda（transform(arr, x -> …)）的参数和遮罩列同名时也算（多遮）
|
||||
（多遮）；不认得的函数又没写列名表的，整份快照都遮。Postgres 里可能交回复合类型、展开成各个字段的也整份遮
|
||||
（masking._spreads_record）：unnest（数组的元素是复合类型时展开成 addr、tel 这些字段，写了列名表也只改前几列的
|
||||
名字），不认得的函数只写了列名、没写类型的列名表（jsonb_populate_record(NULL::contact, …) AS r(a) 后面的字段
|
||||
照原名出）；拆的是普通数组的 unnest 也一样整份遮（元素是什么类型 SQL 里看不出，多遮）。函数参数里没写表名的
|
||||
col->'path'（JSON 箭头，解析器当成以 col 为参数的 lambda）按 col 认，真的 lambda（transform(arr, x -> …)）的参数
|
||||
和遮罩列同名时也算（多遮）。三段以上的名字（Oracle 的 JSON 点号、对象类型的属性 t.profile.tel，解析器读成库、表、列）
|
||||
前面几段也算列，写了库名、表名的列，库名、表名碰巧和遮罩列同名时也算（多遮）
|
||||
- * 把遮罩列带进别的列名（WITH t(a, b, …) AS (SELECT * …) 这类列名表、表的列名表 FROM sales AS s(i, r, p)、UNION 后一支
|
||||
的 *、TABLE x、标量子查询里的 *）：只展开一个来源、那里每一列都数得清的，按位置只遮取自遮罩列的那几列——子查询按它的
|
||||
选出项数，数据源里的表按查询当时冻结的表结构数（几张列一样的月度表 SELECT * 连起来汇总，金额照常）。表结构里查不到
|
||||
@@ -85,18 +90,26 @@
|
||||
按看不见算（当成数据源里的表，多遮）
|
||||
- 认不全的写法整份遮（多遮，同一份里没碰遮罩列的编号、金额也遮）：(表达式).* 展开读到遮罩列的（复合类型的列
|
||||
(p.profile).*、函数交回的记录 (jsonb_each_text(pj)).*、子查询里的 ROW (q.r).*，外面套了括号、起了别名的也算，展开成
|
||||
几列、各叫什么 SQL 里看不出）；语法树和实际执行的对不上的（MySQL、MariaDB 会执行的注释 /*! … */，Postgres 的
|
||||
几列、各叫什么 SQL 里看不出；不只选出项，VALUES 里的、行构造器和函数参数里的也算），VALUES 里的 s.*、(s.*) 读到遮罩列
|
||||
的（同样展开成好几列）；语法树和实际执行的对不上的（MySQL、MariaDB 会执行的注释 /*! … */，Postgres 的
|
||||
U&"…" 名字，没有 * 的一支选出项和结果的列数对不上）；表名、SQL 写在字符串里交给函数执行、整表导出的（query_to_xml、
|
||||
table_to_xml、cursor_to_xml、schema_to_xml、database_to_xml 这一族，dblink、crosstab、ts_stat，Oracle 的
|
||||
DBMS_XMLGEN、XMLTYPE）、按页按文件按日志读原始数据的（pageinspect、逻辑解码、LOAD_FILE）、统计信息和会话记录里存着值
|
||||
的（Postgres 的 pg_stats、pg_stat_statements、pg_stat_activity；SQLite 的 sqlite_stat4；MySQL 的 column_statistics、
|
||||
general_log、PROCESSLIST、performance_schema；MariaDB 的 column_stats；Oracle 的直方图、记着最小最大值的
|
||||
*_tab_columns、V$SQL 一族、审计记录），按名字认（masking 的 INDIRECT_FUNCTIONS、INDIRECT_NAMES，可验证报告用的同一份
|
||||
名单），去掉注释再认一遍（函数名和括号之间夹着注释），自己建的同类函数、视图认不出；information_schema、pg_catalog 里
|
||||
只有表结构的视图不算（核查 agent 查列名、类型常用）。只收支持的几种库的：DuckDB、SQL Server 的同类函数不收；分析出了
|
||||
意外、一长串 CTE 超出工作量上限（ANALYSIS_STEPS）的。整份遮的只收确定取自遮罩列的那几列的值当遮罩值（别的列的值
|
||||
本来就没交给模型，收了会把区域名、列名在同一次核查里处处换掉）；快照记下整份遮（masked_whole），格子引用指到它时照实
|
||||
说写法认不全
|
||||
DBMS_XMLGEN、XMLTYPE）、按页按文件按日志读原始数据的（pageinspect、逻辑解码、LOAD_FILE）、统计信息、会话记录、锁和
|
||||
日志里存着值的（Postgres 的 pg_stats、pg_stat_statements、pg_stat_activity；SQLite 的 sqlite_stat4；MySQL 的
|
||||
column_statistics、general_log、PROCESSLIST、INNODB_TRX、OPTIMIZER_TRACE、performance_schema；MariaDB 的
|
||||
column_stats、INNODB_LOCKS、QUERY_CACHE_INFO；Oracle 的直方图、记着最小最大值的 *_tab_columns、V$SQL 一族、
|
||||
SQL 调优集、固定下来的执行计划、闪回事务、LogMiner、审计记录），按名字认（masking 的 INDIRECT_FUNCTIONS、
|
||||
INDIRECT_NAMES，可验证报告用的同一份名单），去掉注释再认一遍（函数名和括号之间夹着注释），自己建的同类函数、视图
|
||||
认不出；information_schema、pg_catalog 里只有表结构的视图不算（核查 agent 查列名、类型常用）。不是查询的语句里，
|
||||
只读表结构的 SHOW（表、列、索引、库、表状态、建库语句）、DESCRIBE 一张表照常，别的 SHOW（ENGINE INNODB STATUS、
|
||||
BINLOG EVENTS、WARNINGS、CREATE TABLE / VIEW……）、EXPLAIN 一条语句（MySQL 把只有一行的表的值代进计划里的条件）
|
||||
不逐个认，一律整份遮。只收支持的几种库的：DuckDB、SQL Server 的同类函数不收；分析出了意外、一长串 CTE 超出工作量
|
||||
上限（ANALYSIS_STEPS）的。整份遮的只收确定取自遮罩列的那几列的值当遮罩值（别的列的值本来就没交给模型，收了会把区域
|
||||
名、列名在同一次核查里处处换掉）；快照记下整份遮（masked_whole），格子引用指到它时照实说写法认不全
|
||||
- 定义里写死的字面量不认：视图、函数、触发器、约束、默认值的定义经 information_schema（views、routines、triggers、
|
||||
columns 的默认值）、pg_views、pg_get_viewdef() 这类查出来照常交给 agent，写死在定义里的号码（排除测试账号的 WHERE）
|
||||
只在已经认得的遮罩值里时按值换掉；SHOW CREATE TABLE（对视图交回的是视图定义）、SHOW CREATE VIEW、SHOW TRIGGERS 不在
|
||||
只读表结构的几种里，跟着整份遮
|
||||
- 按遮罩列分档的标签（CASE、IIF 的结果全是写死的常量）：标签文字是 agent 写在 SQL 里的，不当遮罩值，SQL、回答正文里
|
||||
照原样；标签列当分组键（每一档一行）照常显示，分档写在子查询、WITH 里、外层（或中间一层）按它分组的也算
|
||||
（masking._grouped_labels），不分组的逐行标签列整列遮（格子里看不出某一笔落在哪一档），可 agent
|
||||
|
||||
+135
-35
@@ -14,22 +14,25 @@
|
||||
- 证据摘录（engine/inquiry.py 的 ExcerptMask，judge.prepare 按它逐段遮）：交给反方审稿、审批质询 agent 的报告摘录同一套
|
||||
口径，格子逐格遮（masked_cell），SQL 只换在找某一个值的字面量，阈值照原样
|
||||
- 哪些列（_derived_masks）：从遮罩列本身起，按 SQL 一层层认出值取自遮罩列的结果列（别名、函数、子查询、WITH、UNION 的
|
||||
另一支、和 * 写在同一支里没起别名的表达式、FROM 里拆开遮罩列的表值函数、函数参数里的 JSON 箭头 col->'path'）；* 把
|
||||
遮罩列带进别的列名的（写了列名表的派生来源、UNION 后一支的 *、标量子查询里的 *）按它展开落到的位置算（_star_places，
|
||||
展开数据源里的表的按查询当时冻结的表结构数位置，_schema_columns；写了库名的表按「库名.表名」查，另一个库里的同名表
|
||||
查不到）。整行和函数参数里的 *（Postgres 的 row_to_json(s)、SELECT s、to_json(s.*)，Oracle 的 JSON_OBJECT(*)）按 *
|
||||
展开的来源算；表的列名表（FROM sales AS s(i, r, p)）、TABLE x、套了括号的 * 先改写成认得出的样子（_normalized）。
|
||||
另一支、和 * 写在同一支里没起别名的表达式、FROM 里拆开遮罩列的表值函数、函数参数里的 JSON 箭头 col->'path'、Oracle
|
||||
的 JSON 点号 t.col.path）；* 把遮罩列带进别的列名的（写了列名表的派生来源、UNION 后一支的 *、标量子查询里的 *）按它
|
||||
展开落到的位置算（_star_places，展开数据源里的表的按查询当时冻结的表结构数位置，_schema_columns；写了库名的表按
|
||||
「库名.表名」查，另一个库里的同名表查不到）。整行和函数参数里的 *（Postgres 的 row_to_json(s)、SELECT s、
|
||||
to_json(s.*)，Oracle 的 JSON_OBJECT(*)）按 * 展开的来源算；表的列名表（FROM sales AS s(i, r, p)）、TABLE x、
|
||||
套了括号的 * 先改写成认得出的样子（_normalized）。
|
||||
WITH 里的名字按作用域认（_scoped_cte）：不带 RECURSIVE 的一段看不见它自己和排在后面的段，同名的是数据源里的表。
|
||||
COUNT 只数个数、窗口的分区键只决定次序、按遮罩列分档又当分组键或套在汇总里数个数的、汇总里的 CASE 只在条件里用到的
|
||||
不算（_value_columns、_band_only）。认不全的写法整份快照的列都算，不再逐个形状补规则：认不出落在哪一列的（PIVOT、
|
||||
UNPIVOT 这类也算），(表达式).* 展开读到遮罩列的（复合类型的列、函数交回的记录，外面套了括号、起了别名的也算，展开成
|
||||
几列、叫什么说不清），语法树和实际执行的对不上的（OPAQUE_SQL：MySQL 会执行的注释、U&"…"；没有 * 的一支选出项和结果
|
||||
的列数对不上），表名、SQL 写在字符串里交给函数、统计信息和会话记录里存着值的（INDIRECT_FUNCTIONS、INDIRECT_NAMES：
|
||||
query_to_xml 这一族、dblink、pg_stats、MySQL 的 column_statistics、Oracle 的直方图和 V$SQL……，可验证报告用的同一份），
|
||||
SQL 解析不了又写着遮罩列（或者 *、TABLE x、上面那几类，或者是 Postgres 的）的，分析出了意外、超出工作量上限
|
||||
（ANALYSIS_STEPS）的；这些只收确定取自遮罩列的那几列的值（_Masks.collect）。一次分析里派生来源对应的那块查询、一项
|
||||
的值取自哪些列记下来（_analysis），一长串 CTE 不再卡住。双盲复算按同一套认法在派盲算之前挡下取自遮罩列的指标
|
||||
（derived_columns）
|
||||
几列、叫什么说不清；不只选出项，VALUES、行构造器、函数参数里的也算；Postgres 里 FROM 中可能交回复合类型的表值函数
|
||||
同理，_spreads_record），语法树和实际执行的对不上的（OPAQUE_SQL：MySQL 会执行的注释、U&"…"；没有 * 的一支选出项
|
||||
和结果的列数对不上），表名、SQL 写在字符串里交给函数、统计信息和会话记录里存着值的（INDIRECT_FUNCTIONS、
|
||||
INDIRECT_NAMES：query_to_xml 这一族、dblink、pg_stats、MySQL 的 column_statistics、INNODB_TRX、Oracle 的直方图和
|
||||
V$SQL……，可验证报告用的同一份），不只读表结构的 SHOW、EXPLAIN 一条语句（_reads_beyond：引擎状态、日志、出错信息、
|
||||
视图定义里的原文，MySQL 把只有一行的表的值代进计划里的条件），SQL 解析不了又写着遮罩列（或者 *、TABLE x、上面那几类，
|
||||
或者是 Postgres 的）的，分析出了意外、超出工作量上限（ANALYSIS_STEPS）的；这些只收确定取自遮罩列的那几列的值
|
||||
（_Masks.collect）。一次分析里派生来源对应的那块查询、一项的值取自哪些列记下来（_analysis），一长串 CTE 不再卡住。
|
||||
双盲复算按同一套认法在派盲算之前挡下取自遮罩列的指标（derived_columns）
|
||||
- 只装着分档标签的列（_label_columns）：标签是 agent 写在 SQL 里的字，不当遮罩值；* 展开数据源里的表时按冻结的表结构
|
||||
认它带不带同名的列（_schema_tables），不知道表结构的按带得出算（多遮）。分档写在派生来源里、外层按它分组的（一档
|
||||
一行，_grouped_labels）同当分组键的 CASE 一样照常显示（_shown_labels）
|
||||
@@ -185,6 +188,17 @@ def _value_columns(node: Any) -> set[str]:
|
||||
elif isinstance(item, exp.Column):
|
||||
if item.name:
|
||||
out.add(item.name.lower())
|
||||
if item.args.get("db") is not None:
|
||||
# 三段以上的名字（t.profile.tel）：Oracle 的 JSON 点号、对象类型的属性写法里中间几段才是列（t 是表的别名），
|
||||
# 解析器读成「库 t、表 profile、列 tel」，只认 tel 时号码原样交了出去。前面几段都算（多遮：写了库名、表名
|
||||
# 的列，库名、表名碰巧和遮罩列同名时也算）
|
||||
out.update(part.name.lower() for key in ("catalog", "db", "table")
|
||||
if (part := item.args.get(key)) is not None and part.name)
|
||||
elif isinstance(item, exp.Dot) and isinstance(item.expression, exp.Star):
|
||||
# (表达式).*（VALUES、行构造器、函数参数里的）：展开的各个字段都取自那个表达式。这个 * 不是在展开 FROM 里的
|
||||
# 来源，不按 _scoped_star_inputs 算（以前算了，没设遮罩的复合类型 (p.address).* 也因为 FROM 里有表被当成
|
||||
# 取自遮罩列）。读到遮罩列的整份遮，见 _tainted_in
|
||||
stack.append(item.this)
|
||||
elif isinstance(item, exp.Count) or (isinstance(item, (exp.Case, exp.If)) and _band_only(item)):
|
||||
continue
|
||||
elif isinstance(item, (exp.Case, exp.If)) and (results := _branch_results(item)) is not None \
|
||||
@@ -880,6 +894,29 @@ def _function_columns(source: Any) -> set[str] | None:
|
||||
return names if listed or name in _TABLE_FUNCTIONS else None
|
||||
|
||||
|
||||
def _spreads_record(source: Any) -> bool:
|
||||
"""Postgres 里这个表值函数这类来源可能交回复合类型、展开成它的各个字段（字段名 SQL 里看不出，同 (表达式).*）：
|
||||
- unnest：数组的元素是复合类型时展开成各个字段（unnest(ARRAY[p.card]) 出 addr、tel），元素是什么类型 SQL 里看不出；
|
||||
- 不在 _TABLE_FUNCTIONS 里的函数写了只有名字、没写类型的列名表：交回复合类型的（jsonb_populate_record(NULL::contact,
|
||||
…)）列名表只改前几列的名字，后面的字段照原名出（Postgres 18 实测，unnest 的列名表也一样）。
|
||||
写了带类型的列定义（jsonb_to_record(…) AS r(tel text)）的出的就是这几列；VALUES、JSON_TABLE 的列写在 SQL 里。"""
|
||||
from sqlglot import exp
|
||||
|
||||
node = source.this if isinstance(source, exp.Lateral) else source
|
||||
alias = source.args.get("alias") or (node.args.get("alias") if node is not source else None)
|
||||
func = node.this if isinstance(node, exp.Table) else node
|
||||
while isinstance(func, exp.Subquery):
|
||||
func = func.this
|
||||
if isinstance(func, (exp.Values, exp.JSONTable)):
|
||||
return False
|
||||
name = (func.name if isinstance(func, exp.Anonymous) else func.sql_name() if isinstance(func, exp.Func)
|
||||
else "").lower()
|
||||
if name == "unnest":
|
||||
return True
|
||||
listed = list(alias.args.get("columns") or []) if alias is not None else []
|
||||
return bool(listed) and name not in _TABLE_FUNCTIONS and not all(isinstance(c, exp.ColumnDef) for c in listed)
|
||||
|
||||
|
||||
def _merging(select: Any, name: str) -> bool:
|
||||
"""FROM 里按同名的 name 这一列连起来（USING 里写着它，或者 NATURAL）：几个来源的这一列合成一列。"""
|
||||
return any(str(join.args.get("method") or "").upper() == "NATURAL"
|
||||
@@ -1173,7 +1210,8 @@ def _normalized(root: Any, read: str | None, tables: dict[str, list[str]] | None
|
||||
套了括号、起了别名的 ((p.profile).*)、(p.profile).* AS x 也算，_unwrapped）：展开成几列、各叫什么（复合类型的字段名、
|
||||
函数的出列名）SQL 里看不出，改成一个对不上任何来源的 *（_star_sources 为 None），只用来数别的项的位置。表达式的值
|
||||
取自哪些列（_value_columns，整行引用已经改写过）记下来交回：读到遮罩列的，整份快照都算（_tainted）——以前只改成对
|
||||
不上来源的 *，最外层、第一支、派生来源里没写列名表的 * 按「照原名带出」认，字段名和遮罩列对不上，号码原样交了出去
|
||||
不上来源的 *，最外层、第一支、派生来源里没写列名表的 * 按「照原名带出」认，字段名和遮罩列对不上，号码原样交了出去。
|
||||
选出项以外的（VALUES、行构造器、函数参数里的）不换，_tainted_in 按它的值取自哪些列认
|
||||
tables 是查询当时冻结的表结构（_schema_columns，按次序的列名），认整行引用时看来源出不出某一列；没有为 None。
|
||||
交回每个 (表达式).* 的值取自哪些列。"""
|
||||
from sqlglot import exp
|
||||
@@ -1458,16 +1496,17 @@ def _tainted(sql: str, kind: str | None, masked: set[str], width: int, tables: d
|
||||
后面的从末尾倒着数。认不出的——两个 * 之间的项、派生来源里没起别名的表达式（外层的 * 会把它原样带出来，列名
|
||||
又认不准）——整份快照的列都算（同解析不了）。
|
||||
FROM、JOIN、LATERAL 里的表值函数（json_each、unnest、JSON_TABLE、LATERAL VALUES）参数里写着这些名字的，它出的列
|
||||
都算（_function_columns）：遮罩列是 JSON 时，json_each 拆开是最自然的写法。出哪几列说不清的整份快照都算。
|
||||
都算（_function_columns）：遮罩列是 JSON 时，json_each 拆开是最自然的写法。出哪几列说不清的整份快照都算，Postgres
|
||||
里可能交回复合类型、展开成各个字段的（unnest、只写了列名的列名表，_spreads_record）也算。
|
||||
* 展开的列换了名字（写了列名表的派生来源、UNION 后一支的 *）：* 带出这些名字的值时，它展开落到的那几列照位置算
|
||||
（_star_places，展开数据源里的表的按 tables 数位置，数不清的整段算）；标量子查询里的 *、函数参数里的 *（to_json(s.*)、
|
||||
JSON_OBJECT(*)）、整行引用（row_to_json(s)）按 _value_columns 算进外层的表达式。表的列名表、TABLE x、套了括号的 *
|
||||
先改写成上面这些认得出的样子（_normalized）。
|
||||
PIVOT、UNPIVOT、MATCH_RECOGNIZE 读到这些名字的：值转进了它们自己起名的列（UNPIVOT 的值列、PIVOT 按 IN 里的值起名的列、
|
||||
MEASURES），出的是哪几列不好认，整份快照的列都算。(表达式).* 读到这些名字的、OPAQUE_SQL 和 _INDIRECT 那几类写法
|
||||
（_opaque）、结果的一支没有 * 而选出项和结果的列数对不上的（MySQL 会执行的注释里多写了几项这类，语法树和实际执行的
|
||||
对不上）同样整份算。解析之后的分析出了意外、超出工作量上限（ANALYSIS_STEPS）的也整份算——不退回「解析不了」只看
|
||||
文字：没写遮罩列、没有 * 的整行引用（SELECT s）那样就放过去了。"""
|
||||
MEASURES），出的是哪几列不好认，整份快照的列都算。(表达式).* 读到这些名字的（树上任何位置的，VALUES 里的也算）、
|
||||
OPAQUE_SQL 和 _INDIRECT 那几类写法（_opaque）、结果的一支没有 * 而选出项和结果的列数对不上的（MySQL 会执行的注释里
|
||||
多写了几项这类，语法树和实际执行的对不上）同样整份算。解析之后的分析出了意外、超出工作量上限（ANALYSIS_STEPS）的
|
||||
也整份算——不退回「解析不了」只看文字：没写遮罩列、没有 * 的整行引用（SELECT s）那样就放过去了。"""
|
||||
from app.data.sql_semantics import _parse, _read
|
||||
|
||||
try:
|
||||
@@ -1493,11 +1532,16 @@ def _tainted_in(root: Any, sql: str, kind: str | None, masked: set[str], width:
|
||||
"""_tainted 解析、改写（_normalized）之后真去认的那一步。opaque 是 (表达式).* 的值取自哪些列（_normalized 交回的）。"""
|
||||
from sqlglot import exp
|
||||
|
||||
from app.data.sql_semantics import _read
|
||||
|
||||
names = set(masked)
|
||||
selects = list(root.find_all(exp.Select))
|
||||
blocks = _blocks(root)
|
||||
sources = [(top, listed) for top, role, listed in blocks if role == "source"]
|
||||
functions = [source for select_ in selects for source in _function_sources(select_)]
|
||||
# Postgres 里可能交回复合类型、展开成各个字段的表值函数（_spreads_record）：读到遮罩列时出哪几列说不清，整份遮
|
||||
spreading = {id(source) for source in functions if _spreads_record(source)} \
|
||||
if _read(kind) in _ROW_DIALECTS else set()
|
||||
unreadable = False
|
||||
for _ in range(len(selects) + len(sources) + len(functions) + 1):
|
||||
before = len(names)
|
||||
@@ -1510,17 +1554,24 @@ def _tainted_in(root: Any, sql: str, kind: str | None, masked: set[str], width:
|
||||
names |= _source_names(top, listed, names, tables)[0]
|
||||
for source in functions:
|
||||
if _reads(names, _value_columns(source)):
|
||||
columns = _function_columns(source)
|
||||
columns = None if id(source) in spreading else _function_columns(source)
|
||||
unreadable = unreadable or columns is None
|
||||
names |= columns or set()
|
||||
if len(names) == before:
|
||||
break
|
||||
# 认不出的整份遮：表值函数出哪几列说不清、派生来源里有认不出名字的列、PIVOT 这类读到遮罩列，(表达式).* 读到遮罩列
|
||||
# （展开成几列、叫什么都说不清），语法树和实际执行的对不上、SQL 写在字符串里交给函数执行、整表导出、统计信息里的值
|
||||
# （_opaque）
|
||||
# （_opaque）。(表达式).* 不只看选出项（那几个 _normalized 已经换掉，值取自哪些列在 opaque 里）：VALUES 里同样展开成
|
||||
# 好几列（Postgres 文档：选出列表、VALUES、行构造器里的 composite.* 都展开），VALUES 里的一项是 s.*、(s.*) 的也一样；
|
||||
# LATERAL VALUES 按表值函数认时一个元组项只算一列，column2 起的号码原样交了出去。不逐个位置认：(表达式).* 在树上任何
|
||||
# 位置（行构造器、函数参数里的不展开成几列，也一样，多遮）、VALUES 里的 *，读到遮罩列就整份遮
|
||||
expanded = [node.this for node in root.find_all(exp.Dot) if isinstance(node.expression, exp.Star)]
|
||||
expanded += [item for values in root.find_all(exp.Values) for row in values.expressions
|
||||
if isinstance(row, exp.Tuple) for item in row.expressions if _is_star(_unwrapped(item))]
|
||||
everything = unreadable or any(_source_names(top, listed, names, tables)[1] for top, listed in sources) \
|
||||
or any(_reads(names, _value_columns(node)) for node in root.find_all(exp.Pivot, exp.MatchRecognize)) \
|
||||
or any(_reads(names, cols) for cols in opaque) or _opaque(sql, kind)
|
||||
or any(_reads(names, cols) for cols in opaque) \
|
||||
or any(_reads(names, _value_columns(node)) for node in expanded) or _opaque(sql, kind)
|
||||
positions: set[int] = set()
|
||||
for top, role, _listed in blocks:
|
||||
if role != "result":
|
||||
@@ -1580,26 +1631,38 @@ INDIRECT_FUNCTIONS = (
|
||||
r"pg_stat_get_(?:backend_)?activity", r"audit_log_read",
|
||||
)
|
||||
#: 不点列名也读得出表里内容的表、视图、包（按名字认）：
|
||||
#: - Oracle 的 DBMS_XMLGEN 这一族（包名后面跟着点），XQuery 里的 ora:view 和 oradb: 表
|
||||
#: - Oracle 的 DBMS_XMLGEN 这一族（包名后面跟着点），XQuery 里的 ora:view 和 oradb: 表；读游标、出报告的包：DBMS_XPLAN 的
|
||||
#: DISPLAY_CURSOR（带 PEEKED_BINDS 时有绑定变量的值）、DBMS_SQLTUNE、DBMS_SQL_MONITOR 的报告、DBMS_WORKLOAD_REPOSITORY
|
||||
#: 的 AWR 报告（排在前面的语句原文）
|
||||
#: - 统计信息里的采样值、全文索引里的词：Postgres 的 pg_stats、pg_statistic 一族，SQLite 的 sqlite_stat*，MySQL 的
|
||||
#: column_statistics（直方图里是列的原值）、innodb_ft_*，MariaDB 的 column_stats，Oracle 的直方图、列统计和记着最小最大值
|
||||
#: 的 *_tab_columns、*_tab_cols（LOW_VALUE、HIGH_VALUE 按字节交回来就是号码、姓名的原文），以及它们的底表 histgrm$、
|
||||
#: hist_head$
|
||||
#: - 别的会话执行过的 SQL 原文：Postgres 的 pg_stat_activity、pg_stat_statements、本连接预备过的语句；MySQL 的 general_log、
|
||||
#: slow_log、PROCESSLIST、performance_schema（语句历史 events_statements_*、线程正在执行的 SQL），sys 库里建在
|
||||
#: performance_schema 上的会话、锁等待视图（带 x$ 的是不排版的那一份；只认这几张，Oracle 常写的 sys.dual 不算）；Oracle 的
|
||||
#: V$SQL 一族（含 v$sql_bind_capture）、打开着的游标、审计记录
|
||||
#: slow_log、PROCESSLIST、正在执行的事务（information_schema.INNODB_TRX 的 trx_query）、performance_schema（语句历史
|
||||
#: events_statements_*、线程正在执行的 SQL），sys 库里建在 performance_schema 上的会话、锁等待视图（只认这几张，Oracle
|
||||
#: 常写的 sys.dual 不算；带 x$ 的不排版的那一份按下面的 X$ 算）；MariaDB 的查询缓存（QUERY_CACHE_INFO）；Oracle 的动态视图
|
||||
#: V$ / GV$ / X$ 一律算，不逐个认（V$SQL 一族、v$sql_bind_capture、打开着的游标、LogMiner 的 V$LOGMNR_CONTENTS、XML 的
|
||||
#: 审计记录都在里面），AWR 历史 DBA_HIST_*、审计记录，SQL 调优集里的语句和绑定变量（*_sqlset_*）、调优顾问记下的
|
||||
#: （*_sqltune_*）、SQL Access Advisor 的工作负载（*_advisor_sqlw_*）、固定下来的执行计划和 SQL Profile、Patch
|
||||
#: - 被锁那一行、改动前后那一行的值：MariaDB（和 MySQL 5.7）的 INNODB_LOCKS 的 lock_data 是被锁那一行的键值；Oracle 的
|
||||
#: 闪回事务（FLASHBACK_TRANSACTION_QUERY 的 UNDO_SQL）、LogMiner（V$LOGMNR_CONTENTS 的 SQL_REDO / SQL_UNDO）写着整行的值
|
||||
#: - MySQL、MariaDB 的 OPTIMIZER_TRACE：同 EXPLAIN 一条语句（_reads_beyond），优化器把只有一行的表的值代进条件写在里面
|
||||
#: （要开着优化器跟踪才有内容）
|
||||
#: - SQLite 按页读出的原始数据
|
||||
#: information_schema、pg_catalog 不整个收：大多只有表结构，核查 agent 查列名、类型时常用，里面存着值的那几张上面逐个列了
|
||||
#: （可验证报告那边宁可多挡，整个收，见 SYSTEM_SCHEMAS）
|
||||
#: （可验证报告那边宁可多挡，整个收，见 SYSTEM_SCHEMAS）。SHOW、EXPLAIN 这类不是查询的语句另按 _reads_beyond 认
|
||||
INDIRECT_NAMES = (
|
||||
r"dbms_xml\w*", r"ora:view", r"oradb",
|
||||
r"dbms_xml\w*", r"ora:view", r"oradb", r"dbms_(?:xplan|sqltune|sql_monitor|workload_repository)",
|
||||
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)\$",
|
||||
r"pg_stat_activity", r"pg_stat_statements", r"pg_prepared_statements", r"general_log", r"slow_log",
|
||||
r"processlist", r"performance_schema", r"events_statements_\w+",
|
||||
r"processlist", r"innodb_trx", r"innodb_locks", r"query_cache_info", r"optimizer_trace", r"performance_schema",
|
||||
r"events_statements_\w+",
|
||||
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"(?:g?v|x)\$\w+", r"(?:dba|cdb)_hist_\w+",
|
||||
r"(?:all|dba|user|cdb)_(?:sqlset|sqltune|advisor_sqlw)_\w+", r"(?:dba|cdb)_sql_(?:plan_baselines|profiles|patches)",
|
||||
r"flashback_transaction_query",
|
||||
r"(?:dba|cdb|user)_(?:\w+_)?audit_trail", r"unified_audit_trail", r"aud\$\w*", r"fga_log\$",
|
||||
r"sqlite_dbpage",
|
||||
)
|
||||
@@ -1620,10 +1683,43 @@ def indirect_reads(*names: str) -> re.Pattern[str]:
|
||||
_INDIRECT = indirect_reads()
|
||||
|
||||
|
||||
#: 守卫放行的、不是查询的语句里只读得出表结构、库清单的几种（MySQL、MariaDB）：SHOW 表、列、索引、库、表状态、建库语句，
|
||||
#: DESCRIBE / EXPLAIN 一张表（看它的列，后面可以跟一个列名或通配的写法）。核查 agent 探查表结构靠它们，照常。
|
||||
#: SHOW CREATE TABLE 不在里面：对视图交回的是视图定义（MySQL 8.0 实测，同 SHOW CREATE VIEW，WHERE 里写死的号码原样），
|
||||
#: 建表语句里还有默认值、CHECK 约束、分区的取值；以前因为写着 TABLE x 就整份遮，现在照旧
|
||||
_STRUCTURE_ONLY = re.compile(
|
||||
r"show\s+(?:(?:extended|full)\s+)*(?:tables|columns|fields|index|indexes|keys|databases|schemas|table\s+status"
|
||||
r"|create\s+(?:database|schema))\b"
|
||||
r"|(?:explain|describe|desc)\s+(?!(?:select|with|table|values|analyze|format|for|extended|partitions|plan|query"
|
||||
r"|verbose)\b)[\w$`\".]+(?:\s+[\w$`\"'%.]+)?\s*;?\s*$",
|
||||
re.I)
|
||||
#: 不是查询、读得出别的东西的语句：SHOW、EXPLAIN、DESCRIBE（EXPLAIN 一条语句也写作 DESCRIBE）
|
||||
_STATEMENT_VERBS = frozenset({"show", "explain", "describe", "desc"})
|
||||
|
||||
|
||||
def _reads_beyond(bare: str, kind: str | None) -> bool:
|
||||
"""去掉注释的 SQL（bare）是 SHOW、EXPLAIN 这类不是查询的语句，读的又不只是表结构（_STRUCTURE_ONLY 以外的）：
|
||||
- SHOW ENGINE INNODB STATUS（最近的死锁、正在执行的事务带着别的会话的 SQL）、BINLOG / RELAYLOG EVENTS（语句格式的
|
||||
日志就是 SQL 原文）、REPLICA STATUS 的出错信息、PROFILES、WARNINGS（EXPLAIN 之后改写过的语句里，只有一行的表的
|
||||
每一列都代成了常量）、视图和触发器的定义、建表语句（对视图交回的是视图定义）、MariaDB 的 SHOW EXPLAIN FOR 别的
|
||||
连接……交回的是原文，可能带着号码；
|
||||
- EXPLAIN 一条语句：MySQL 优化时把只有一行的表（按主键取一行）读出来、把值当常量代进条件，FORMAT=TREE / JSON、
|
||||
EXPLAIN ANALYZE 写出的条件里就是那一行的值（MySQL 8.0 实测）。FOR CONNECTION 看别的连接正在执行的语句，也算。
|
||||
守卫放行所有 SHOW、EXPLAIN（只读），这里不逐个认哪一种带着值：只读表结构的几种以外一律整份遮。Postgres 解析不了的
|
||||
本来就整份遮（_derived_masks），SQLite 的 EXPLAIN 只有字节码和语句里写的字面量，照常。首关键字的认法同守卫的
|
||||
first_verb（括号开头的也认）。"""
|
||||
text = bare.strip()
|
||||
verb = re.match(r"[A-Za-z_]+", text.lstrip("(").lstrip())
|
||||
if verb is None or verb.group(0).lower() not in _STATEMENT_VERBS or (kind or "").lower() == "sqlite":
|
||||
return False
|
||||
return not _STRUCTURE_ONLY.match(text)
|
||||
|
||||
|
||||
def _opaque(sql: str, kind: str | None) -> bool:
|
||||
"""SQL 里有认不全的写法：OPAQUE_SQL（MySQL 会执行的注释、U&"…"），或者 _INDIRECT 那几类。按原文认（字符串、注释里
|
||||
写着的也算，多遮），再按去掉注释的认：函数名和括号之间夹着注释（query_to_xml/**/(…)、query_to_xml --换行(…)）解析得了，
|
||||
语法树上又不认函数名，原文里名字后面不紧跟着括号，以前就放过去了。去注释照守卫按数据源的方言切（data/guard.py）。"""
|
||||
语法树上又不认函数名，原文里名字后面不紧跟着括号，以前就放过去了。去注释照守卫按数据源的方言切（data/guard.py）。
|
||||
不只读表结构的 SHOW、EXPLAIN 一条语句也算（_reads_beyond）。"""
|
||||
if OPAQUE_SQL.search(sql) or _INDIRECT.search(sql):
|
||||
return True
|
||||
from app.data.guard import strip_comments
|
||||
@@ -1632,13 +1728,16 @@ def _opaque(sql: str, kind: str | None) -> bool:
|
||||
bare = strip_comments(sql, dialect=kind)
|
||||
except Exception: # noqa: BLE001 - 守卫切不开的写法：认不准，按认不全算（宁可多遮）
|
||||
return True
|
||||
return bare != sql and bool(_INDIRECT.search(bare))
|
||||
return (bare != sql and bool(_INDIRECT.search(bare))) or _reads_beyond(bare, kind)
|
||||
|
||||
|
||||
def _implicit(sql: str, kind: str | None = None) -> bool:
|
||||
"""解析不了的 SQL 里有没有不写列名也会带出遮罩列的写法（*、TABLE x，OPAQUE_SQL、_INDIRECT 那几类，_opaque）：它们能
|
||||
把遮罩列带进别的列名（列名表、UNION 的后一支）、整表交出来，同写着遮罩列一样整份遮（_derived_hidden）。只看文字，
|
||||
字符串、注释里的也算（多遮）。"""
|
||||
字符串、注释里的也算（多遮）。只读表结构的 SHOW、DESCRIBE（_STRUCTURE_ONLY）不按 *、TABLE x 认：SHOW TABLE STATUS
|
||||
里的 TABLE 不是 TABLE x 语句，以前因此整份遮，表状态看不到（SHOW CREATE TABLE 照旧整份遮，见 _STRUCTURE_ONLY）。"""
|
||||
if _STRUCTURE_ONLY.match(sql.strip()):
|
||||
return _opaque(sql, kind)
|
||||
return bool(_IMPLICIT.search(_COUNT_STAR.sub("", sql))) or _opaque(sql, kind)
|
||||
|
||||
|
||||
@@ -2607,9 +2706,10 @@ def mask_query_result(payload: dict[str, Any], masked: Iterable[str], kind: str
|
||||
- 整列遮：列名直接对上遮罩列的，加上按 SQL 认出值取自遮罩列的（_derived_masks：别名、函数、子查询、WITH、
|
||||
SELECT *（展开的列换了名字的也算）、表值函数、整行引用和函数参数里的 *、列名表、TABLE x；COUNT、只在条件里用到、
|
||||
分档当分组键的不算，分档写在派生来源里、外层按它分组的也不算）；认不全的（SQL 解析不了又写着遮罩列、*、TABLE x，
|
||||
或者是 Postgres 的；(表达式).* 读到遮罩列，MySQL 会执行的注释、U&"…"，表名、SQL 写在字符串里交给函数，统计信息、
|
||||
会话记录里存着值的视图，选出项和结果的列数对不上，分析出意外、超出工作量上限）整份结果的列都遮（QueryMask.whole）。
|
||||
这几列的格子换成「已遮罩」，类型记成文本
|
||||
或者是 Postgres 的；(表达式).* 读到遮罩列（VALUES 里的也算）、Postgres 里可能交回复合类型的表值函数读到遮罩列，
|
||||
MySQL 会执行的注释、U&"…"，表名、SQL 写在字符串里交给函数，统计信息、会话记录里存着值的视图，不只读表结构的
|
||||
SHOW、EXPLAIN 一条语句，选出项和结果的列数对不上，分析出意外、超出工作量上限）整份结果的列都遮
|
||||
（QueryMask.whole）。这几列的格子换成「已遮罩」，类型记成文本
|
||||
- 别的列按值换，只换文字格：整格就是遮罩值的换掉，文字格里夹着的那一截换掉（换了写法的号码也认，
|
||||
_Secrets.loose_spans），换过的格子记下来（QueryMask.cells）。遮罩值是这份结果里的加上 known：先查号码、再查写着
|
||||
号码的备注、没设遮罩的联系方式表（别的库里的也一样）时，后一份同样遮好。门槛同按值遮（SECRET_DIGITS、SECRET_MIN）；
|
||||
|
||||
@@ -146,6 +146,27 @@ def test_a_json_arrow_over_an_unmasked_column_stays_visible():
|
||||
assert hidden_of("SELECT transform(scores, x -> x + 1) AS 分 FROM sales", ["分"], "postgres") == MASKED
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, shown", [
|
||||
pytest.param("SELECT t.profile.tel AS 电话 FROM sales t", ["电话"], ["电话"], id="JSON 点号"),
|
||||
pytest.param("SELECT t.profile.contact.tel FROM sales t", ["tel"], ["tel"], id="JSON 点号两层"),
|
||||
pytest.param("SELECT UPPER(t.profile.tel) AS 电话, t.region FROM sales t", ["电话", "region"], ["电话"],
|
||||
id="套函数"),
|
||||
pytest.param("SELECT x.电话 FROM (SELECT t.profile.tel AS 电话 FROM sales t) x", ["电话"], ["电话"],
|
||||
id="子查询里取出再选"),
|
||||
])
|
||||
def test_oracle_dot_notation_into_a_masked_json_column_is_read_as_that_column(sql, columns, shown):
|
||||
"""Oracle 的 JSON 点号、对象类型的属性写法（t.profile.tel）：解析器读成「库 t、表 profile、列 tel」，以前只认 tel，
|
||||
取出来的号码原样交给模型、落库。中间几段也算列（多遮：写了库名、表名的列，表名碰巧和遮罩列同名时也算）。"""
|
||||
hidden = hidden_of(sql, columns, "oracle")
|
||||
assert {c.lower() for c in shown} <= hidden and "region" not in hidden
|
||||
|
||||
|
||||
def test_oracle_dot_notation_into_an_unmasked_column_stays_visible():
|
||||
"""点号取的不是遮罩列、写了库名表名的普通列：一列都不遮。"""
|
||||
assert hidden_of("SELECT t.tags.name AS 标签 FROM sales t", ["标签"], "oracle") == MASKED
|
||||
assert hidden_of("SELECT shop.sales.region FROM shop.sales", ["region"], "oracle") == MASKED
|
||||
|
||||
|
||||
def test_a_json_arrow_inside_json_each_is_masked_when_really_run():
|
||||
"""SQLite 里真跑一遍：json_each(profile->'$.tels') 拆出来的号码在交给模型之前就遮掉。"""
|
||||
import sqlite3
|
||||
@@ -159,3 +180,72 @@ def test_a_json_arrow_inside_json_each_is_masked_when_really_run():
|
||||
db.close()
|
||||
done = masking.mask_query_result(payload, ["profile"], "sqlite")
|
||||
assert done.columns == ["v"] and done.payload["rows"] == [["已遮罩"], ["已遮罩"]]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 交回复合类型的表值函数（K3b）：Postgres 把复合类型展开成它的各个字段
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
#: people3.card 是设了遮罩的复合类型（addr, tel），people2.pj 是设了遮罩的 jsonb（照着 contact 类型拆），sales 是订单
|
||||
COMPOSITE_SCHEMA = {"tables": {
|
||||
"people3": {"qualified": "people3", "schema": None, "columns": [{"name": c} for c in ["id", "card", "city"]]},
|
||||
"people2": {"qualified": "people2", "schema": None, "columns": [{"name": c} for c in ["id", "pj"]]},
|
||||
"sales": {"qualified": "sales", "schema": None, "columns": [{"name": c} for c in SALES]},
|
||||
}}
|
||||
COMPOSITE_MASKED = ["card", "pj", "buyer_phone"]
|
||||
|
||||
|
||||
def _boundary(sql: str, columns: list[str], row: list) -> masking.QueryMask:
|
||||
payload = {"columns": columns, "rows": [list(row)], "sql": sql}
|
||||
return masking.mask_query_result(payload, COMPOSITE_MASKED, "postgresql", schema=COMPOSITE_SCHEMA)
|
||||
|
||||
|
||||
#: (SQL, 结果列, 一行)：结果的样子照 Postgres 18 实测。unnest 的数组元素是复合类型时，FROM 里的 unnest 展开成复合类型的
|
||||
#: 各个字段（addr、tel），写了列名表也只改前几列的名字；交回复合类型的函数（jsonb_populate_record(NULL::contact, …)）
|
||||
#: 写了只有名字、没写类型的列名表时同样只改前几列，后面的字段照原名出
|
||||
COMPOSITE_RESULTS = [
|
||||
pytest.param("SELECT v.* FROM people3 p, unnest(ARRAY[p.card]) v", ["addr", "tel"], ["某路 1 号", "13900000001"],
|
||||
id="unnest 复合类型的数组"),
|
||||
pytest.param("SELECT * FROM unnest(ARRAY[(SELECT card FROM people3 WHERE id = 1)])", ["addr", "tel"],
|
||||
["某路 1 号", "13900000001"], id="参数是子查询"),
|
||||
pytest.param("SELECT v.tel FROM people3 p, LATERAL unnest(ARRAY[p.card]) v", ["tel"], ["13900000001"],
|
||||
id="LATERAL 取一个字段"),
|
||||
pytest.param("SELECT * FROM people3 p, unnest(ARRAY[p.card]) AS u(a)", ["id", "card", "city", "a", "tel"],
|
||||
[1, "(某路 1 号,13900000001)", "杭州", "某路 1 号", "13900000001"], id="列名表比字段少"),
|
||||
pytest.param("SELECT u.* FROM people3 p, unnest(ARRAY[p.card], ARRAY[p.city]) AS u", ["addr", "tel", "unnest"],
|
||||
["某路 1 号", "13900000001", "杭州"], id="unnest 几个数组"),
|
||||
pytest.param("SELECT u.* FROM people3 p, unnest(ARRAY[p.card]) WITH ORDINALITY AS u(a)",
|
||||
["a", "tel", "ordinality"], ["某路 1 号", "13900000001", 1], id="WITH ORDINALITY"),
|
||||
pytest.param("SELECT * FROM (SELECT p.card FROM people3 p) s, LATERAL unnest(ARRAY[s.card]) WITH ORDINALITY u",
|
||||
["card", "addr", "tel", "ordinality"], ["(某路 1 号,13900000001)", "某路 1 号", "13900000001", 1],
|
||||
id="参数取自派生来源"),
|
||||
pytest.param("SELECT r.* FROM people2 p, jsonb_populate_record(NULL::contact, p.pj) AS r(a)", ["a", "tel"],
|
||||
["丙路", "13900000021"], id="populate_record 列名表比字段少"),
|
||||
pytest.param("SELECT r.* FROM people2 p, LATERAL jsonb_populate_recordset(NULL::contact, jsonb_build_array(p.pj)) "
|
||||
"AS r(a)", ["a", "tel"], ["丙路", "13900000021"], id="populate_recordset 列名表比字段少"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, row", COMPOSITE_RESULTS)
|
||||
def test_a_table_function_spreading_a_composite_masks_the_whole_result(sql, columns, row):
|
||||
"""FROM 里的函数交回复合类型、读到遮罩列：展开成哪几个字段、各叫什么 SQL 里看不出（同 (表达式).*），整份遮。以前
|
||||
unnest 按「出一列，名字是别名或函数名」认，列名表按写全了认，tel 这一列的号码原样交给模型、落库。"""
|
||||
done = _boundary(sql, columns, row)
|
||||
assert done.whole and done.columns == columns
|
||||
assert done.payload["rows"] == [["已遮罩"] * len(columns)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, row, hidden", [
|
||||
pytest.param("SELECT r.* FROM people2 p, jsonb_to_record(p.pj) AS r(tel text, addr text)", ["tel", "addr"],
|
||||
["13900000021", "丙路"], ["tel", "addr"], id="写了带类型的列定义"),
|
||||
pytest.param("SELECT k, v FROM people2 p, jsonb_each_text(p.pj) AS x(k, v)", ["k", "v"], ["tel", "13900000021"],
|
||||
["k", "v"], id="出列认得全的函数写了列名表"),
|
||||
pytest.param("SELECT s.id, u FROM sales s, unnest(ARRAY[s.region]) AS u", ["id", "u"], [1, "华东"], [],
|
||||
id="unnest 没碰遮罩列"),
|
||||
])
|
||||
def test_a_table_function_whose_columns_are_spelled_out_is_read_as_before(sql, columns, row, hidden):
|
||||
"""列写全了的（带类型的列定义、出列认得全的函数）照旧按名字认，没碰遮罩列的照常。"""
|
||||
done = _boundary(sql, columns, row)
|
||||
assert not done.whole and done.columns == hidden
|
||||
assert done.payload["rows"] == [["已遮罩" if c in hidden else v for c, v in zip(columns, row)]]
|
||||
|
||||
@@ -0,0 +1,308 @@
|
||||
"""认不全的写法整份遮，K3b 复核第 2 轮补的两类（engine/masking.py 的 _tainted_in、_opaque）：
|
||||
|
||||
- VALUES 里的 (表达式).*：Postgres 在 VALUES 里同样把复合类型的值、记录展开成好几列（文档：选出列表、VALUES、行构造器
|
||||
里的 composite.* 都展开），LATERAL VALUES 按表值函数认时一个元组项只算一列，column2 起的号码原样交给模型、落库。现在
|
||||
树上任何位置的 (表达式).* 读到遮罩列，整份结果都遮——LATERAL、JOIN LATERAL、CTE、FROM 里的 VALUES 都一样
|
||||
- 记着别的会话的 SQL 原文、被锁那一行的值的来源：MySQL 的 information_schema.INNODB_TRX、OPTIMIZER_TRACE，MariaDB 的
|
||||
INNODB_LOCKS、QUERY_CACHE_INFO，Oracle 的 SQL 调优集、固定下来的执行计划、闪回事务、LogMiner；不只读表结构的 SHOW
|
||||
（ENGINE INNODB STATUS、BINLOG / RELAYLOG EVENTS、REPLICA STATUS、WARNINGS、CREATE TABLE / VIEW……）和 EXPLAIN 一条
|
||||
语句。以前工具边界的名单里没有，可验证报告那份整个挡 information_schema、不附带解析不了的快照，反而挡得住
|
||||
|
||||
这里没有真的数据库：结果按它交回的样子造（复合类型展开成几列、SHOW 交回的几列，照 Postgres 18、MySQL 8.0 实测的样子）。
|
||||
数据都是虚构的：手机号、地址、区域。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from app.core import artifact_store
|
||||
from app.data.engine import engines
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource
|
||||
from app.engine import masking
|
||||
from app.main import app
|
||||
from tests.test_verify_agent import Script, q
|
||||
from tests.test_verify_agent_masking import ask, clean_spend # noqa: F401 - clean_spend 是自动用上的夹具
|
||||
|
||||
MASKED_WORD = "已遮罩"
|
||||
#: people3.card 是复合类型（addr, tel），people.address 是没设遮罩的复合类型（street, city），sales 是订单。设了遮罩的是
|
||||
#: card、buyer_phone
|
||||
SCHEMA = {"tables": {
|
||||
"people3": {"qualified": "people3", "schema": None, "columns": [{"name": c} for c in ["id", "card", "city"]]},
|
||||
"people": {"qualified": "people", "schema": None, "columns": [{"name": c} for c in ["id", "address"]]},
|
||||
"sales": {"qualified": "sales", "schema": None,
|
||||
"columns": [{"name": c} for c in ["id", "region", "buyer_phone", "amount"]]},
|
||||
}}
|
||||
MASKED = ["card", "buyer_phone"]
|
||||
PHONE = "13900000001"
|
||||
|
||||
|
||||
def _leaks(done: masking.QueryMask) -> bool:
|
||||
return "1390000000" in json.dumps(done.payload, ensure_ascii=False)
|
||||
|
||||
|
||||
def masked(sql: str, columns: list[str], rows: list[list], kind: str = "postgresql") -> masking.QueryMask:
|
||||
payload = {"columns": columns, "rows": rows, "sql": sql}
|
||||
return masking.mask_query_result(payload, MASKED, kind, schema=SCHEMA)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# VALUES 里的 (表达式).*
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
#: 复合类型的列、子查询交回的复合值、匿名 ROW、整行 × LATERAL、JOIN LATERAL、CTE、FROM 里的 VALUES：(SQL, 结果列, 一行)。
|
||||
#: 号码都在第二列起：以前 LATERAL VALUES 一个元组项只算 column1，column2 起的号码原样交了出去
|
||||
VALUES_EXPANSIONS = [
|
||||
pytest.param("SELECT v.* FROM people3 p, LATERAL (VALUES ((p.card).*)) v", ["column1", "column2"],
|
||||
["某路 1 号", PHONE], id="LATERAL-取 *"),
|
||||
pytest.param("SELECT v.column2 AS c FROM people3 p, LATERAL (VALUES ((p.card).*)) v", ["c"], [PHONE],
|
||||
id="LATERAL-取第二列"),
|
||||
pytest.param("SELECT x.column2 FROM people3 p JOIN LATERAL (VALUES ((p.card).*)) x ON true", ["column2"], [PHONE],
|
||||
id="JOIN LATERAL"),
|
||||
pytest.param("WITH v AS (VALUES ((SELECT card FROM people3 WHERE id = 1).*)) SELECT * FROM v",
|
||||
["column1", "column2"], ["某路 1 号", PHONE], id="CTE 里的 VALUES"),
|
||||
pytest.param("SELECT t FROM (VALUES ((SELECT card FROM people3 WHERE id = 1).*)) v(a, t)", ["t"], [PHONE],
|
||||
id="FROM 里的 VALUES-列名表"),
|
||||
pytest.param("SELECT v.column2 FROM sales s, LATERAL (VALUES ((ROW(s.region, s.buyer_phone)).*)) v", ["column2"],
|
||||
[PHONE], id="匿名 ROW"),
|
||||
pytest.param("SELECT v.* FROM sales s, LATERAL (VALUES ((s).*)) v", ["column1", "column2", "column3", "column4"],
|
||||
[1, "华东", PHONE, 1200], id="整行"),
|
||||
pytest.param("SELECT v.column3 FROM people3 p, LATERAL (VALUES (p.id, (p.card).*)) v", ["column3"], [PHONE],
|
||||
id="和别的项写在一个元组里"),
|
||||
# VALUES 里的 s.*、(s.*) 同样展开成好几列（s 是表，或者取了遮罩列的子查询）
|
||||
pytest.param("SELECT v.* FROM sales s, LATERAL (VALUES (s.*)) v", ["column1", "column2", "column3", "column4"],
|
||||
[1, "华东", PHONE, 1200], id="表的 *"),
|
||||
pytest.param("SELECT v.column3 FROM sales s, LATERAL (VALUES ((s.*))) v", ["column3"], [PHONE],
|
||||
id="套了括号的表的 *"),
|
||||
pytest.param("SELECT v.column2 FROM (SELECT id, buyer_phone AS p FROM sales) q, LATERAL (VALUES (q.*)) v",
|
||||
["column2"], [PHONE], id="子查询的 *"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, row", VALUES_EXPANSIONS)
|
||||
def test_an_expression_star_inside_values_masks_the_whole_result(sql, columns, row):
|
||||
"""VALUES 里的 (表达式).* 读到遮罩列：展开成几列说不清，整份结果的每一格都遮。以前只遮了 column1（或一列不遮），
|
||||
号码原样交给模型、落库。"""
|
||||
done = masked(sql, columns, [list(row)])
|
||||
assert done.whole and done.columns == columns
|
||||
assert done.payload["rows"] == [[MASKED_WORD] * len(columns)] and not _leaks(done)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, row, hidden", [
|
||||
pytest.param("SELECT v.* FROM people p, LATERAL (VALUES ((p.address).*)) v", ["column1", "column2"],
|
||||
["某路", "某市"], [], id="没设遮罩的复合类型"),
|
||||
pytest.param("SELECT v.column1 AS r FROM sales s, LATERAL (VALUES (s.region)) v", ["r"], ["华东"], [],
|
||||
id="没碰遮罩列的 VALUES"),
|
||||
pytest.param("SELECT v.column1 AS p FROM sales s, LATERAL (VALUES (s.buyer_phone)) v", ["p"], [PHONE], ["p"],
|
||||
id="VALUES 里直接写遮罩列"),
|
||||
pytest.param("SELECT v.column2 AS r FROM (SELECT id, region FROM sales) q, LATERAL (VALUES (q.*)) v", ["r"],
|
||||
["华东"], [], id="没取遮罩列的子查询的 *"),
|
||||
])
|
||||
def test_values_without_an_expression_star_on_a_masked_column_is_read_as_before(sql, columns, row, hidden):
|
||||
"""VALUES 里没有读到遮罩列的 (表达式).*：照旧按列认，没碰遮罩列的照常，直接写遮罩列的只遮那一列。"""
|
||||
done = masked(sql, columns, [list(row)])
|
||||
assert not done.whole and done.columns == hidden
|
||||
assert done.payload["rows"] == [[MASKED_WORD if c in hidden else v for c, v in zip(columns, row)]]
|
||||
|
||||
|
||||
def test_an_expression_star_inside_a_function_argument_masks_the_whole_result_too():
|
||||
"""行构造器、函数参数里的 (表达式).*：不逐个位置认展开不展开，读到遮罩列的一样整份遮（多遮，编号也遮）。"""
|
||||
done = masked("SELECT p.id AS n, row_to_json(ROW((p.card).*)) AS j FROM people3 p", ["n", "j"],
|
||||
[[1, json.dumps({"f1": "某路 1 号", "f2": PHONE})]])
|
||||
assert done.whole and done.payload["rows"] == [[MASKED_WORD, MASKED_WORD]] and not _leaks(done)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 端到端：核查 agent 的真实查询工具
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def fake_source():
|
||||
"""造出来的数据源（不连库，查询交给换掉的 run_query）：make(kind) 建一个、交回源名。"""
|
||||
made: list[str] = []
|
||||
|
||||
async def make(kind: str) -> str:
|
||||
name = f"{kind[:2]}_{uuid.uuid4().hex[:8]}"
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=name, kind=kind, host="127.0.0.1", port=5432, username="probe", database="shop",
|
||||
readonly=True, enabled=True, options={"mask_columns": ", ".join(MASKED)},
|
||||
schema_cache=SCHEMA)
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
made.append(row.id)
|
||||
return name
|
||||
|
||||
yield make
|
||||
for source_id in made:
|
||||
await engines.invalidate(source_id)
|
||||
|
||||
|
||||
async def _asked(monkeypatch, fake_source, kind: str, sql: str, columns: list[str], rows: list[list]):
|
||||
"""按剧本让核查 agent 查一次（结果由换掉的 run_query 交回），交回 (工具结果原文, 落库的快照, 通用工件接口给的原文)。"""
|
||||
from app.data.engine import QueryResult
|
||||
from app.tools import datasource
|
||||
|
||||
async def driver(source, statement, **kwargs):
|
||||
assert statement == sql
|
||||
return QueryResult(columns=list(columns), rows=[list(r) for r in rows], row_count=len(rows), truncated=False,
|
||||
elapsed_ms=1, sql=statement)
|
||||
|
||||
monkeypatch.setattr(datasource, "run_query", driver)
|
||||
name = await fake_source(kind)
|
||||
script = Script(monkeypatch, [q(name, sql, "c1")], {"value": None, "rationale": "x"})
|
||||
out, _ = await ask(name)
|
||||
snapshot = artifact_store.load(out.queries[0]["artifact"])
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||||
got = await client.get(f"/api/artifacts/{out.queries[0]['artifact']}")
|
||||
assert got.status_code == 200
|
||||
return script.tool_results()["c1"], snapshot, got.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, rows", [
|
||||
("SELECT v.* FROM people3 p, LATERAL (VALUES ((p.card).*)) v", ["column1", "column2"],
|
||||
[["某路 1 号", PHONE], ["某路 2 号", "13900000002"]]),
|
||||
("SELECT v.column2 AS c FROM people3 p, LATERAL (VALUES ((p.card).*)) v", ["c"], [[PHONE], ["13900000002"]]),
|
||||
], ids=["取 *", "取第二列"])
|
||||
async def test_values_expansions_never_reach_the_model_or_the_store(monkeypatch, fake_source, sql, columns, rows):
|
||||
"""交给模型的、落库的、通用工件接口给的都只有「已遮罩」，masked_at_query 是全部的列，快照记着整份遮。以前
|
||||
masked_at_query 是 ['column1'] 或空的，号码原样交给模型、落库。"""
|
||||
seen, snapshot, api = await _asked(monkeypatch, fake_source, "postgresql", sql, columns, rows)
|
||||
assert MASKED_WORD in seen and "1390000000" not in seen, seen
|
||||
assert snapshot["masked_at_query"] == columns and snapshot["masked_whole"] is True
|
||||
assert "1390000000" not in json.dumps(snapshot, ensure_ascii=False) and "1390000000" not in api
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 记着别的会话的 SQL、被锁那一行的值的视图
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
#: 另一个会话正在执行的语句（号码写在条件里）：会话、事务、锁、日志一类的视图交回的就是它
|
||||
BUSY = f"SELECT id, SLEEP(4) FROM sales WHERE buyer_phone = '{PHONE}' FOR UPDATE"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, kind", [
|
||||
pytest.param("SELECT trx_query AS t FROM information_schema.INNODB_TRX", "mysql", id="MySQL 正在执行的事务"),
|
||||
pytest.param("SELECT trx_query FROM INFORMATION_SCHEMA.innodb_trx WHERE trx_state = 'RUNNING'", "mysql",
|
||||
id="大小写混写"),
|
||||
pytest.param("SELECT lock_data AS d FROM information_schema.INNODB_LOCKS", "mariadb", id="MariaDB 被锁的行"),
|
||||
pytest.param("SELECT statement_text AS t FROM information_schema.QUERY_CACHE_INFO", "mariadb",
|
||||
id="MariaDB 查询缓存"),
|
||||
pytest.param("SELECT sql_text AS t FROM dba_sqlset_statements", "oracle", id="Oracle SQL 调优集"),
|
||||
pytest.param("SELECT value_string AS t FROM dba_sqlset_binds", "oracle", id="Oracle 调优集的绑定变量"),
|
||||
pytest.param("SELECT sql_text AS t FROM dba_sql_plan_baselines", "oracle", id="Oracle 固定的执行计划"),
|
||||
pytest.param("SELECT sql_text AS t FROM dba_sql_profiles", "oracle", id="Oracle SQL Profile"),
|
||||
pytest.param("SELECT value AS t FROM dba_sqltune_binds", "oracle", id="Oracle 调优顾问的绑定变量"),
|
||||
pytest.param("SELECT undo_sql AS t FROM flashback_transaction_query", "oracle", id="Oracle 闪回事务"),
|
||||
pytest.param("SELECT sql_redo AS t FROM v$logmnr_contents", "oracle", id="Oracle LogMiner"),
|
||||
pytest.param("SELECT sql_text AS t FROM v$xml_audit_trail", "oracle", id="Oracle XML 审计"),
|
||||
pytest.param("SELECT trace AS t FROM information_schema.OPTIMIZER_TRACE", "mysql", id="MySQL 优化器跟踪"),
|
||||
# 动态视图 V$ / GV$ / X$、AWR 历史不逐个认，一律算
|
||||
pytest.param("SELECT client_info AS t FROM v$session", "oracle", id="Oracle V$ 动态视图"),
|
||||
pytest.param("SELECT kglnaobj AS t FROM x$kglob", "oracle", id="Oracle X$ 底表"),
|
||||
pytest.param("SELECT sql_text AS t FROM gv$sql_monitor", "oracle", id="Oracle GV$"),
|
||||
pytest.param("SELECT event AS t FROM dba_hist_active_sess_history", "oracle", id="Oracle AWR 历史"),
|
||||
pytest.param("SELECT plan_table_output AS t FROM TABLE(DBMS_XPLAN.DISPLAY_CURSOR('a1', NULL, 'ADVANCED "
|
||||
"+PEEKED_BINDS'))", "oracle", id="Oracle 游标的执行计划和绑定变量"),
|
||||
pytest.param("SELECT DBMS_SQLTUNE.REPORT_SQL_MONITOR(sql_id => 'a1', type => 'TEXT') AS t FROM dual", "oracle",
|
||||
id="Oracle SQL 监控报告"),
|
||||
])
|
||||
def test_views_that_keep_other_sessions_sql_mask_the_whole_result(sql, kind):
|
||||
"""别的会话执行过的 SQL 原文、被锁那一行的键值、闪回和日志里的改动前后的值：读的是哪一列认不出，整份遮。以前
|
||||
masked=[]，条件里的号码原样交给模型、落库。"""
|
||||
done = masked(sql, ["t"], [[BUSY]], kind=kind)
|
||||
assert done.whole and done.payload["rows"] == [[MASKED_WORD]]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns", [
|
||||
pytest.param("SHOW ENGINE INNODB STATUS", ["Type", "Name", "Status"], id="InnoDB 状态"),
|
||||
pytest.param("show engine innodb status", ["Type", "Name", "Status"], id="小写"),
|
||||
pytest.param("SHOW BINLOG EVENTS IN 'binlog.000002'", ["Log_name", "Pos", "Info"], id="binlog 事件"),
|
||||
pytest.param("SHOW RELAYLOG EVENTS", ["Log_name", "Pos", "Info"], id="relay log 事件"),
|
||||
pytest.param("SHOW REPLICA STATUS", ["Last_SQL_Error"], id="复制状态"),
|
||||
pytest.param("SHOW SLAVE STATUS", ["Last_SQL_Error"], id="复制状态-旧写法"),
|
||||
pytest.param("SHOW PROFILES", ["Query_ID", "Duration", "Query"], id="PROFILES"),
|
||||
pytest.param("SHOW WARNINGS", ["Level", "Code", "Message"], id="WARNINGS"),
|
||||
pytest.param("SHOW CREATE VIEW vip_sales", ["View", "Create View"], id="视图定义"),
|
||||
# SHOW CREATE TABLE 对视图交回的是视图定义（MySQL 8.0 实测），建表语句里还有默认值、CHECK 约束、分区的取值
|
||||
pytest.param("SHOW CREATE TABLE vip_sales", ["View", "Create View"], id="建表语句-视图"),
|
||||
pytest.param("SHOW CREATE TABLE sales", ["Table", "Create Table"], id="建表语句"),
|
||||
pytest.param("SHOW TRIGGERS", ["Trigger", "Statement"], id="触发器"),
|
||||
pytest.param("SHOW EXPLAIN FOR 12", ["id", "Extra"], id="MariaDB 别的连接的计划"),
|
||||
])
|
||||
def test_show_statements_beyond_the_table_structure_mask_the_whole_result(sql, columns):
|
||||
"""SHOW 里只交回表结构、库清单的几种以外（引擎状态、日志、复制、出错信息、视图和触发器的定义、建表语句……）：交回的
|
||||
是原文，可能带着别的会话 SQL 里的号码、定义里写死的号码，整份遮。以前守卫放行（SHOW 是只读的），遮罩认不出，号码
|
||||
原样。"""
|
||||
row = [BUSY] * len(columns)
|
||||
done = masked(sql, columns, [row], kind="mysql")
|
||||
assert done.whole and done.payload["rows"] == [[MASKED_WORD] * len(columns)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, row", [
|
||||
pytest.param("SHOW TABLES", ["Tables_in_shop"], ["sales"], id="表清单"),
|
||||
pytest.param("SHOW FULL TABLES FROM shop", ["Tables_in_shop", "Table_type"], ["sales", "BASE TABLE"],
|
||||
id="FULL TABLES"),
|
||||
pytest.param("SHOW COLUMNS FROM sales", ["Field", "Type"], ["region", "varchar(20)"], id="列"),
|
||||
pytest.param("SHOW FULL FIELDS FROM sales", ["Field", "Type"], ["region", "varchar(20)"], id="FULL FIELDS"),
|
||||
pytest.param("SHOW INDEX FROM sales", ["Table", "Key_name"], ["sales", "PRIMARY"], id="索引"),
|
||||
pytest.param("SHOW KEYS FROM sales", ["Table", "Key_name"], ["sales", "PRIMARY"], id="KEYS"),
|
||||
pytest.param("SHOW DATABASES", ["Database"], ["shop"], id="库清单"),
|
||||
pytest.param("SHOW CREATE DATABASE shop", ["Database", "Create Database"], ["shop", "CREATE DATABASE `shop`"],
|
||||
id="建库语句"),
|
||||
pytest.param("SHOW TABLE STATUS LIKE 'sales'", ["Name", "Engine", "Rows"], ["sales", "InnoDB", 3], id="表状态"),
|
||||
pytest.param("DESCRIBE sales", ["Field", "Type"], ["region", "varchar(20)"], id="DESCRIBE 表"),
|
||||
pytest.param("EXPLAIN sales", ["Field", "Type"], ["region", "varchar(20)"], id="EXPLAIN 表"),
|
||||
pytest.param("DESC sales region", ["Field", "Type"], ["region", "varchar(20)"], id="DESC 表的一列"),
|
||||
])
|
||||
def test_show_and_describe_of_the_table_structure_stay_visible(sql, columns, row):
|
||||
"""只交回表结构、库清单的 SHOW、DESCRIBE（核查 agent 探查表结构靠它们）照常。"""
|
||||
done = masked(sql, columns, [list(row)], kind="mysql")
|
||||
assert not done.whole and done.payload["rows"] == [list(row)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql", [
|
||||
"EXPLAIN FORMAT=TREE SELECT s.id FROM sales s JOIN people3 p ON p.id = s.id WHERE s.id = 1",
|
||||
"EXPLAIN FORMAT = JSON SELECT id FROM sales WHERE id = 1",
|
||||
"EXPLAIN ANALYZE SELECT id FROM sales WHERE id = 1",
|
||||
"DESCRIBE SELECT id FROM sales WHERE id = 1",
|
||||
"EXPLAIN FOR CONNECTION 12",
|
||||
"EXPLAIN FORMAT=JSON FOR CONNECTION 12",
|
||||
"EXPLAIN (SELECT id FROM sales)",
|
||||
"EXPLAIN WITH t AS (SELECT id FROM sales) SELECT * FROM t",
|
||||
"EXPLAIN EXTENDED SELECT id FROM sales",
|
||||
])
|
||||
def test_explaining_a_statement_masks_the_whole_result(sql):
|
||||
"""EXPLAIN 一条语句：MySQL 优化时把只有一行的表读出来、把值当常量代进条件，FORMAT=TREE / JSON、EXPLAIN ANALYZE 写出的
|
||||
条件里就是那一行的值；FOR CONNECTION 读的是别的连接正在执行的语句。整份遮。"""
|
||||
done = masked(sql, ["EXPLAIN"], [[f"-> Filter: (s.buyer_phone = '{PHONE}')"]], kind="mysql")
|
||||
assert done.whole and done.payload["rows"] == [[MASKED_WORD]]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql, columns, rows", [
|
||||
("SELECT trx_query AS t FROM information_schema.INNODB_TRX", ["t"], [[BUSY]]),
|
||||
("SHOW ENGINE INNODB STATUS", ["Type", "Name", "Status"],
|
||||
[["InnoDB", "", f"---TRANSACTION 1234, ACTIVE 2 sec\nMySQL thread id 9, query id 70 localhost root\n{BUSY}"]]),
|
||||
], ids=["INNODB_TRX", "SHOW ENGINE INNODB STATUS"])
|
||||
async def test_other_sessions_sql_never_reaches_the_model_or_the_store(monkeypatch, fake_source, sql, columns, rows):
|
||||
"""端到端（MySQL）：另一个会话正在执行的 SQL 里写着号码，交给模型的、落库的、通用工件接口给的都只有「已遮罩」，
|
||||
快照记着整份遮。以前 masked_at_query 为空，SQL 原文连同号码原样交出。"""
|
||||
seen, snapshot, api = await _asked(monkeypatch, fake_source, "mysql", sql, columns, rows)
|
||||
assert MASKED_WORD in seen and "1390000000" not in seen, seen
|
||||
assert snapshot["masked_at_query"] == columns and snapshot["masked_whole"] is True
|
||||
assert "1390000000" not in json.dumps(snapshot, ensure_ascii=False) and "1390000000" not in api
|
||||
|
||||
|
||||
def test_the_new_views_are_on_the_list_the_bundle_shares():
|
||||
"""可验证报告和工具边界共用一份名单：新补的视图两边都认（报告那边本来就整个挡 information_schema）。"""
|
||||
from app.bundle import build
|
||||
|
||||
for sql in ("SELECT trx_query FROM information_schema.INNODB_TRX", "SELECT sql_text FROM dba_sqlset_statements",
|
||||
"SELECT undo_sql FROM flashback_transaction_query", "SELECT sql_redo FROM v$logmnr_contents"):
|
||||
assert build._INDIRECT_READS.search(sql) and masking._INDIRECT.search(sql), sql
|
||||
Reference in New Issue
Block a user