From e89ea9becf411148a680c05d2b539d93cb6a0ef7 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Thu, 3 Sep 2026 13:01:43 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=204=20=E4=B8=AA=20P1?= =?UTF-8?q?=20issue=20(#224/#232/#223/#215)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - #224 screener 自定义 SQL 的内存连接关闭 enable_external_access, 注入的 read_parquet/COPY 文件读写直接报错 (安全) - #232 指数展示缓存百分数口径在消费边界显式 /100: pipeline._bench_rt_pct_of 与 abnormal_moves._bench_rt_pct 两处, 修复 3/10/30 日偏离值被放大两个数量级 - #223 盘后管道按同日 daily/enriched 行数比较检测实时合并提前 创建的部分分区, 删除后由增量重算全市场补齐 - #215 _basic_filter_for_asset 扩展中和股票专属键 (price_min/max/boards), 并应用到回测/挖掘/策略扫描三个运行期入口, 修复 ETF 静默零信号 --- backend/app/backtest/strategy.py | 35 ++++-- backend/app/indicators/pipeline.py | 9 +- backend/app/jobs/daily_pipeline.py | 47 ++++++++ backend/app/services/abnormal_moves.py | 9 +- backend/app/services/screener.py | 6 +- backend/app/strategy/engine.py | 5 + backend/tests/test_abnormal_moves.py | 35 +++++- .../tests/test_enriched_partial_partition.py | 62 +++++++++++ backend/tests/test_matrix_etf_share_fields.py | 27 +++++ .../tests/test_screener_external_access.py | 101 ++++++++++++++++++ 10 files changed, 321 insertions(+), 15 deletions(-) create mode 100644 backend/tests/test_enriched_partial_partition.py create mode 100644 backend/tests/test_screener_external_access.py diff --git a/backend/app/backtest/strategy.py b/backend/app/backtest/strategy.py index 7540c20..6b1b99a 100644 --- a/backend/app/backtest/strategy.py +++ b/backend/app/backtest/strategy.py @@ -497,20 +497,32 @@ _TURNOVER_FILTER_KEYS = ( "turnover_max", ) +# 股票专属的价格界与板块过滤对非股票资产同样不可满足 (#215): +# ETF 单价普遍 0.5~7 元, 会被 price_min=3 整列误杀; boards 按股票代码 +# 前缀匹配, ETF 代码不属于任何板块 → 掩码全 False, 静默零信号。 +_STOCK_ONLY_FILTER_KEYS = ( + *_SHARE_CAP_FILTER_KEYS, + *_TURNOVER_FILTER_KEYS, + "price_min", + "price_max", + "boards", +) + def _basic_filter_for_asset(basic_filter: dict, asset_type: str) -> dict: """非股票资产没有股本数据 (etf/index 维表只有 symbol/name), 市值、流通 - 市值与换手率界对它们既无意义也不可满足: 依赖解析前先置 None, 避免解析出 - total_shares/float_shares/turnover_rate 字段需求导致矩阵加载直接失败。 + 市值与换手率界对它们既无意义也不可满足: 依赖解析与运行期过滤前先置 + None。价格界 (price_min/max) 与板块过滤 (boards) 是股票专属口径, 对 + ETF 同样不可满足, 一并中和, 否则入场候选在运行期被静默清零 (#215)。 - 运行期过滤无需同步修改 —— polars 侧有列守卫 (engine._basic_filter_expr), - 矩阵侧 _optional_field 对缺失字段返回全 NaN 且 _apply_bound 跳过全 NaN - 界, 二者对缺失股本/换手率列本就降级为 no-op。 + 置 None 后: 依赖解析不再产出 total_shares/float_shares/turnover_rate + 需求; polars 侧有列守卫 (engine._basic_filter_expr), 矩阵侧 + _optional_field 对缺失字段返回全 NaN 且 _apply_bound 跳过全 NaN 界。 """ if asset_type == "stock" or not basic_filter: return basic_filter sanitized = dict(basic_filter) - for key in (*_SHARE_CAP_FILTER_KEYS, *_TURNOVER_FILTER_KEYS): + for key in _STOCK_ONLY_FILTER_KEYS: sanitized[key] = None return sanitized @@ -844,7 +856,11 @@ class StrategyBacktestService: ) overrides = first.overrides or {} - basic_filter = self._effective_basic_filter(strategy, overrides) + # 运行期过滤用的也是同一份 basic_filter: 在入口处按资产类型中和, + # 否则 boards/price_min 会在掩码阶段静默清零 ETF 候选 (#215) + basic_filter = _basic_filter_for_asset( + self._effective_basic_filter(strategy, overrides), first.asset_type + ) entry_signals = self._effective_signals(overrides, "entry_signals", strategy.entry_signals) exit_signals = self._effective_signals(overrides, "exit_signals", strategy.exit_signals) resolver = StrategyDependencyResolver() @@ -1021,7 +1037,10 @@ class StrategyBacktestService: params = self._normalize_params(config.params or {}, s) overrides = config.overrides or {} - basic_filter = self._effective_basic_filter(s, overrides) + # 同回测 run 路径: 挖掘运行期也要按资产类型中和股票专属过滤键 (#215) + basic_filter = _basic_filter_for_asset( + self._effective_basic_filter(s, overrides), config.asset_type + ) entry_signals = self._effective_signals(overrides, "entry_signals", s.entry_signals) exit_signals = self._effective_signals(overrides, "exit_signals", s.exit_signals) if config.exit_fill == "signal_next_minute": diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index d625c1f..143e32a 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -1181,7 +1181,12 @@ def attach_deviation_columns(df: pl.DataFrame, data_dir: Path) -> pl.DataFrame: def _bench_rt_pct_of(index_quotes: pl.DataFrame | None, candidates: list[str]) -> float: - """从实时指数行情取某交易所首选基准的今日涨跌, 缺数据时 0。""" + """从实时指数行情取某交易所首选基准的今日涨跌 (小数制), 缺数据时 0。 + + 入参 index_quotes 来自 quote_service 的指数展示缓存, 其 change_pct/pct/pct_change + 列为百分数口径 (CONTRIBUTING §3.1), 消费前必须显式 /100 (#232); + close/prev_close 兜底路径本身就是小数, 不转换。 + """ if index_quotes is None or index_quotes.is_empty(): return 0.0 df = index_quotes.filter(pl.col("symbol").is_in(candidates)) @@ -1196,7 +1201,7 @@ def _bench_rt_pct_of(index_quotes: pl.DataFrame | None, candidates: list[str]) - for col in ("change_pct", "pct", "pct_change"): v = row.get(col) if v is not None: - return float(v) + return float(v) / 100.0 if row.get("close") is not None and row.get("prev_close") is not None and row["prev_close"]: return float(row["close"] / row["prev_close"] - 1) return 0.0 diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index 3dc47f5..f50b87f 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -31,6 +31,41 @@ logger = logging.getLogger(__name__) ProgressCb = Callable[..., None] +def _prune_partial_enriched_partitions(daily_dir: Path, enriched_dir: Path) -> list[str]: + """删除 symbol 覆盖不完整的 enriched 日期分区, 返回被删的日期 (#223)。 + + 自选实时路径会在全市场 enriched 生成前提前创建当日分区 (只有几只自选), + 仅按日期目录计数比较会把它误判为完整分区而跳过计算, 造成日K缺失与 + 均线错误。同日 enriched 行数 < daily 行数即判定为部分分区: 删除后 + run_pipeline(new_dates_only=True) 会把它们当"新日期"全市场补齐, 与 + data_integrity.prune_enriched_partitions 的修复语义一致。 + daily 同日分区不存在 (今日日K尚未同步) 时不处理, 留给当日正常流程。 + """ + import shutil + + import pyarrow.parquet as pq + + def _rows(part_dir: Path) -> int: + total = 0 + for f in part_dir.glob("*.parquet"): + try: + total += pq.ParquetFile(f).metadata.num_rows + except Exception: # noqa: BLE001 + return -1 # 不可读 → 不动, 交给既有完整性检查兜底 + return total + + pruned: list[str] = [] + for part in enriched_dir.glob("date=*"): + daily_part = daily_dir / part.stem + if not daily_part.exists(): + continue + e_rows, d_rows = _rows(part), _rows(daily_part) + if e_rows >= 0 and d_rows > 0 and e_rows < d_rows: + shutil.rmtree(part, ignore_errors=True) + pruned.append(part.stem.split("=")[1]) + return pruned + + class PipelineStageError(RuntimeError): """管道有阶段软失败(数据可能陈旧)时抛出, 让上层 job_store 把任务标记为 failed。 @@ -353,6 +388,18 @@ def run_now( daily_days = len(list(daily_dir.glob("date=*"))) if daily_dir.exists() else 0 prev_enriched_days = len(list(enriched_dir.glob("date=*"))) if enriched_exists else 0 + # 部分分区修复 (#223): 删除被实时合并提前创建、覆盖不全的 enriched 分区, + # 让下方计数比较与增量计算把它们重新当新日期处理 + if enriched_exists: + partial_pruned = _prune_partial_enriched_partitions(daily_dir, enriched_dir) + if partial_pruned: + logger.warning( + "compute_enriched: 发现 %d 个覆盖不全的 enriched 分区, 已删除待重算: %s", + len(partial_pruned), ", ".join(sorted(partial_pruned)[:10]), + ) + enriched_exists = enriched_dir.exists() and any(enriched_dir.glob("date=*")) + prev_enriched_days = len(list(enriched_dir.glob("date=*"))) if enriched_exists else 0 + # 判断新日期方向: 找 daily 和 enriched 的日期集合做比较 forward_incremental = False backward_extension = False diff --git a/backend/app/services/abnormal_moves.py b/backend/app/services/abnormal_moves.py index 7e583b1..582dc64 100644 --- a/backend/app/services/abnormal_moves.py +++ b/backend/app/services/abnormal_moves.py @@ -134,7 +134,12 @@ def _hist_snapshot(repo: Any) -> dict[str, Any]: def _bench_rt_pct(quote_service: Any) -> float: - """基准指数今日实时涨跌 (各候选均值, 缺数据时 0)。""" + """基准指数今日实时涨跌 (各候选均值, 小数制, 缺数据时 0)。 + + quote_service.get_index_quotes() 返回指数展示缓存, change_pct/pct/pct_change + 为百分数口径 (CONTRIBUTING §3.1), 消费前显式 /100, 与 enriched 侧小数制 + change_pct 对齐 (#232); close/prev_close 兜底路径本身是小数, 不转换。 + """ try: df = quote_service.get_index_quotes() except Exception: @@ -148,7 +153,7 @@ def _bench_rt_pct(quote_service: Any) -> float: if col in df.columns: vals = df[col].drop_nulls() if vals.len() > 0: - return float(vals.mean()) + return float(vals.mean()) / 100.0 if {"close", "prev_close"} <= set(df.columns): sub = df.select(["close", "prev_close"]).drop_nulls() if sub.height > 0: diff --git a/backend/app/services/screener.py b/backend/app/services/screener.py index c11cf64..9e1ae0e 100644 --- a/backend/app/services/screener.py +++ b/backend/app/services/screener.py @@ -335,10 +335,14 @@ class ScreenerService: # 用独立的 :memory: 连接 (而非复用 repo 共享连接的 cursor): conditions 是用户 # 传入的 SQL 片段, 隔离连接下注入至多能碰 read_csv/read_parquet 文件; 若复用共享 # 连接则会把 app 已注册的真实业务表也暴露给注入, 扩大攻击面。隔离连接创建开销极低。 + # 再关闭 external_access, 让注入的文件读写函数 (read_parquet/COPY 等) 直接报错, + # 视图数据仍通过 con.register 注入, 不受该开关影响 (#224)。 con = None try: import duckdb - con = duckdb.connect(database=":memory:") + con = duckdb.connect( + database=":memory:", config={"enable_external_access": False} + ) con.register("enriched", df.to_arrow()) where = " AND ".join(f"({c})" for c in conditions) sql = f"SELECT * FROM enriched WHERE {where}" diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index 517e897..0636f98 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -1244,6 +1244,11 @@ class StrategyEngine: basic_filter = dict(strategy.basic_filter or {}) if overrides.get("basic_filter"): basic_filter.update(overrides["basic_filter"]) + # 策略扫描的运行期过滤同样要按资产类型中和股票专属键 (boards/价格界), + # 否则 ETF 候选在矩阵掩码阶段被静默清零 (#215); 函数级导入避免 + # engine ↔ backtest.strategy 的模块级循环依赖 (与上方 matrix 导入同模式) + from app.backtest.strategy import _basic_filter_for_asset + basic_filter = _basic_filter_for_asset(basic_filter, context.asset_type) scoring = effective_scoring(strategy.meta.get("scoring"), overrides) asset_mask = None if pool: diff --git a/backend/tests/test_abnormal_moves.py b/backend/tests/test_abnormal_moves.py index 521de28..6c1d079 100644 --- a/backend/tests/test_abnormal_moves.py +++ b/backend/tests/test_abnormal_moves.py @@ -12,6 +12,7 @@ from app.indicators.pipeline import ( load_benchmark_momentum, ) from app.services.abnormal_moves import ( + _bench_rt_pct, _hist_cache, _hist_cache_lock, board_of, @@ -87,7 +88,9 @@ def _write_sh_bench(tmp_path) -> None: def test_benchmark_momentum_today_math(tmp_path) -> None: _write_sh_bench(tmp_path) - quotes = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [0.10]}) + # 指数展示缓存为百分数口径 (quote_service._build_index_quotes 已 x100), + # 10.0 表示 +10%; 早期测试直接传小数 0.10 绕过了该契约, 未覆盖 #232 + quotes = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [10.0]}) out = benchmark_momentum_today(tmp_path, quotes) row = out.row(0, named=True) @@ -103,6 +106,18 @@ def test_benchmark_momentum_today_math(tmp_path) -> None: assert abs(out0.row(0, named=True)["bench_mom3d"] - (15.0 / 13 - 1)) < 1e-9 +def test_benchmark_momentum_today_percent_not_treated_as_decimal(tmp_path) -> None: + """#232 回归: 百分数 -1.88 (实际 -1.88%) 不得被当小数 (否则 1+rt=-0.88, + 构造出负的指数点位, 偏离值被放大两个数量级)。""" + _write_sh_bench(tmp_path) + quotes = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [-1.88]}) + + out = benchmark_momentum_today(tmp_path, quotes) + row = out.row(0, named=True) + # 今收 = 15 x (1 - 0.0188) = 14.718; mom3d = 14.718/13 - 1 + assert abs(row["bench_mom3d"] - (15.0 * 0.9812 / 13 - 1)) < 1e-9 + + def test_benchmark_momentum_today_excludes_today_rows(tmp_path) -> None: # 指数监控盘写入的今日行不能当昨收 (否则实时涨跌被重复叠加) today = date.today() @@ -116,7 +131,8 @@ def test_benchmark_momentum_today_excludes_today_rows(tmp_path) -> None: def test_attach_deviation_columns_today(tmp_path) -> None: _write_sh_bench(tmp_path) - quotes = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [0.10]}) + # 百分数口径: 10.0 = +10% (#232) + quotes = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [10.0]}) # 单日帧: 增量路径产出的 momentum 列 (无 date 历史, 无法 shift 补算) today_df = pl.DataFrame( { @@ -207,6 +223,21 @@ class _FakeQuotes: ) +class _PercentQuotes: + """返回百分数口径的指数缓存 (与 _build_index_quotes 的 x100 输出同形态)。""" + + def get_index_quotes(self): + return pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [-1.88]}) + + +def test_bench_rt_pct_converts_percent_to_decimal() -> None: + """#232 回归: _bench_rt_pct 必须把指数展示缓存的百分数转成小数, + 与 enriched 侧小数制 change_pct 同口径参与 rt_delta 计算。""" + assert abs(_bench_rt_pct(_PercentQuotes()) - (-0.0188)) < 1e-12 + # close/prev_close 兜底路径本身是小数, 不受影响 + assert abs(_bench_rt_pct(_FakeQuotes()) - (3300.0 / 3270.0 - 1)) < 1e-12 + + def test_build_overview_closeness_and_status() -> None: with _hist_cache_lock: _hist_cache.clear() diff --git a/backend/tests/test_enriched_partial_partition.py b/backend/tests/test_enriched_partial_partition.py new file mode 100644 index 0000000..2eff00f --- /dev/null +++ b/backend/tests/test_enriched_partial_partition.py @@ -0,0 +1,62 @@ +"""#223 回归: 盘后管道必须识别并重算覆盖不全的 enriched 分区。 + +自选实时路径 (merge_live_enriched_asset) 会在全市场 enriched 生成前提前 +创建当日分区 (只有几只自选); 仅按日期目录计数比较会把它误判为完整分区 +而跳过计算, 造成日K连续缺失与均线错误。_prune_partial_enriched_partitions +按同日 daily/enriched 行数比较, 发现部分分区即删除, 让增量重算按"新日期" +全市场补齐。 +""" +from __future__ import annotations + +from datetime import date +from pathlib import Path + +import polars as pl + +from app.jobs.daily_pipeline import _prune_partial_enriched_partitions + + +def _write_partition(base: Path, day: str, symbols: list[str]) -> None: + part = base / f"date={day}" + part.mkdir(parents=True, exist_ok=True) + pl.DataFrame({"symbol": symbols, "close": [1.0] * len(symbols)}).write_parquet( + part / "part.parquet" + ) + + +def test_partial_partition_is_pruned(tmp_path) -> None: + daily = tmp_path / "kline_daily" + enriched = tmp_path / "kline_daily_enriched" + # 2026-08-28: daily 全市场 100 只, enriched 只有实时写入的 5 只 (issue 实测形态) + _write_partition(daily, "2026-08-28", [f"s{i:06d}" for i in range(100)]) + _write_partition(enriched, "2026-08-28", [f"s{i:06d}" for i in range(5)]) + # 前一日两边都是完整的 → 不动 + _write_partition(daily, "2026-08-27", ["a", "b"]) + _write_partition(enriched, "2026-08-27", ["a", "b"]) + + pruned = _prune_partial_enriched_partitions(daily, enriched) + + assert pruned == ["2026-08-28"] + assert not (enriched / "date=2026-08-28").exists() + assert (enriched / "date=2026-08-27").exists() # 完整分区保留 + + +def test_complete_partitions_untouched(tmp_path) -> None: + daily = tmp_path / "kline_daily" + enriched = tmp_path / "kline_daily_enriched" + syms = [f"s{i:06d}" for i in range(50)] + _write_partition(daily, "2026-09-01", syms) + _write_partition(enriched, "2026-09-01", syms) + + assert _prune_partial_enriched_partitions(daily, enriched) == [] + assert (enriched / "date=2026-09-01" / "part.parquet").exists() + + +def test_enriched_date_without_daily_is_left_alone(tmp_path) -> None: + # 今日日K尚未同步时, 实时创建的当日分区留给当日正常流程处理 + daily = tmp_path / "kline_daily" + enriched = tmp_path / "kline_daily_enriched" + _write_partition(enriched, str(date.today()), ["only_watchlist"]) + + assert _prune_partial_enriched_partitions(daily, enriched) == [] + assert (enriched / f"date={date.today()}").exists() diff --git a/backend/tests/test_matrix_etf_share_fields.py b/backend/tests/test_matrix_etf_share_fields.py index ad69944..e0b5997 100644 --- a/backend/tests/test_matrix_etf_share_fields.py +++ b/backend/tests/test_matrix_etf_share_fields.py @@ -259,3 +259,30 @@ def test_stock_turnover_rate_still_derived_when_float_shares_present(tmp_path): assert np.isfinite(fields["turnover_rate"]).all() assert float(fields["turnover_rate"][0, 0]) == pytest.approx(1.0) assert float(fields["turnover_rate"][1, 0]) == pytest.approx(1.5) + + +# ── #215: 运行期股票专属过滤键中和 ──────────────────────────── + +def test_basic_filter_for_asset_neutralizes_stock_only_runtime_keys() -> None: + """#215 回归: boards 按股票代码前缀匹配、price_min=3 是股票专属口径, + 对 ETF 不可满足 —— 运行期不中和会让入场候选在掩码阶段静默清零 + (回测"正常完成"但零信号)。""" + from app.backtest.strategy import _basic_filter_for_asset + from app.strategy.engine import DEFAULT_BASIC_FILTER + + sanitized = _basic_filter_for_asset(dict(DEFAULT_BASIC_FILTER), "etf") + for key in ( + "price_min", "price_max", "boards", + "market_cap_min", "float_cap_min", "float_cap_max", + "turnover_min", "turnover_max", + ): + assert sanitized[key] is None, f"{key} 应被中和" + + # 与资产类型无关的键保留原值 + assert sanitized["amount_min"] == DEFAULT_BASIC_FILTER["amount_min"] + assert sanitized["exclude_st"] is True + assert sanitized["exclude_new_days"] == 30 + + # 股票口径完全不变 + stock = _basic_filter_for_asset(dict(DEFAULT_BASIC_FILTER), "stock") + assert stock == DEFAULT_BASIC_FILTER diff --git a/backend/tests/test_screener_external_access.py b/backend/tests/test_screener_external_access.py new file mode 100644 index 0000000..6f4c2d8 --- /dev/null +++ b/backend/tests/test_screener_external_access.py @@ -0,0 +1,101 @@ +"""#224 回归: screener 自定义 SQL 的内存连接必须关闭 external_access。 + +conditions/order_by 是用户可控的 SQL 片段; 隔离连接若允许外部访问, +注入的 read_parquet/COPY 可读写任意文件 (文件写 RCE)。 +""" +from __future__ import annotations + +from datetime import date +from unittest.mock import MagicMock + +import polars as pl + +from app.services.screener import ScreenerService + + +def _service_with_panel(panel: pl.DataFrame) -> ScreenerService: + svc = ScreenerService(MagicMock(), asset_type="stock") + svc._load_enriched_for_date = lambda d: panel # type: ignore[method-assign] + return svc + + +def _panel() -> pl.DataFrame: + return pl.DataFrame( + { + "symbol": ["600000.SH", "000001.SZ"], + "close": [10.0, 20.0], + "turnover_rate": [1.0, 2.0], + } + ) + + +def test_normal_condition_still_works() -> None: + svc = _service_with_panel(_panel()) + result = svc.run(date(2026, 9, 2), ["close > 15"], limit=10) + assert [r["symbol"] for r in result.rows] == ["000001.SZ"] + + +def test_injected_read_parquet_is_rejected() -> None: + # 注入试图读任意文件: external_access 关闭后 DuckDB 直接报错, + # run_custom 的 except 分支吞错返回空结果, 而非泄漏文件内容 + svc = _service_with_panel(_panel()) + result = svc.run( + date(2026, 9, 2), + ["1=1) UNION SELECT * FROM read_parquet('/etc/passwd') --"], + limit=10, + ) + assert result.rows == [] + + +def test_injected_copy_write_is_rejected(tmp_path) -> None: + target = tmp_path / "pwned.csv" + svc = _service_with_panel(_panel()) + result = svc.run( + date(2026, 9, 2), + [f"close > 0); COPY enriched TO '{target}' --"], + limit=10, + ) + assert result.rows == [] + assert not target.exists() + + +def test_order_by_injection_also_isolated(tmp_path) -> None: + # order_by 同样是拼接片段, 不能借 external 函数逃逸 + svc = _service_with_panel(_panel()) + result = svc.run( + date(2026, 9, 2), + ["close > 0"], + order_by=f"close; COPY enriched TO '{tmp_path / 'x.csv'}'", + limit=10, + ) + assert result.rows == [] + assert not (tmp_path / "x.csv").exists() + + +def test_external_access_switch_is_the_effective_barrier(tmp_path) -> None: + # 正反对照: 同一条注入 SQL, 未关 external_access 的普通内存连接能读到 + # 任意 parquet 文件 (证明攻击面真实存在); 关闭后直接报错。 + import duckdb + + victim = tmp_path / "victim.parquet" + _panel().write_parquet(victim) + inject = f"SELECT * FROM read_parquet('{victim}')" + + plain = duckdb.connect(database=":memory:") + try: + assert plain.execute(inject).pl().height == 2 # 普通连接: 可读 → 攻击面成立 + finally: + plain.close() + + hardened = duckdb.connect( + database=":memory:", config={"enable_external_access": False} + ) + try: + raised = False + try: + hardened.execute(inject) + except Exception: + raised = True + assert raised, "external_access=False 的连接不应能读外部文件" + finally: + hardened.close()