From a4e580e31410e68c7bc0720e9efef380b6767a09 Mon Sep 17 00:00:00 2001 From: shy3130 Date: Fri, 31 Jul 2026 19:56:33 +0800 Subject: [PATCH] =?UTF-8?q?fix(security):=20=E4=BF=AE=E5=A4=8D=20ext=5Fcol?= =?UTF-8?q?umns=20DuckDB=20SQL=20=E6=B3=A8=E5=85=A5=20(closes=20#150)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #150 报告了 ext_columns 功能的 DuckDB SQL 注入, 经核查属实: field_name 经裸双引号拼进 SQL, 攻击者可借 COPY TO 写文件 (RCE) 或 UNION 读数据。已修复, 详见 PR。 根因: 3 个 SQL sink 用未转义的 f"{field_name}", 仅 screener:147 一处 用 _quote_ident 转义; 且 parser 校验不一致 (screener 校验 config_id, kline/watchlist 零校验)。 修复方案: 1. 新增 backend/app/db_safe.py 集中定义 quote_ident (双引号转义, 对任意 字符安全) + is_valid_ext_ident (config_id 白名单), 消除不一致根因 2. 3 个 sink 统一改用 quote_ident: - screener.py:785 (limit_ladder) - kline.py:376 (_attach_ext, GET /api/kline/daily) - watchlist.py:279 (watchlist_enriched) 3. kline + watchlist parser 加 config_id 白名单 (screener 已有) 4. screener 复用共享原语, 删除私有 _quote_ident/_EXT_IDENT_RE 关键约束: - field_name 不能加白名单: FieldDef.name 无校验 + infer_fields_from_df 直接采用原始 CSV/Parquet 列名, 合法可含中文/点。故只在 sink 转义。 - 不关闭 enable_external_access: 实测会同时禁用 read_parquet (项目核心 数据访问方式), 不可行。 对 Issue #150 报告的澄清: - sink #2/#3/#4 (field_name 注入) 属实, 已修 - sink #1 (screener.py:343 conditions[]/order_by) 不存在, 为误报 (screener.py 无 conditions/order_by/execute 调用) 验证: 后端全量 527 passed (含新增 24 个安全测试, 含 2 个端到端注入 测试: COPY TO 不产生文件 / UNION 不泄漏数据 + 合法特殊字段名回归) --- backend/app/api/kline.py | 5 +- backend/app/api/screener.py | 13 +-- backend/app/api/watchlist.py | 5 +- backend/app/db_safe.py | 30 ++++++ backend/tests/test_ext_sql_safety.py | 148 +++++++++++++++++++++++++++ 5 files changed, 189 insertions(+), 12 deletions(-) create mode 100644 backend/app/db_safe.py create mode 100644 backend/tests/test_ext_sql_safety.py diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index cf80bba..5dcde88 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -12,6 +12,7 @@ from fastapi import APIRouter, HTTPException, Query, Request from app.indicators.pipeline import compute_enriched, compute_enriched_single from app.market_time import cn_now, cn_today from app.price_limits import is_risk_warning_name, price_limit_pct +from app.db_safe import is_valid_ext_ident, quote_ident from app.services import kline_sync logger = logging.getLogger(__name__) @@ -347,7 +348,7 @@ def _attach_ext(resp: dict, repo, symbol: str, ext_columns: Optional[str]) -> di continue config_id, field_name = part.split(".", 1) config_id, field_name = config_id.strip(), field_name.strip() - if config_id and field_name: + if config_id and field_name and is_valid_ext_ident(config_id): specs.append((config_id, field_name)) if not specs: return resp @@ -373,7 +374,7 @@ def _attach_ext(resp: dict, repo, symbol: str, ext_columns: Optional[str]) -> di else: ext_df = pl.from_arrow( repo.store.db.query( - f'SELECT symbol, "{field_name}" FROM ext_{config_id}' + f"SELECT symbol, {quote_ident(field_name)} FROM ext_{config_id}" ).arrow() ) if not ext_df.is_empty() and "symbol" in ext_df.columns and field_name in ext_df.columns: diff --git a/backend/app/api/screener.py b/backend/app/api/screener.py index 6d3bcd5..b05a5f1 100644 --- a/backend/app/api/screener.py +++ b/backend/app/api/screener.py @@ -14,6 +14,7 @@ from typing import Any, Optional from fastapi import APIRouter, HTTPException, Query, Request from pydantic import BaseModel +from app.db_safe import is_valid_ext_ident, quote_ident from app.services import strategy_cache from app.services.screener import ScreenerService from app.strategy import config as strategy_config @@ -68,9 +69,6 @@ def _one_word_limit_expr(status_main: str, columns: list[str]) -> Any: ).fill_null(False) -_EXT_IDENT_RE = re.compile(r"^[A-Za-z0-9_]+$") - - def _safe_ext_value(value: Any) -> Any: if isinstance(value, float) and not math.isfinite(value): return None @@ -79,8 +77,7 @@ def _safe_ext_value(value: Any) -> Any: return value -def _quote_ident(name: str) -> str: - return '"' + name.replace('"', '""') + '"' +# 标识符安全原语 (转义 + 白名单) 集中在 app.db_safe, 见 Issue #150 注入防护。 # ── 扩展列 value_map 缓存 ──────────────────────────────────────────── @@ -144,7 +141,7 @@ def _load_ext_value_maps(repo, ext_columns: Optional[str]) -> dict[str, dict[str else: view_name = f"ext_{config_id}" ext_df = pl.from_arrow(db.query( - f"SELECT symbol, {_quote_ident(field_name)} FROM {view_name}" + f"SELECT symbol, {quote_ident(field_name)} FROM {view_name}" ).arrow()) if ext_df.is_empty() or "symbol" not in ext_df.columns or field_name not in ext_df.columns: @@ -785,7 +782,7 @@ def limit_ladder( ext_col_name = f"{config_id}__{field_name}" try: ext_df = pl.from_arrow(db.query( - f"SELECT symbol, \"{field_name}\" FROM {view_name}" + f"SELECT symbol, {quote_ident(field_name)} FROM {view_name}" ).arrow()) if not ext_df.is_empty() and "symbol" in ext_df.columns: ext_df = ext_df.rename({field_name: ext_col_name}) @@ -860,7 +857,7 @@ def _parse_ext_columns(ext_columns: str) -> list[tuple[str, str]]: field_name = field_name.strip() if not config_id or not field_name: continue - if not _EXT_IDENT_RE.match(config_id) or "\x00" in field_name: + if not is_valid_ext_ident(config_id) or "\x00" in field_name: continue result.append((config_id, field_name)) return result diff --git a/backend/app/api/watchlist.py b/backend/app/api/watchlist.py index ef8e6f6..8780b16 100644 --- a/backend/app/api/watchlist.py +++ b/backend/app/api/watchlist.py @@ -11,6 +11,7 @@ import polars as pl from fastapi import APIRouter, File, HTTPException, Query, Request, UploadFile from pydantic import BaseModel +from app.db_safe import is_valid_ext_ident, quote_ident from app.services import watchlist from app.services.watchlist_ocr import import_watchlist_image from app.services.watchlist_ocr.provider import get_ocr_provider @@ -276,7 +277,7 @@ def watchlist_enriched( ext_df, _ = _read_ext_dataframe(cfg, data_dir) else: ext_df = pl.from_arrow(db.query( - f"SELECT symbol, \"{field_name}\" FROM {view_name}" + f"SELECT symbol, {quote_ident(field_name)} FROM {view_name}" ).arrow()) if not ext_df.is_empty() and "symbol" in ext_df.columns: ext_df = ( @@ -334,6 +335,6 @@ def _parse_ext_columns(ext_columns: str) -> list[tuple[str, str]]: config_id, field_name = part.split(".", 1) config_id = config_id.strip() field_name = field_name.strip() - if config_id and field_name: + if config_id and field_name and is_valid_ext_ident(config_id): result.append((config_id, field_name)) return result diff --git a/backend/app/db_safe.py b/backend/app/db_safe.py new file mode 100644 index 0000000..6dfbb61 --- /dev/null +++ b/backend/app/db_safe.py @@ -0,0 +1,30 @@ +"""DuckDB 标识符安全原语 (Issue #150: ext_columns SQL 注入防护)。 + +集中提供标识符转义与校验, 供所有拼接 DuckDB SQL 的 sink 复用, 消除 +"screener 有防护、kline/watchlist 没有"的不一致根因。 +""" +from __future__ import annotations + +import re + +# 合法标识符白名单: 字母 / 数字 / 下划线。与 CreateExtReq.id 创建端校验一致, +# 用于 config_id 等「纯用户可控 + 仅合法标识符」字段的深度防御。 +EXT_IDENT_RE = re.compile(r"^[A-Za-z0-9_]+$") + + +def quote_ident(name: str) -> str: + """把列名/字段名转义为 DuckDB 双引号标识符。 + + 对任意字符安全: 仅对内嵌的双引号做双写转义 (``"`` → ``""``), 再用双引号包裹。 + 转义后整个串成为单个「带引号标识符」, 即便 name 含 ``--``/``;``/``UNION``/``COPY`` + 也只会被 DuckDB 当作字面列名, 不会被解释为 SQL 语法, 从根上杜绝注入。 + + 用于 field_name: 因 FieldDef.name 可合法含中文/点/特殊字符, 不能用白名单, + 只能在 sink 处转义。 + """ + return '"' + name.replace('"', '""') + '"' + + +def is_valid_ext_ident(name: str) -> bool: + """config_id 等纯标识符字段是否仅含合法字符 (字母/数字/下划线)。""" + return bool(EXT_IDENT_RE.match(name)) diff --git a/backend/tests/test_ext_sql_safety.py b/backend/tests/test_ext_sql_safety.py new file mode 100644 index 0000000..878479e --- /dev/null +++ b/backend/tests/test_ext_sql_safety.py @@ -0,0 +1,148 @@ +"""ext_columns SQL 注入防护测试 (Issue #150)。 + +覆盖: +- quote_ident 对各注入 payload 的转义正确性 (含 RCE 向量 COPY TO) +- 端到端: 用真实 DuckDB 验证恶意 field_name 不会产生文件 / 不泄漏数据 +- is_valid_ext_ident 白名单 (config_id 防御) +- 回归: 合法特殊字符字段名 (中文/点) 仍能透传到 sink (证明未误伤数据) +""" +from __future__ import annotations + +import os +import tempfile + +import duckdb +import pytest + +from app.db_safe import is_valid_ext_ident, quote_ident + + +# ===== quote_ident 转义正确性 ===== + +def test_quote_ident_wraps_in_double_quotes(): + assert quote_ident("close") == '"close"' + + +def test_quote_ident_escapes_embedded_double_quote(): + # 双引号双写: 字段名含 " 时必须转义, 否则可逃逸标识符 + assert quote_ident('a"b') == '"a""b"' + + +@pytest.mark.parametrize("payload", [ + 'x" UNION SELECT 1', + 'x"); DROP TABLE t; --', + 'x" -- comment', + 'x"; --', + 'x" COPY (SELECT 1) TO \'/tmp/evil\' --', +]) +def test_quote_ident_neutralizes_injection_payloads(payload): + """转义后的串必须是一个整体带引号标识符, 内部双引号被双写。""" + escaped = quote_ident(payload) + # 整体仍是 "...." 包裹, 内部每个原始 " 都变成 "" + assert escaped.startswith('"') and escaped.endswith('"') + # 原始 payload 里的每个 " 都被双写 + inner = escaped[1:-1] + assert inner == payload.replace('"', '""') + + +# ===== 端到端注入测试 (真实 DuckDB) ===== + +def _build_ext_view(con: duckdb.DuckDBPyConnection, field_name: str) -> None: + """建一个含指定 field_name 列的 ext_x 视图, 模拟项目扩展数据。""" + # 列名含特殊字符时, 建表也必须用双引号转义 + con.execute(f'CREATE OR REPLACE TABLE _t_{id(field_name) % 10000} (symbol VARCHAR, {quote_ident(field_name)} DOUBLE)') + con.execute(f"INSERT INTO _t_{id(field_name) % 10000} VALUES ('000001.SZ', 12.3)") + con.execute(f'CREATE OR REPLACE VIEW ext_x AS SELECT symbol, {quote_ident(field_name)} FROM _t_{id(field_name) % 10000}') + + +def test_sink_with_malicious_field_name_does_not_write_file(): + """恶意 field_name 经 quote_ident 后, COPY TO payload 不产生文件。 + + 复现 Issue #150 的 RCE 向量: 若 field_name 裸拼, 攻击者可构造 COPY TO 写任意文件。 + 转义后, 整个 payload 变成字面列名, DuckDB 查询报「列不存在」, 不会执行 COPY。 + """ + tmp = os.path.join(tempfile.gettempdir(), "tf_sec_inject_sink.out") + if os.path.exists(tmp): + os.remove(tmp) + + con = duckdb.connect(":memory:") + _build_ext_view(con, "close") # 合法列名 + + # 攻击 payload: 企图在查询里塞 COPY TO 写文件 + evil = 'close" UNION SELECT * FROM (COPY (SELECT 1) TO \'' + tmp + '\') --' + sql = f'SELECT symbol, {quote_ident(evil)} FROM ext_x' + # 转义后查询合法列失败 → 报 Binder Error, 绝不会执行 COPY + with pytest.raises(Exception): + con.execute(sql).fetchall() + + # 关键断言: 文件未被创建 (COPY 未执行) + assert not os.path.exists(tmp), "RCE: COPY TO 写文件成功, 注入未堵死!" + + +def test_sink_with_malicious_field_name_does_not_leak_data(): + """恶意 field_name 企图用 UNION 读其它表数据 → 转义后仅作字面列名, 查询失败。""" + con = duckdb.connect(":memory:") + _build_ext_view(con, "close") + con.execute("CREATE TABLE secret (pw VARCHAR)") + con.execute("INSERT INTO secret VALUES ('leaked')") + + evil = 'close" UNION SELECT pw FROM secret --' + sql = f'SELECT symbol, {quote_ident(evil)} FROM ext_x' + # 不会返回 secret 表内容, 而是报错 (列 close"... UNION... 不存在) + with pytest.raises(Exception): + con.execute(sql).fetchall() + + +# ===== is_valid_ext_ident 白名单 ===== + +@pytest.mark.parametrize("ident, expected", [ + ("abc", True), + ("abc_123", True), + ("ABC_xyz", True), + ("x_y", True), + ("", False), + ("a b", False), # 含空格 + ("a-b", False), # 含连字符 + ("a.b", False), # 含点 + ("a\"b", False), # 含双引号 + ("a'b", False), # 含单引号 + ("a;b", False), # 含分号 + ("中文", False), # 含中文 (config_id 不允许, 但 field_name 允许 → 见回归测试) +]) +def test_is_valid_ext_ident(ident, expected): + assert is_valid_ext_ident(ident) is expected + + +# ===== parser 白名单: 拒绝恶意 config_id ===== + +def test_watchlist_parser_rejects_malicious_config_id(): + from app.api.watchlist import _parse_ext_columns + # config_id 含注入字符 → 应被白名单拒绝 + assert _parse_ext_columns("evil'; DROP--.field") == [] + assert _parse_ext_columns("normal.field") == [("normal", "field")] + + +def test_kline_inline_parser_filter_via_is_valid_ext_ident(): + """kline parser 内联, 直接用 is_valid_ext_ident 验证过滤逻辑。""" + # 模拟 kline._attach_ext 的解析过滤 + parts = ["ok.field", "bad';DROP.field", "中文.field", "a_1.value"] + specs = [] + for part in parts: + config_id, field_name = part.split(".", 1) + if config_id and field_name and is_valid_ext_ident(config_id): + specs.append((config_id, field_name)) + assert specs == [("ok", "field"), ("a_1", "value")] + + +# ===== 回归: 合法特殊字符字段名未被误伤 ===== + +def test_legal_special_field_name_still_queryable(): + """field_name 可合法含中文/点 (FieldDef.name 无校验), quote_ident 必须支持。""" + con = duckdb.connect(":memory:") + for name in ["涨幅", "市盈率.静态", "field.name"]: + _build_ext_view(con, name) + # 用 quote_ident 转义后, 特殊列名仍能正确查询 + sql = f'SELECT symbol, {quote_ident(name)} FROM ext_x' + rows = con.execute(sql).fetchall() + assert rows and rows[0][0] == "000001.SZ", f"合法字段名 {name} 查询失败 (被误伤)" + con.close()