diff --git a/backend/app/engine/inquiry.py b/backend/app/engine/inquiry.py index ef8d021f..9e652f68 100644 --- a/backend/app/engine/inquiry.py +++ b/backend/app/engine/inquiry.py @@ -12,7 +12,11 @@ 上传表格不算，见 _usable_sources）；上限按设置（每次的查询次数、时长，全局每日金额）。回答先给结论，数字只能引用 agent 自己查询的格子，或者报告里已有的指标、运行输入；回答文档由 compose_doc 组装，裸数字被标出来（提问人自己写的数除外）。 封存之后的追问查的是数据源现在的数据，任务里给报告生成的时间，回答末尾注明；触顶、查询失败、引用解析不了也在末尾注明 -3. 遮罩（_plan）：哪些列要遮照证据面板的规则（查询当时记下的，加上数据源现在设的），另按 SQL 认出改了名字的遮罩列 +3. 遮罩（_plan）：agent 的查询结果在核查 agent 的工具边界就遮好了（engine/masking.py 的 mask_query_result）：取自 + 遮罩列的格子交给模型、落进快照之前就是「已遮罩」，模型看不到遮罩值。这里是第二道防线：查询时遮掉的值由 verify_agent + 交回（masked_values，只在内存里），连同报告自己快照里的遮罩值，照下面的办法按值遮；事后取质询工件时内存里的值已经 + 不在，SQL 检查结果照事件里记下的 SQL 换（_recorded_swaps）。 + 哪些列要遮照证据面板的规则（查询当时记下的，加上数据源现在设的），另按 SQL 认出改了名字的遮罩列 （起别名、套函数、经子查询转一道、和 * 写在同一支里，_derived_hidden；认不出落在哪一列的，整份快照遮掉；按遮罩列 分档、当分组键或套在汇总里数个数的不算，_band_only）；快照按列遮，再按值遮（别的列里写着遮罩值的格子，_masked_view；质询 工件里，文字格里夹着的遮罩值另按子串换）。回答正文、查询清单里的遮罩值换成「已遮罩」（_Secrets、_scrub_sql）： @@ -73,8 +77,10 @@ - 质询工件里 SQL 检查结果的片段是检查器按语法树重新写出来的，按整条 SQL 里整个换掉了的字面量照值换：同一个值在 片段里别处出现也换（多遮，算式里换掉的 0、1 这类也是）；截断处半截的值去掉（停在字符串、带引号的名字、注释里的， 从开头的引号、注释起整个去掉）。说明里码值表列出的码值和别的字一样按值换，恰好是遮罩值的码值也换（多遮） -- 运行详情的「工件」页签、通用的工件接口按工件原样给：质询的查询快照记在这次运行名下（同报告自己的查询快照），从 - 那里打开不经这里的遮罩 +- 运行详情的「工件」页签、通用的工件接口按工件原样给：质询的查询快照落库时就遮好了（工具边界先遮），从那里打开也 + 看不到取自遮罩列的值；但工具边界只认这份结果里的遮罩值，别的查询、报告快照里遮罩列的值写在这份结果没设遮罩的列里时 + （没设遮罩的联系方式列、备注），从那里打开照原样（质询工件接口按回答时的办法换）。报告自己的查询快照（写作时的查询） + 仍原样存着，从那里打开不经这里的遮罩 - 遮罩值最多收 SECRETS_MAX 种写法，再多的按值认不出（按列遮照旧） - 换了写法的号码只认数字之间夹着一到三个空格、连字符、点、括号（含全角），带 +86 / 0086 / 86 前缀、全角数字：写成 汉字（一三九……）、中间夹着别的字、拆成几处写的认不出；和别的数字只隔着一个连字符或点连着的（13900000001-2、 @@ -84,6 +90,7 @@ from __future__ import annotations import asyncio import contextlib +import difflib import itertools import logging import operator @@ -91,7 +98,7 @@ import re import time from collections import OrderedDict from collections.abc import Callable, Iterable, Iterator -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from datetime import datetime, timezone from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal from typing import Any @@ -1618,13 +1625,15 @@ async def _source_kinds(names: Iterable[str]) -> dict[str, str | None]: async def _plan(own: dict[str, dict[str, Any]], others: list[dict[str, Any]], given: list[str | None], shown: list[tuple[str, str, str]] | None = None, live: dict[str, list[str]] | None = None, *, - embedded: frozenset[str] = frozenset()) -> _Plan: + embedded: frozenset[str] = frozenset(), seen: _Secrets | None = None) -> _Plan: """own：agent 自己的查询快照（工件 id → 内容）；others：报告自己的查询快照（被问的那句可能直接写着遮罩列的值， agent 会照着复述、写进查询条件）；given：问题、选中的原文、报告里列出的指标和运行输入的值（都已经摆在读者眼前， 里面的值哪里都不换）；shown：agent 执行的查询，(SQL, 数据源名, 快照工件 id)（照原样留在 SQL 里的阈值、LIMIT， 回答正文里不换，见 _Secrets.show；取质询工件不用：快照里的格子照旧按值遮）；live：用得上的数据源现在设的遮罩列 （快照没取到的查询按它认 SQL 里的比较，见 _compared）；embedded：这几份快照的文字格里夹着的遮罩值也按子串换 - （只给要取的那份质询工件，见 _masked_view）。 + （只给要取的那份质询工件，见 _masked_view）；seen：agent 查询时在工具边界遮掉的值（verify_agent 交回的 + masked_values，只在内存里）。agent 自己的快照查询时就遮好了，遮罩列里收不到值；模型照着别处看到的号码写进回答、 + 查询条件时，按值遮要靠它认出来。 查数据源的遮罩要读库，在这里读；解析 SQL、收上万个值、逐格比对是纯计算，放进线程：放在事件循环里会挡住同一进程 里别的运行（同 tools/datasource 的 _sql_checks）。""" @@ -1636,13 +1645,13 @@ async def _plan(own: dict[str, dict[str, Any]], others: list[dict[str, Any]], gi based = [(snapshot, (await _hidden_columns(snapshot, masks))[0]) for snapshot in others] kinds = await _source_kinds([*(str(s.get("source") or "") for s in [*own.values(), *others]), *(name for _, name, _ in shown)]) - return await asyncio.to_thread(_settle_plan, own, named, based, kinds, given, shown, live or {}, embedded) + return await asyncio.to_thread(_settle_plan, own, named, based, kinds, given, shown, live or {}, embedded, seen) def _settle_plan(own: dict[str, dict[str, Any]], named: dict[str, tuple[set[str], list[str]]], based: list[tuple[dict[str, Any], set[str]]], kinds: dict[str, str | None], given: list[str | None], shown: list[tuple[str, str, str]], live: dict[str, list[str]], - embedded: frozenset[str]) -> _Plan: + embedded: frozenset[str], seen: _Secrets | None = None) -> _Plan: secrets = _Secrets() hidden: dict[str, set[str]] = {} tainted: dict[str, set[str]] = {} @@ -1654,6 +1663,7 @@ def _settle_plan(own: dict[str, dict[str, Any]], named: dict[str, tuple[set[str] kind = kinds.get(str(snapshot.get("source") or "")) secrets.add(snapshot, _derived_hidden(snapshot, base, kind)[0], labels=_snapshot_labels(snapshot, kind), base=base) + secrets.merge(seen) secrets.protect(given) secrets.show(_shown_literals([(sql, kinds.get(name), _compared(tainted, artifact, live, name)) for sql, name, artifact in shown], secrets)) @@ -1755,7 +1765,8 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict output_schema=INQUIRY_SCHEMA, max_queries=part.get("max_queries"), timeout_s=part.get("timeout_s"), daily_max_usd=settings.get("daily_max_usd"), run_id=op.run_id, data_versions=op.versions) - # 遮罩：agent 自己的查询快照、报告自己的查询快照（被问的那句可能直接写着遮罩列的值，agent 会照着复述） + # 遮罩：agent 自己的查询快照（查询时就遮好了，遮掉的值在 outcome.masked_values 里）、报告自己的查询快照（被问的 + # 那句可能直接写着遮罩列的值，agent 会照着复述） own: dict[str, dict[str, Any]] = {} for query in outcome.queries: snapshot = _quiet_load(artifact_store.load, query.get("artifact")) @@ -1765,7 +1776,7 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict and _query_shaped(snapshot := _quiet_load(op.loader, entry.get("artifact")))] plan = await _plan(own, others, [ask.question, ask.quote, *_ref_values(refs)], [(str(q.get("sql") or ""), str(q.get("source") or ""), str(q.get("artifact") or "")) - for q in outcome.queries], live) + for q in outcome.queries], live, seen=outcome.masked_values) text, doc_id, failed = "", None, None source = (outcome.rationale or {}).get("source") if outcome.result is not None else None @@ -2059,22 +2070,57 @@ async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]: elif isinstance(shown.get("sql"), str): shown = {**shown, "sql": _scrub_sql(shown["sql"], plan.secrets, compared, kind)} if "checks" in shown: - shown = {**shown, "checks": _masked_checks(shown["checks"], content.get("sql"), plan.secrets, compared, kind)} + whole, pieces = _recorded_swaps(content.get("sql"), recorded, kind) if recorded is not None else (set(), set()) + shown = {**shown, "checks": _masked_checks(shown["checks"], content.get("sql"), plan.secrets, compared, kind, + swapped=whole, pieces=pieces)} out: dict[str, Any] = {"id": artifact_id, "content": shown, "masked": plan.masked[artifact_id]} if gone := plan.gone.get(artifact_id): out["mask_note"] = SOURCE_GONE.format(source="、".join(gone)) return out -def _masked_checks(checks: Any, sql: Any, secrets: _Secrets, compared: set[str], kind: str | None +def _recorded_swaps(sql: Any, recorded: str, kind: str | None) -> tuple[set[str], set[str]]: + """回答时这条 SQL 里换掉的值：快照里原样的那条和事件里记下的那条（_scrub_sql 换过的）逐词对，原样那条里的字面量 + 到了记下的那条里整个换成了「已遮罩」的（第一项，字面量的字），或者其中一截换成了「已遮罩」的（第二项，那一截， + LIKE '%13900000001%' 这类）。_scrub_sql 只换字面量、名字和注释里的值，一个词换成一个词，两条分出来的词一一对得上； + 对不上（分词出错这类）、没有记下的那条时，什么都不认。""" + if not isinstance(sql, str) or not sql.strip() or not recorded or sql == recorded: + return set(), set() + before, after = _sql_tokens(sql, kind), _sql_tokens(recorded, kind) + if before is None or after is None or len(before[0]) != len(after[0]): + return set(), set() + word = _masked_word() + whole: set[str] = set() + pieces: set[str] = set() + for old, kind_of, new in zip(before[0], before[1], after[0]): + if not _is_literal(kind_of) or old.text == new.text or word not in new.text: + continue + if new.text == word: + whole.add(old.text) + continue + for op, i1, i2, j1, j2 in difflib.SequenceMatcher(None, old.text, new.text, autojunk=False).get_opcodes(): + if op == "replace" and new.text[j1:j2] == word and old.text[i1:i2].strip(): + pieces.add(old.text[i1:i2]) + return whole, pieces + + +def _masked_checks(checks: Any, sql: Any, secrets: _Secrets, compared: set[str], kind: str | None, *, + swapped: set[str] | frozenset[str] = frozenset(), pieces: set[str] | frozenset[str] = frozenset() ) -> list[dict[str, Any]]: """快照里记下的 SQL 检查结果（数据源查询工具对照数据目录查出的问题）交给质询工件的样子：字段照证据面板 （_query_checks），给模型的那句改法不交出去——它照抄了 SQL 片段。片段（sql_excerpt）、说明（message）摘自、照引 同一条 SQL（快照里原样的那条），照它在查询清单里的换法换掉遮罩值（_literal_swaps）：SQL 换了、片段和说明没换， - 等于把换掉的值从旁边露出来；片段单拿出来认不出在找哪一个值，不能自己另认一遍。""" + 等于把换掉的值从旁边露出来；片段单拿出来认不出在找哪一个值，不能自己另认一遍。 + swapped / pieces：回答时这条 SQL 里整个换掉的字面量、换掉的一截（_recorded_swaps，照事件里记下的那条认）。回答时 + agent 自己查询里遮掉的值还在内存里，事后取工件时已经不在了（快照查询时就遮好了，收不到值），照记下的 SQL 补上。""" from app.api.evidence import _query_checks whole = _literal_swaps(sql, secrets, compared, kind) + if swapped: + whole = (whole or set()) | set(swapped) + if pieces: + secrets = replace(secrets, strong=secrets.strong | set(pieces), weak=secrets.weak | set(pieces), _index=None, + _compiled={}) out: list[dict[str, Any]] = [] for check in _query_checks({"checks": checks}): if isinstance(check.get("sql_excerpt"), str): diff --git a/backend/app/engine/masking.py b/backend/app/engine/masking.py index 39d68420..082d8000 100644 --- a/backend/app/engine/masking.py +++ b/backend/app/engine/masking.py @@ -2,18 +2,24 @@ 数据源在 options.mask_columns 里设了遮罩的列（手机号、姓名这类），证据面板按列名遮（api/evidence.py）。核查 agent （engine/verify_agent.py：双盲复算、反方审稿、审批质询共用）自己写 SQL，起别名、套函数、经子查询转一道，结果列名就 -对不上遮罩列；按值认又有门槛和写法的问题。认法集中在这里，审批质询（engine/inquiry.py）、双盲复算（nodes/dual_check.py， -经 inquiry 的同名导出）共用： +对不上遮罩列；按值认又有门槛和写法的问题。认法集中在这里，核查 agent 的查询工具（tools/datasource 的遮罩开关）、审批 +质询（engine/inquiry.py）、双盲复算（nodes/dual_check.py，经 inquiry 的同名导出）共用： + +- 工具边界先遮（mask_query_result）：核查 agent 的查询结果交给模型、落进快照之前，取自遮罩列的格子换成「已遮罩」， + 同一份结果里别的文字格夹着的遮罩值也换掉。模型看不到遮罩值，就不会把它写进回答、理由、SQL——事后按值认每一轮 + 复核都找到新的漏法，根治靠这一步；审批质询回答里的按值遮降为第二道防线 - 哪些列（_derived_hidden）：从遮罩列本身起，按 SQL 一层层认出值取自遮罩列的结果列（别名、函数、子查询、WITH、UNION 的 另一支、和 * 写在同一支里没起别名的表达式）；COUNT 只数个数、窗口的分区键只决定次序、按遮罩列分档又当分组键或套在 汇总里数个数的不算（_value_columns、_band_only）。认不出落在哪一列的、SQL 解析不了又写着遮罩列的，整份快照的列都算 - 只装着分档标签的列（_label_columns）：标签是 agent 写在 SQL 里的字，不当遮罩值 - 哪些值（_Secrets）：遮罩列里出现过的值，按值认有门槛（SECRET_MIN、SECRET_DIGITS），换的时候先按词、按开头几个字筛出 - 正文里真有的那几个值，不现编上万个分支的正则 + 正文里真有的那几个值，不现编上万个分支的正则；正文、文字格里换了写法的长号码（199-1474-2785、+86 …、全角数字）按 + 纯数字写法在集合里查（loose_spans） - 快照遮好的样子（_masked_view） -这里的函数原来都在 engine/inquiry.py 里，名字不变，inquiry 照旧导出（别处按 inquiry 的名字引用）。 +除 mask_query_result、换了写法的号码之外，这里的函数原来都在 engine/inquiry.py 里，名字不变，inquiry 照旧导出（别处按 +inquiry 的名字引用）。 """ from __future__ import annotations @@ -810,19 +816,17 @@ def _loose_whole(text: str, run: re.Match[str], groups: list[tuple[int, int, str def _loose_span(text: str, run: re.Match[str], groups: list[tuple[int, int, str]], i: int, j: int) -> tuple[int, int]: - """第 i 到第 j 截这个号码在 text 里换掉的那一段：从第一截到最后一截，从这一串开头起的连同前面的 + 和括号。括号配成 - 对：号码里有落单的左括号，紧跟着的右括号一起换掉；配不上的落单左括号留着不换（「（电话 199-1474-2785）」两个括号 - 都留着）；落单的右括号，紧挨在前面的左括号一起换掉（(199)1474-2785 从第二截起认出来时）。""" + """第 i 到第 j 截这个号码在 text 里换掉的那一段：从第一截到最后一截，紧挨在前面的 + 一起换。括号只换号码里的：号码 + 里有落单的右括号（(199)1474-2785、(+86)199…），把前面配对的左括号一起换掉；落单的左括号，把后面紧跟着的右括号一起 + 换掉。整个号码外面套着的一对括号是标点（「来自（13900000001）」），留着不换。""" start, end = groups[i][0], groups[j][1] - if i == 0: - start = run.start() + if i == 0 and start > run.start() and text[start - 1] in "+＋": + start -= 1 depth = sum((c in _OPENS) - (c in _CLOSES) for c in text[start:end]) - while depth > 0 and end < len(text) and text[end] in _CLOSES: - end, depth = end + 1, depth - 1 - while depth > 0 and text[start] in _OPENS: - start, depth = start + 1, depth - 1 while depth < 0 and start > 0 and text[start - 1] in _OPENS: start, depth = start - 1, depth + 1 + while depth > 0 and end < len(text) and text[end] in _CLOSES: + end, depth = end + 1, depth - 1 return start, end @@ -844,8 +848,9 @@ def _pattern(value: str) -> str: @dataclass class _Secrets: - """遮罩列里出现过的值。模型看得到原始的查询结果（工具交回的是原样），回答里照着复述、或者把值写进查询条件， - 都会把它带进回答正文和查询清单；按值找出来换成「已遮罩」。这只减少暴露，不是安全边界（同证据面板的遮罩）。 + """遮罩列里出现过的值。模型照着报告正文、提问、别处的文字格里看到的值复述、或者把值写进查询条件，都会把它带进 + 回答正文和查询清单；按值找出来换成「已遮罩」。核查 agent 自己的查询结果在工具边界就遮好了（mask_query_result）， + 这里是第二道防线。这只减少暴露，不是安全边界（同证据面板的遮罩）。 strong 是按值认得出的写法（够长的数、两个字以上的文字），正文、SQL 里的字面量、名字和注释、别的列里的格子、列名都按 它换；weak 是遮罩列里的全部写法（含短的数），只用在 SQL 里和遮罩列做等值比较的字面量上；shown 是 SQL 里照原样 @@ -870,6 +875,8 @@ class _Secrets: written = {name: texts for name, texts in (labels or {}).items() if name not in own} columns = [str(c) for c in snapshot.get("columns") or []] index = [i for i, c in enumerate(columns) if c.lower() in hidden] + # 核查 agent 的查询快照在工具边界就遮好了（mask_query_result）：遮罩列的格子是「已遮罩」，不是遮罩值 + word = _masked_word() for row in snapshot["rows"]: if len(self.weak) >= SECRETS_MAX: break @@ -880,7 +887,7 @@ class _Secrets: else: continue for name, value in cells: - if value is None or isinstance(value, bool): + if value is None or isinstance(value, bool) or value == word: continue forms = _forms(value) if forms & written.get(name, set()): @@ -891,6 +898,18 @@ class _Secrets: self.strong.add(form) self._index = None + def merge(self, other: _Secrets | None) -> None: + """并进另一份收好的遮罩值（核查 agent 查询时在工具边界遮掉的，见 mask_query_result）。合起来最多 SECRETS_MAX 种。""" + if other is None or not other.weak: + return + for form in other.weak: + if len(self.weak) >= SECRETS_MAX: + break + self.weak.add(form) + if form in other.strong: + self.strong.add(form) + self._index, self._compiled = None, {} + def protect(self, texts: Iterable[str | None]) -> None: """已经摆在读者眼前的值不换：提问人自己写进问题、选中原文里的（「13900000001 这位客户」「排除 300 以下的」）， 报告里列出的指标、运行输入的值——换了只会让回答、查询对不上问题和报告。正文、格子、SQL 里都不换。""" @@ -1039,8 +1058,8 @@ class _Secrets: return self._index -def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets, *, embedded: bool = False - ) -> tuple[dict[str, Any], list[str]]: +def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets, *, embedded: bool = False, + rename: bool = True) -> tuple[dict[str, Any], list[str]]: """一份查询快照遮好的样子和遮掉的列：要遮的列（hidden，小写）每一格都换成「已遮罩」、类型记成文本（渲染时不当数）； 别的列里写着按值认得出的遮罩值的格子也换（另一张表里没设遮罩的同一个手机号、改了名字没认出来的遮罩列）。 列名里写着遮罩值的（agent 把遮罩值写进了别名，「13900000001 的单」），列名也换（同 SQL 里的名字，_scrub_sql）。 @@ -1048,7 +1067,8 @@ def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets, embedded 只给质询工件用：别的列里一格文字夹着遮罩值（备注里写着手机号，换了写法的也算），把那一截换成「已遮罩」， 这一格别的字照原样，门槛同整格（_Secrets.scrub）；这一列不记进遮掉的列——界面把记进去的列整列画成「已遮罩」，这一列在数据源 - 里也没设遮罩。回答一路不这样换：渲染出来露了就整份不交（_leaked），不然回答被问的那一格看着像查到了别的东西。""" + 里也没设遮罩。回答一路不这样换：渲染出来露了就整份不交（_leaked），不然回答被问的那一格看着像查到了别的东西。 + rename 为假时列名不换（工具边界先遮时：列名是模型自己写的，换了它引用的列名就对不上）。""" columns = [str(c) for c in snapshot.get("columns") or []] index = {i for i, c in enumerate(columns) if c.lower() in hidden} by_value = bool(secrets.strong) @@ -1058,7 +1078,7 @@ def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets, touched = set(index) edited = False position = {c: i for i, c in enumerate(columns)} - names = [secrets.scrub(c) for c in columns] if by_value else columns + names = [secrets.scrub(c) for c in columns] if by_value and rename else columns def cell(i: int | None, value: Any) -> Any: nonlocal edited @@ -1097,3 +1117,53 @@ def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets, types.update({names[i]: "text" for i in index}) view = {**snapshot, "rows": rows, "column_types": types, **({"columns": names} if renamed else {})} return view, [names[i] for i in sorted(touched)] + + +# -------------------------------------------------------------------------- +# 工具边界：查询结果交给核查 agent 之前先遮 +# -------------------------------------------------------------------------- + + +@dataclass +class QueryMask: + """一份查询结果在工具边界遮好的样子（mask_query_result）。 + + payload：遮好的结果，交给模型的、落进快照的都是它。columns：查询时就整列遮掉的结果列（结果里的列名，按先后）。 + secrets：这份结果里遮掉的值，只在内存里交给调用方（审批质询按值遮的第二道防线：模型照着别处看到的号码写进回答、 + 查询条件时认得出），不进快照、不交给模型。""" + + payload: dict[str, Any] + columns: list[str] + secrets: _Secrets + + +def mask_query_result(payload: dict[str, Any], masked: Iterable[str], kind: str | None) -> QueryMask: + """一份查询结果（tools/datasource 的查询工具交回的样子：columns / rows / column_types / sql）在交给核查 agent、落进 + 快照之前先遮。masked 是数据源设了遮罩的列，kind 是数据源的类型（SQL 按它的方言解析）。 + + - 整列遮：列名直接对上遮罩列的，加上按 SQL 认出值取自遮罩列的（_derived_hidden：别名、函数、子查询、WITH、 + SELECT *；COUNT、只在条件里用到、分档当分组键的不算）；SQL 解析不了、又写着遮罩列的，整份结果的列都遮。这几列的 + 格子换成「已遮罩」，类型记成文本 + - 别的列：整格就是这份结果里某个遮罩值的换掉，文字格里夹着的那一截换掉（换了写法的号码也认，_Secrets.loose_spans）； + 门槛同按值遮（SECRET_DIGITS、SECRET_MIN），只装着分档标签的列里的标签不算遮罩值 + - 列名不换（模型自己写的，换了它引用的列名就对不上）；结果里一列都没取自遮罩列的原样交回：数值的汇总、只数个数的 + 查询不受影响 + + 分析本身出了意外，整份结果的列都遮：这一步挡在模型前面，分析的缺陷不能把原值放过去（宁可多遮）。""" + base = {str(c).strip().lower() for c in masked if str(c).strip()} + columns = [str(c) for c in payload.get("columns") or []] + if not base or not columns or not _query_shaped(payload): + return QueryMask(payload, [], _Secrets()) + try: + hidden, _ = _derived_hidden(payload, base, kind) + if not any(c.lower() in hidden for c in columns): + return QueryMask(payload, [], _Secrets()) + secrets = _Secrets() + secrets.add(payload, hidden, labels=_snapshot_labels(payload, kind), base=base) + secrets.prepare() + view, _ = _masked_view(payload, hidden, secrets, embedded=True, rename=False) + except Exception: # noqa: BLE001 - 认不准就整份遮掉，原值不交给模型 + logger.warning("核查查询的遮罩分析出错，整份结果按遮罩处理", exc_info=True) + hidden, secrets = {c.lower() for c in columns}, _Secrets() + view, _ = _masked_view(payload, hidden, secrets, rename=False) + return QueryMask(view, [c for c in columns if c.lower() in hidden], secrets) diff --git a/backend/app/engine/verify_agent.py b/backend/app/engine/verify_agent.py index b127eaef..55397916 100644 --- a/backend/app/engine/verify_agent.py +++ b/backend/app/engine/verify_agent.py @@ -6,6 +6,13 @@ **只读。** 工具只建给定数据源的 db_schema / db_query，建的时候 force_readonly：可写源上的写语句也被守卫拒绝 （tools/datasource.py）。这一路没有审批关卡，不能指望人在半路拦下一条 DELETE。 +**先遮再交。** 建查询工具时还打开遮罩开关（tools/datasource 的 mask_results）：数据源设了遮罩时，取自遮罩列的格子 +（按 SQL 认，起别名、套函数、经子查询或 WITH 转一道、SELECT * 都算）在交给模型、落进查询快照之前就换成「已遮罩」， +同一份结果里别的文字格夹着的遮罩值也换掉（engine/masking.py 的 mask_query_result）。模型看不到遮罩值，就不会把它 +写进回答、理由、SQL；以前是事后按值认，每一轮复核都找到新的漏法。快照里的 masked_at_query 记下查询时就遮了的列， +格子引用指到这几列的不采信。遮掉的值只在内存里交回调用方（VerifyOutcome.masked_values，审批质询按值遮的第二道防线 +用），不进快照、不交给模型、不进 as_dict。 + **数从快照取。** 每次成功的查询是一份查询快照（工具自己落的 put_json），按这个 agent 自己的顺序编 Q1、Q2…。 最后交结果时，调用方结构里的数值字段一律改成格子引用 {alias, row, column}，由代码按引用从快照里取值：模型自己 写的数字不采信，取不到就是 None，不兜底成 0。数值字段的各种写法都认（type 列表、anyOf 里的数、$ref 引用进来的、 @@ -45,7 +52,7 @@ import json import logging import time from collections.abc import Callable, Iterable -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass, field, replace from enum import Enum from typing import Any @@ -59,13 +66,14 @@ from app.engine import spend from app.engine.evidence import build_catalog, cell_eid, compose_doc, query_entry_fields, schema_ledger_entry from app.engine.expressions import CellError, column_kind, locate_cell, numeric_text from app.engine.judge import FALLBACK_MAX_TOKENS, MAX_TOKENS, _json_in, _tokens, _usd +from app.engine.masking import _Secrets from app.engine.nodes.llm import _DIGEST_BUDGET, _digest, _split_structured, _typed, explain_model_error from app.engine.state import message_text from app.engine.toolcalls import ToolTimeout, limit_fields, limit_of, run_bounded from app.providers import catalog as pricing from app.providers.budget import BudgetFallback, budget_refused, cut_off, cut_words, error_usage, truncated_error from app.providers.factory import ModelSpec, ProviderNotConfigured, bind_tools_safely, get_chat_model -from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX, build_datasource_tools +from app.tools.datasource import MASKED_VALUES_KEY, QUERY_PREFIX, SCHEMA_PREFIX, build_datasource_tools from app.tools.registry import ToolArgsError, ToolContext, Versions, args_model_of, prepare_args logger = logging.getLogger(__name__) @@ -103,6 +111,8 @@ VERIFY_SYSTEM = ( "- 先用 db_schema 工具看表结构，确认表名、字段名，再用 db_query 工具写只读 SQL。不能改数据，写语句会被拒绝\n" "- 每次成功的查询前面都标着编号【证据 Qn】，最后交结果时用它说明每个数取自哪次查询的哪一格\n" "- {limit}先想清楚要查什么，不要重复同样的查询\n" + "- 数据源设了遮罩的列（手机号、姓名这类），查询结果里显示为「已遮罩」：不要设法绕过，交结果、写理由时也不要引用" + "这些格子\n" "- 不要编造数字，查不到就照实说" ) VERIFY_FINAL_PROMPT = ( @@ -149,6 +159,8 @@ class VerifyOutcome: output_tokens, total_tokens, cost_usd, calls, unpriced_calls}。limit：触顶原因（LIMITS 之一），没触顶为 None。 answered：模型是不是自己停下来的（没被上限掐断，也没因为输出到了上限被截断）。被截断的一律没有结果，error 是 TRUNCATED_ERROR。 + masked_values：查询时在工具边界遮掉的值（engine/masking 的 _Secrets）。只在内存里交给调用方按值遮（审批质询的回答）， + 不进 as_dict、不往外交。 """ result: dict[str, Any] | None @@ -162,9 +174,12 @@ class VerifyOutcome: answered: bool = False notes: list[str] = field(default_factory=list) duration_ms: int = 0 + masked_values: Any = field(default=None, repr=False, compare=False) def as_dict(self) -> dict[str, Any]: - return asdict(self) + out = asdict(replace(self, masked_values=None)) + out.pop("masked_values", None) + return out # -------------------------------------------------------------------------- @@ -407,6 +422,12 @@ class _Cells: self.cells.append({"path": path, "status": "unresolved", "alias": alias, "artifact": query["artifact"], "reason": f"{alias} {e}"}) return None + if str(name).lower() in _masked_at_query(content): + # 查询时就遮了的列：格子里是「已遮罩」，不是数据源里的值。照实说是遮罩列，不说成「不是数值」 + self.cells.append({"path": path, "status": "unresolved", "ref": f"{alias}.r{row}.{name}", "alias": alias, + "row": row, "column": name, "artifact": query["artifact"], + "reason": f"{alias} 的「{name}」取自设置了遮罩的列，查询时已遮罩，不采信"}) + return None truth = _typed(raw, {"type": sorted(types)}, column_kind(content, name)) cited = {"ref": f"{alias}.r{row}.{name}", "alias": alias, "row": row, "column": name, "artifact": query["artifact"], "eid": cell_eid(query["artifact"], row, name), "value": truth} @@ -502,6 +523,12 @@ class _Cells: return None +def _masked_at_query(content: Any) -> set[str]: + """查询快照里查询时就整列遮掉的列（小写，tools/datasource 记下的 masked_at_query）。""" + listed = content.get("masked_at_query") if isinstance(content, dict) else None + return {c.lower() for c in listed if isinstance(c, str)} if isinstance(listed, list) else set() + + # -------------------------------------------------------------------------- # 花费 # -------------------------------------------------------------------------- @@ -636,6 +663,8 @@ class _State: queries: list[dict[str, Any]] = field(default_factory=list) ledger: list[dict[str, Any]] = field(default_factory=list) model: str | None = None + #: 查询工具在工具边界遮掉的值（经 ToolContext.extra 并进来，只在内存里） + seen: _Secrets = field(default_factory=_Secrets) def outcome(self, **fields: Any) -> VerifyOutcome: usage = self.usage @@ -645,7 +674,8 @@ class _State: for q in self.queries] base: dict[str, Any] = {"result": None, "cells": [], "queries": shown, "rationale": None, "usage": usage, "notes": self.notes, "model": self.model, - "duration_ms": int((time.monotonic() - self.started) * 1000)} + "duration_ms": int((time.monotonic() - self.started) * 1000), + "masked_values": self.seen} return VerifyOutcome(**{**base, **fields}) @@ -728,12 +758,13 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou notes.append(note) await _announce(emit, EventType.LOG, level="warn", code="verify_unpriced", message=note) - # ---- 工具：只读的 db_schema / db_query + # ---- 工具：只读的 db_schema / db_query，查询结果先遮再交给模型（遮掉的值并进 state.seen） wanted = list(dict.fromkeys(s for s in sources if isinstance(s, str) and s)) - ctx = ToolContext(run_id=run_id, node_id=node_id, data_versions=data_versions, emit=emit) + ctx = ToolContext(run_id=run_id, node_id=node_id, data_versions=data_versions, emit=emit, + extra={MASKED_VALUES_KEY: state.seen}) async with SessionLocal() as session: tools = await build_datasource_tools([f"{p}{s}" for s in wanted for p in (SCHEMA_PREFIX, QUERY_PREFIX)], ctx, - session, force_readonly=True) + session, force_readonly=True, mask_results=True) tool_map = {t.name: t for t in tools} if lost := [s for s in wanted if f"{QUERY_PREFIX}{s}" not in tool_map]: notes.append("数据源" + "、".join(f"「{s}」" for s in lost) + "不存在或已停用，核查没有查它") diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index 88aa8b03..6374c322 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -53,6 +53,9 @@ RUN_VERSIONS_KEY = "data_versions" DATA_UNAVAILABLE = "查询失败（数据不可用）：" #: 运行中途补固定一个源时发的那条日志事件的 code（_pin_late） LATE_PIN_CODE = "data_version_pinned" +#: ToolContext.extra 里放「查询时遮掉的值」的键（engine/masking 的 _Secrets）：核查 agent（engine/verify_agent.py）放进去， +#: 打开了遮罩开关的查询工具把每次遮掉的值并进去。只在内存里给调用方按值遮用，不进快照、不交给模型 +MASKED_VALUES_KEY = "masked_values" #: 每次查询之后基于数据目录的 SQL 检查最多等这么久。正常一条 SQL 几毫秒；等不到就放弃这一次检查， #: 查询结果照常交回——检查是附加的，不能拖慢查询 @@ -199,7 +202,8 @@ def _query_description(source: Any, *, readonly: bool | None = None) -> str: async def build_datasource_tools( - names: list[str], ctx: ToolContext, session: AsyncSession, *, force_readonly: bool = False + names: list[str], ctx: ToolContext, session: AsyncSession, *, force_readonly: bool = False, + mask_results: bool = False, ) -> list[StructuredTool]: """按工具名反查数据源并构造工具。build_tools 的第四个来源。 @@ -209,6 +213,9 @@ async def build_datasource_tools( force_readonly：查询工具一律按只读守卫检查，可写源上的写语句也拒绝（见 _make_query_tool）。 核查（engine/verify_agent.py）建工具时传它：那一路没有审批关卡，绝不能写。 + + mask_results：查询结果交给模型、落进快照之前先遮（见 _make_query_tool）。核查建工具时也传它：模型看不到遮罩列的 + 值，就不会把它们写进回答、理由。缺省关着：Agent 节点、调用工具节点的查询照旧原样。 """ wanted: dict[str, list[str]] = {} for name in names: @@ -238,7 +245,8 @@ async def build_datasource_tools( entries = await _catalog_of(row) for prefix in prefixes: if prefix == QUERY_PREFIX: - tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries, force_readonly=force_readonly)) + tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries, force_readonly=force_readonly, + mask_results=mask_results)) else: tools.append(_make_schema_tool(source, fixed=fixed, catalog=entries)) return tools @@ -450,7 +458,7 @@ def _unavailable_tool(row: DataSource, prefix: str, reason: str) -> StructuredTo def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, catalog: Mapping[str, data_catalog.CatalogEntry] | None = None, - force_readonly: bool = False) -> StructuredTool: + force_readonly: bool = False, mask_results: bool = False) -> StructuredTool: """source 是手工源的 DataSource，或上传源绑定快照的 SourceView（字段同名）。 fixed：上传源的版本是这次运行固定的（找不到时的说法不同）。 @@ -458,6 +466,8 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, force_readonly：不管源可不可写，一律按只读守卫检查，写语句交回 FORCED_READONLY、不执行。只靠守卫这一道： 连接池按源缓存、不分只读与否（engine.cache_key），可写源借不到连接层只读的连接——别为此另起一个只读池， 缓存键相同，会把这个源平时用的可写连接也换成只读的。 + mask_results：数据源设了遮罩时，取自遮罩列的格子（按 SQL 认，改了名字的也算）在交给模型、落进快照之前就换成 + 「已遮罩」，见 engine/masking.py 的 mask_query_result；这一次遮掉的值并进 ctx.extra[MASKED_VALUES_KEY]（有的话）。 """ # 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句， # 引擎按它告诉界面上限是多少（metadata["timeout_s"]） @@ -505,6 +515,17 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, payload = result.to_payload() payload["source"] = source.name + mask = masked_columns(source.options) + # 查询时就整列遮掉的结果列；没开遮罩开关、数据源没设遮罩时为 None（快照和原来一字不差） + masked_now: list[str] | None = None + if mask_results and mask: + from app.engine.masking import mask_query_result + + # 按 SQL 认列、逐格比对是纯计算，放进线程（同 _sql_checks） + done = await asyncio.to_thread(mask_query_result, payload, mask, source.kind) + payload, masked_now = done.payload, done.columns + if (sink := ctx.extra.get(MASKED_VALUES_KEY)) is not None: + sink.merge(done.secrets) # 对照数据目录检查这条 SQL（一对多关联后重复计算、存量跨期求和……）。结果交给模型（Agent 据此改写）， # 也随查询快照存下：证据面板看得到，口径卡据此给指标标「存疑」、出具降档。没查出问题不加这个键， # 快照和以前一字不差 @@ -518,14 +539,19 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, try: from app.core.artifact_store import put_json - # 只进快照、不进交给模型的结果的两样（模型看到多余的字段只会误会：以为数据被遮了， + # 只进快照、不进交给模型的结果的几样（模型看到多余的字段只会误会：以为数据被遮了， # 或者把一串十六进制抄进报告）： - # - 查询当时数据源设的遮罩列：数据源事后改名、删掉，证据面板仍按当时的遮； + # - 查询当时数据源设的遮罩列：数据源事后改名、删掉，证据面板仍按当时的遮。查询时就遮了的结果列（改了名字的 + # 也算）一并记进去，证据面板、质询工件按这个键把它们画成遮罩列； + # - masked_at_query：开了遮罩开关时，查询时就整列遮掉的结果列（这几列的格子是「已遮罩」，不是数据源里的值）； # - data_version：上传表格查的是哪个快照，事后据此认得出这个数出自哪一版数据。 # 手工源、没设遮罩的，快照和原来一字不差 extra: dict[str, Any] = {} - if mask := masked_columns(source.options): - extra["mask_columns"] = mask + if mask: + own = {m.lower() for m in mask} + extra["mask_columns"] = [*mask, *(c for c in masked_now or [] if c.lower() not in own)] + if masked_now is not None: + extra["masked_at_query"] = masked_now if version := getattr(source, "snapshot_id", None): extra["data_version"] = version # 工件引用的 meta 记下源和 SQL 里用到的表：数据目录按它数每张表被运行查询过几次 diff --git a/backend/tests/test_challenge_node.py b/backend/tests/test_challenge_node.py index 33ab28c4..597d15f7 100644 --- a/backend/tests/test_challenge_node.py +++ b/backend/tests/test_challenge_node.py @@ -1102,6 +1102,32 @@ async def test_the_task_gives_the_claim_its_evidence_and_only_its_sources(monkey await engines.invalidate(source_id) +async def test_the_challenge_agent_never_sees_masked_values(monkeypatch, tmp_path): + """反查 agent 自己的查询选出了遮罩列（起了别名）：交给它的工具结果、它的查询快照里都是「已遮罩」，理由里引用到那一格 + 渲染出来也是「已遮罩」；审稿记录、运行事件里都没有原值。以前查询结果原样交给模型、原样存下，理由也原样渲染。""" + name, source_id = await _source(tmp_path, options={"mask_columns": ["buyer_phone"]}) + try: + top = "SELECT buyer_phone AS 电话, amount AS 金额 FROM sales WHERE region = '华东' ORDER BY amount DESC" + plays = {"居各区域之首": Play([top], "stands", "华东最大一笔来自 [[v:Q1.r0.电话]]，金额 [[v:Q1.r0.金额]]。[[see:Q1]]")} + models = Models(monkeypatch, name, plays=plays) + row = await finish(graph(name), run_class="exploratory") + assert row.status == "succeeded", row.error + seen = [str(m.content) for _, key, messages in models.calls if key == "居各区域之首" + for m in messages if isinstance(m, ToolMessage)] + assert seen and any("已遮罩" in s for s in seen) and all("1390000000" not in s for s in seen) + record = artifact_store.load((await output_of(row.id, "chal"))["challenge_artifact"]) + [claim] = [c for c in record["claims"] if "居各区域之首" in c["text"]] + assert claim["verdict"] == "stands" and claim["rationale"] == "华东最大一笔来自 已遮罩，金额 1,200。" + snapshot = artifact_store.load(claim["queries"][0]["artifact"]) + assert {r[0] for r in snapshot["rows"]} == {"已遮罩"} and snapshot["masked_at_query"] == ["电话"] + assert "1390000000" not in json.dumps(record, ensure_ascii=False) + async with SessionLocal() as session: + evs = (await session.execute(select(RunEvent).where(RunEvent.run_id == row.id))).scalars().all() + assert evs and all("1390000000" not in json.dumps(e.data, ensure_ascii=False) for e in evs) + finally: + await engines.invalidate(source_id) + + async def test_a_metric_excerpt_marks_the_report_query_numbers_too(monkeypatch, shop): """句子挂的是口径卡指标（正式报告的主路径「取数 → 口径卡 → 报告」）：摘录正文里「取自查询 Q1」「查询 Q1 的 SQL」 也标成原报告的编号，和反查 agent 自己的 Q1 分开；工具按指标的输入取自的查询找源。""" diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 8c5587c2..b20aea80 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -1034,6 +1034,27 @@ async def test_a_blind_cell_on_a_masked_column_is_not_used(monkeypatch, masked_s assert "30.25" not in json.dumps(out, ensure_ascii=False) +async def test_a_renamed_masked_column_in_the_blind_query_is_masked_before_the_blind_agent_sees_it( + monkeypatch, masked_shop): + """盲算的查询选出了改了名字的遮罩列（region AS 区域）：交给盲算模型、落进盲算快照之前就是「已遮罩」，快照记下查询 + 时遮了哪一列；没碰遮罩列的那一列（销售额）照常取值、照常比对。""" + name, _ = masked_shop + east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + renamed = (f"SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} " + "GROUP BY region ORDER BY region") + Blind(monkeypatch, name, {"华东销售额": (renamed, "gmv")}) + run = await start(graph(name, BY_REGION, metrics=[east], dual={"definitions": {}})) + assert run.status == "succeeded", run.error + evs = await events(run.id) + out = dual_output(run, evs) + assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed" + [end] = [e.data for e in evs if e.type == "tool.end" and e.node_id == "dual" and e.data.get("query_artifact")] + snapshot = artifact_store.load(end["query_artifact"]) + assert {row[0] for row in snapshot["rows"]} == {"已遮罩"} and snapshot["masked_at_query"] == ["区域"] + shown = json.loads(end["preview"].split("\n", 1)[1]) + assert {row[0] for row in shown["rows"]} == {"已遮罩"}, "交给盲算模型的结果里也是「已遮罩」" + + async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch, masked_shop): name, _ = masked_shop first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} diff --git a/backend/tests/test_inquiry_api.py b/backend/tests/test_inquiry_api.py index 3ce61dd7..e638ec8c 100644 --- a/backend/tests/test_inquiry_api.py +++ b/backend/tests/test_inquiry_api.py @@ -904,6 +904,35 @@ async def test_resuming_after_an_inquiry_does_not_repeat_the_nodes(monkeypatch, ALIASED = "SELECT buyer_phone AS 手机号, amount AS 金额 FROM sales ORDER BY amount DESC LIMIT 1" +async def test_the_inquiry_agent_sees_and_stores_its_queries_masked(monkeypatch, masked_shop, client): + """质询 agent 的查询结果在交给模型之前就遮好了（核查 agent 建查询工具时打开遮罩开关）：模型看到的工具结果、落库的 + 查询快照、运行详情「工件」页签走的通用工件接口里，遮罩列（起了别名的也算）都是「已遮罩」。以前快照原样存着，模型 + 看得到原值，通用接口也原样给出。回答照常按值遮（查询时遮掉的值只在内存里交回来）。""" + plays = {"买得最多": Play([ALIASED], "这一笔来自 [[v:Q1.r0.手机号]]（13900000001），金额 [[v:Q1.r0.金额]]。[[see:Q1]]")} + Models(monkeypatch, masked_shop, plays=plays) + seen: list[str] = [] + answering = MockChatModel._decide + + def spy(model, messages): + seen.extend(str(m.content) for m in messages if isinstance(m, ToolMessage)) + return answering(model, messages) + + monkeypatch.setattr(MockChatModel, "_decide", spy) + paused = await pause(masked_shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question="买得最多的是哪位客户？", + approval_id=approval.id) + assert r.status_code == 200, r.text + assert r.json()["inquiry"]["answer"]["text"] == "这一笔来自 已遮罩（已遮罩），金额 1,200。" + [query] = r.json()["inquiry"]["queries"] + assert seen and any("已遮罩" in s for s in seen) and all("1390000000" not in s for s in seen) + got = await client.get(f"/api/artifacts/{query['artifact']}") + assert got.status_code == 200 and "1390000000" not in got.text + content = got.json()["content"] + assert content["rows"] == [["已遮罩", 1200]] and content["masked_at_query"] == ["手机号"] + assert content == artifact_store.load(query["artifact"]) + + async def test_a_masked_column_under_another_name_is_masked_everywhere(monkeypatch, masked_shop, client): """遮罩列起了中文别名（本仓库查询的惯例）：回答里引用到的格子、事件、回答文档、质询工件里都不露出原值。 报告自己的查询没选手机号，原值哪里都认不出来，只能按 SQL 认出「手机号」就是 buyer_phone。""" diff --git a/backend/tests/test_inquiry_masking.py b/backend/tests/test_inquiry_masking.py index 513a375c..323c0548 100644 --- a/backend/tests/test_inquiry_masking.py +++ b/backend/tests/test_inquiry_masking.py @@ -1197,6 +1197,27 @@ def test_sql_check_excerpts_in_the_artifact_are_masked_like_the_sql(): "sql_excerpt": "SUM(balance)"}] +def test_check_excerpts_follow_the_sql_recorded_at_answer_time(): + """agent 自己的查询快照查询时就遮好了，事后取质询工件时收不到遮罩列里的值：片段、说明照事件里记下的那条 SQL（回答时 + 按内存里的遮罩值换过）换——那条里整个换成「已遮罩」的字面量、换掉的那一截（LIKE 的 %…%），片段、说明里一样换。""" + sql = ("SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '13900000001' AND age = 25 " + "AND note LIKE '%13900000002%'") + recorded = "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '已遮罩' AND age = 已遮罩 AND note LIKE '%已遮罩%'" + assert inquiry_mod._recorded_swaps(sql, recorded, "sqlite") == ({"13900000001", "25"}, {"13900000002"}) + assert inquiry_mod._recorded_swaps(sql, sql, "sqlite") == (set(), set()) + assert inquiry_mod._recorded_swaps(sql, recorded + " LIMIT 5", "sqlite") == (set(), set()), "对不上就不认" + checks = [{"code": "c1", "level": "warning", "table": "sales", "message": "没有码值「25」", + "sql_excerpt": "buyer_phone = '13900000001' AND age = 25"}, + {"code": "c2", "level": "info", "table": "sales", "message": "按备注筛选", + "sql_excerpt": "note LIKE '%13900000002%'"}] + whole, pieces = inquiry_mod._recorded_swaps(sql, recorded, "sqlite") + assert inquiry_mod._masked_checks(checks, sql, inquiry_mod._Secrets(), set(), "sqlite", swapped=whole, + pieces=pieces) == [ + {"code": "c1", "level": "warning", "table": "sales", "message": "没有码值「已遮罩」", + "sql_excerpt": "buyer_phone = '已遮罩' AND age = 已遮罩"}, + {"code": "c2", "level": "info", "table": "sales", "message": "按备注筛选", "sql_excerpt": "note LIKE '%已遮罩%'"}] + + @pytest.mark.parametrize("sql, excerpt, shown", [ # 片段只是一段表达式，单独拿出来认不出在找哪一个值：按整条 SQL 里换掉的字面量换 ("SELECT SUM(CASE age WHEN 25 THEN balance END) AS 余额 FROM accounts", diff --git a/backend/tests/test_masking_reformatted_numbers.py b/backend/tests/test_masking_reformatted_numbers.py index 44d9076b..cb3c0306 100644 --- a/backend/tests/test_masking_reformatted_numbers.py +++ b/backend/tests/test_masking_reformatted_numbers.py @@ -59,6 +59,17 @@ def test_a_long_number_in_another_shape_is_masked_in_prose(stored, written): assert secrets.scrub(f"客户{written}来电", prose=True) == f"客户{MASKED}来电" +@pytest.mark.parametrize("text, shown", [ + ("来自（19914742785），共 3 笔", f"来自（{MASKED}），共 3 笔"), + ("来自(+86 199 1474 2785)，共 3 笔", f"来自({MASKED})，共 3 笔"), + ("电话（+86）199 1474 2785。", f"电话{MASKED}。"), + ("（电话 199-1474-2785）", f"（电话 {MASKED}）"), +]) +def test_brackets_around_a_number_are_kept_and_brackets_inside_it_are_replaced(text, shown): + """整个号码外面套着的一对括号是标点，留着；号码里的括号（区号、国家码）和号码一起换掉。""" + assert secrets_of([PHONE]).scrub(text, prose=True) == shown + + @pytest.mark.parametrize("text", [ # 日期、时间、小数、千分位、更长的数里的一截、连字符连着的编号、别的号码 "2026-10-06 19:14 下单", "金额 19914.74 元", "共 1,991,474 笔", "号码 1991474278512", "批次 19914742785-2", diff --git a/backend/tests/test_verify_agent_masking.py b/backend/tests/test_verify_agent_masking.py new file mode 100644 index 00000000..d2f93c66 --- /dev/null +++ b/backend/tests/test_verify_agent_masking.py @@ -0,0 +1,248 @@ +"""核查 agent 看到的查询结果先遮、存下的快照也遮（engine/verify_agent.py 建查询工具时打开 tools/datasource 的遮罩开关）。 + +双盲复算、反方审稿、审批质询共用的核查 agent 以前把查询结果原样交给模型，遮罩是事后按值做的（审批质询的 _Secrets）， +每一轮复核都找到新的漏法。现在在工具边界先遮：数据源设了遮罩时，按 SQL 认出取自遮罩列的结果列（别名、函数、子查询、 +WITH、SELECT *），连同列名直接对上的，这些格子在交给模型的结果、落库的快照里都是「已遮罩」，快照记下哪些列在查询时 +就遮了；别的文字格里夹着这份结果里遮罩值的那一截同样换掉（换了写法的号码也认）。运行详情「工件」页签走的通用接口 +GET /api/artifacts/{id} 给的也就是遮好的。没碰遮罩列的聚合查询原样。遮掉的值只在内存里交回调用方（审批质询按值遮的 +第二道防线用），不进快照、不交给模型。遮罩开关缺省关着，别的调用方（Agent 节点、调用工具节点）照旧。 + +模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，手机号、姓名、金额都是虚构的。 +""" +from __future__ import annotations + +import json +import sqlite3 +import uuid + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete + +from app.core import artifact_store +from app.core.events import EventType +from app.data import introspect +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource, Setting +from app.engine import masking, spend +from app.engine.verify_agent import run_verify_agent +from app.main import app +from app.tools.datasource import build_datasource_tools +from app.tools.registry import ToolContext, Versions +from tests.test_verify_agent import SPEC, Script, q + +MASKED = "已遮罩" +#: 遮罩列里的值（和备注里换了写法的号码）：交给模型的、存下的、接口给的都不能出现 +SECRETS = ("1390000000", "张伟", "李娜", "王芳", "刘洋", "139-0000-0003") +SALES = ("CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, buyer_phone TEXT, buyer_name TEXT, amount INTEGER," + " note TEXT);" + "INSERT INTO sales VALUES (1, '华东', '13900000001', '张伟', 1200, '客户13900000001来电改址')," + " (2, '华东', '13900000002', '李娜', 300, '例行回访'), (3, '华南', '13900000003', '王芳', 900, '回访：联系 139-0000-0003')," + " (4, '华北', '13900000004', '刘洋', 400, '');") +VALUE_SCHEMA = {"type": "object", "properties": {"value": {"type": "number", "description": "要核对的数"}}, + "required": ["value"]} + + +@pytest.fixture(autouse=True) +async def clean_spend(): + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_([spend.SPEND_KEY, "verify"]))) + await session.commit() + yield + + +async def _source(tmp_path, options: dict | None) -> tuple[str, int]: + path = tmp_path / f"{uuid.uuid4().hex[:8]}.db" + db = sqlite3.connect(path) + db.executescript(SALES) + db.commit() + db.close() + name = f"shop_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, options=options or {}) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + return name, row.id + + +@pytest.fixture +async def masked_shop(tmp_path): + """四笔订单，买家手机号、姓名两列设了遮罩；备注没设遮罩，其中两条写着手机号（一条换了写法）。返回源名。""" + name, source_id = await _source(tmp_path, {"mask_columns": "buyer_phone, buyer_name"}) + yield name + await engines.invalidate(source_id) + + +@pytest.fixture +async def plain_shop(tmp_path): + name, source_id = await _source(tmp_path, None) + yield name + await engines.invalidate(source_id) + + +async def ask(name: str, **kwargs): + """按剧本跑一次核查 agent（剧本由各个测试先用 Script 装好），交回结果和发出的事件。""" + events: list[tuple] = [] + out = await run_verify_agent(spec=SPEC, system="核对一个数。", task="查一下。", sources=[name], + output_schema=VALUE_SCHEMA, data_versions=Versions.CURRENT, run_id="run-mask", + node_id="check", emit=lambda kind, **data: events.append((kind, data)), **kwargs) + return out, events + + +def _clean(text: str) -> bool: + return not any(s in text for s in SECRETS) + + +def _cells(snapshot: dict, column: str) -> list: + i = snapshot["columns"].index(column) + return [row[i] for row in snapshot["rows"]] + + +@pytest.mark.parametrize("sql, masked", [ + pytest.param("SELECT buyer_phone, amount FROM sales ORDER BY id", ["buyer_phone"], id="直接选出"), + pytest.param("SELECT buyer_phone AS 手机号, amount AS 金额 FROM sales ORDER BY id", ["手机号"], id="起别名"), + pytest.param("SELECT UPPER(buyer_name) AS 姓名, amount FROM sales ORDER BY id", ["姓名"], id="套函数"), + pytest.param("SELECT t.p AS 号码, t.amount FROM (SELECT id, buyer_phone AS p, amount FROM sales) t ORDER BY t.id", + ["号码"], id="子查询"), + pytest.param("WITH t AS (SELECT id, buyer_phone AS p, amount FROM sales) SELECT p, amount FROM t ORDER BY id", + ["p"], id="WITH"), + pytest.param("SELECT * FROM sales ORDER BY id", ["buyer_phone", "buyer_name"], id="SELECT *"), +]) +async def test_masked_columns_are_masked_before_the_model_sees_them(monkeypatch, masked_shop, sql, masked): + name = masked_shop + amount = "amount" if "金额" not in sql else "金额" + script = Script(monkeypatch, [q(name, sql, "c1")], + {"value": {"alias": "Q1", "row": 0, "column": amount}, "rationale": "见 [[see:Q1]]。"}) + out, events = await ask(name) + assert out.error is None and out.result["value"] == 1200, "没遮的列照常取值" + + # 交给模型的工具结果：遮罩列是「已遮罩」，别的列照原样 + seen = script.tool_results()["c1"] + assert seen.startswith("【证据 Q1】") and MASKED in seen and _clean(seen), seen + payload = json.loads(seen.split("\n", 1)[1]) + for column in masked: + assert set(_cells(payload, column)) == {MASKED} + assert _cells(payload, amount) == [1200, 300, 900, 400] + assert "mask_columns" not in payload and "masked_at_query" not in payload, "交给模型的结果不多出字段" + + # 落库的快照：同样遮好，记下查询时就遮了的列；数据源设的遮罩列照记，换了名字的结果列也记进去（证据面板据此画成遮罩） + [query] = out.queries + snapshot = artifact_store.load(query["artifact"]) + assert _clean(json.dumps(snapshot, ensure_ascii=False)) + for column in masked: + assert set(_cells(snapshot, column)) == {MASKED} + assert snapshot.get("column_types", {}).get(column, "text") == "text" + assert _cells(snapshot, amount) == [1200, 300, 900, 400] + assert snapshot["masked_at_query"] == masked + assert {c.lower() for c in snapshot["mask_columns"]} == {"buyer_phone", "buyer_name", *(c.lower() for c in masked)} + + # 运行详情「工件」页签走的通用接口：给的就是遮好的 + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + got = await client.get(f"/api/artifacts/{query['artifact']}") + assert got.status_code == 200 and got.json()["content"] == snapshot and _clean(got.text) + + # 事件（tool.end 的预览）、收尾时交给模型的查询结果里也没有 + [end] = [d for k, d in events if k == EventType.TOOL_END] + assert MASKED in end["preview"] and _clean(end["preview"]) + final = "\n".join(str(m.content) for m in script.calls[-1][1]) + assert MASKED in final and _clean(final) + + # 遮掉的值只在内存里交回调用方，不跟着结果往外交 + assert out.masked_values is not None and out.masked_values.strong + assert "masked_values" not in out.as_dict() + + +@pytest.mark.parametrize("sql, value", [ + ("SELECT SUM(amount) AS total FROM sales", 2800), + ("SELECT COUNT(DISTINCT buyer_phone) AS total FROM sales", 4), + ("SELECT SUM(amount) AS total FROM sales WHERE buyer_phone = '13900000001'", 1200), + ("SELECT region, SUM(amount) AS total FROM sales GROUP BY region ORDER BY total DESC", 1500), + ("SELECT region, note, amount AS total FROM sales ORDER BY id", 1200), +]) +async def test_queries_that_do_not_read_masked_values_are_untouched(monkeypatch, masked_shop, sql, value): + """没碰遮罩列的值（汇总、只数个数、只在条件里用到）：结果原样，取值照常，快照记下查询时一列都没遮。只遮这份结果 + 里的遮罩值：备注里写着手机号、这份结果却没有选出手机号的，按值认不出（那是数据源的遮罩没设全）。""" + name = masked_shop + script = Script(monkeypatch, [q(name, sql, "c1")], + {"value": {"alias": "Q1", "row": 0, "column": "total"}, "rationale": "见 [[see:Q1]]。"}) + out, _ = await ask(name) + assert out.error is None and out.result["value"] == value + seen = script.tool_results()["c1"] + assert MASKED not in seen + snapshot = artifact_store.load(out.queries[0]["artifact"]) + assert snapshot["masked_at_query"] == [] and MASKED not in json.dumps(snapshot, ensure_ascii=False) + assert [c.lower() for c in snapshot["mask_columns"]] == ["buyer_phone", "buyer_name"] + + +async def test_masked_values_inside_other_text_cells_are_cut_out(monkeypatch, masked_shop): + """同一份结果里选出了手机号：备注里夹着的同一个号码（换了写法的也算）换成「已遮罩」，备注别的字照原样。""" + name = masked_shop + sql = "SELECT buyer_phone, note, amount AS total FROM sales ORDER BY id" + script = Script(monkeypatch, [q(name, sql, "c1")], + {"value": {"alias": "Q1", "row": 0, "column": "total"}, "rationale": "见 [[see:Q1]]。"}) + out, _ = await ask(name) + seen = script.tool_results()["c1"] + assert _clean(seen) + snapshot = artifact_store.load(out.queries[0]["artifact"]) + assert _cells(snapshot, "note") == [f"客户{MASKED}来电改址", "例行回访", f"回访：联系 {MASKED}", ""] + assert snapshot["masked_at_query"] == ["buyer_phone"], "备注只换了一截，不算整列遮" + + +async def test_a_cell_reference_into_a_masked_column_is_not_taken(monkeypatch, masked_shop): + """模型交的格子引用指到查询时就遮了的那一列：不采信，照实说是遮罩列，不说成「不是数值」。""" + name = masked_shop + sql = "SELECT buyer_phone AS 手机号, amount FROM sales ORDER BY id" + Script(monkeypatch, [q(name, sql, "c1")], {"value": {"alias": "Q1", "row": 0, "column": "手机号"}, "rationale": "x"}) + out, _ = await ask(name) + assert out.result["value"] is None + [cell] = out.cells + assert cell["status"] == "unresolved" and "1390000000" not in json.dumps(cell) + assert cell["reason"] == "Q1 的「手机号」取自设置了遮罩的列，查询时已遮罩，不采信" + + +async def test_sql_that_cannot_be_read_but_names_a_masked_column_masks_the_whole_result(monkeypatch, masked_shop): + """SQL 认不出哪一列取自遮罩列、里面又写着遮罩列：整份结果的列都遮（和双盲复算、审批质询同一个保守规则）。""" + name = masked_shop + monkeypatch.setattr(masking, "_tainted", lambda *args, **kwargs: None) + sql = "SELECT buyer_phone, amount FROM sales ORDER BY id" + Script(monkeypatch, [q(name, sql, "c1")], {"value": {"alias": "Q1", "row": 0, "column": "amount"}, "rationale": "x"}) + out, _ = await ask(name) + snapshot = artifact_store.load(out.queries[0]["artifact"]) + assert snapshot["masked_at_query"] == ["buyer_phone", "amount"] + assert set(_cells(snapshot, "amount")) == {MASKED} and out.result["value"] is None + + +async def test_the_switch_is_off_by_default_and_a_source_without_masks_is_untouched(masked_shop, plain_shop): + """遮罩开关缺省关着：Agent 节点、调用工具节点建的查询工具照旧交回原样的结果。没设遮罩的数据源开着开关也一字不差 + （快照里不多出键）。""" + sql = "SELECT buyer_phone, amount FROM sales ORDER BY id" + ctx = ToolContext(run_id="run-switch", node_id="n1", data_versions=Versions.CURRENT) + async with SessionLocal() as session: + [plain_tool] = await build_datasource_tools([f"db_query__{masked_shop}"], ctx, session) + [masking_tool] = await build_datasource_tools([f"db_query__{masked_shop}"], ctx, session, mask_results=True) + [unmasked_tool] = await build_datasource_tools([f"db_query__{plain_shop}"], ctx, session, mask_results=True) + raw = json.loads(await plain_tool.ainvoke({"sql": sql})) + assert _cells(raw, "buyer_phone")[0] == "13900000001" + stored = artifact_store.load(raw["artifact"]) + assert _cells(stored, "buyer_phone")[0] == "13900000001" and "masked_at_query" not in stored + shown = json.loads(await masking_tool.ainvoke({"sql": sql})) + assert set(_cells(shown, "buyer_phone")) == {MASKED} + untouched = json.loads(await unmasked_tool.ainvoke({"sql": sql})) + stored = artifact_store.load(untouched["artifact"]) + assert _cells(stored, "buyer_phone")[0] == "13900000001" + assert "masked_at_query" not in stored and "mask_columns" not in stored + + +async def test_column_names_the_model_wrote_are_kept(monkeypatch, masked_shop): + """列名是模型自己写在 SQL 里的（哪怕写着一个号码）：工具边界只遮格子、不改列名，它引用的列名才对得上。""" + name = masked_shop + sql = 'SELECT buyer_phone, amount AS "13900000001 的金额" FROM sales ORDER BY id' + Script(monkeypatch, [q(name, sql, "c1")], + {"value": {"alias": "Q1", "row": 0, "column": "13900000001 的金额"}, "rationale": "x"}) + out, _ = await ask(name) + snapshot = artifact_store.load(out.queries[0]["artifact"]) + assert snapshot["columns"] == ["buyer_phone", "13900000001 的金额"] + assert set(_cells(snapshot, "buyer_phone")) == {MASKED} and out.result["value"] == 1200