From 26ff8e1e064b74df36ced33374f3f3f280cda900 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 25 Sep 2026 14:44:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=AA=E8=AF=BB=E6=95=B0=E6=8D=AE=E6=BA=90?= =?UTF-8?q?=E7=9C=9F=E7=9A=84=E5=8F=AA=E8=AF=BB=E4=BA=86=EF=BC=9AWITH=20?= =?UTF-8?q?=E2=80=A6=20DELETE=20=E4=B8=8D=E5=86=8D=E4=BB=8E=E9=A6=96?= =?UTF-8?q?=E5=85=B3=E9=94=AE=E5=AD=97=E4=B8=8B=E9=9D=A2=E9=92=BB=E8=BF=87?= =?UTF-8?q?=E5=8E=BB=EF=BC=8C=E8=BF=9E=E6=8E=A5=E6=9C=AC=E8=BA=AB=E4=B9=9F?= =?UTF-8?q?=E5=86=99=E4=B8=8D=E8=BF=9B=E5=8E=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 守卫只看首关键字，而 with / pragma 都在只读白名单里。在只读的 SQLite 源上 实测：`WITH k AS (SELECT 1) DELETE FROM t` 被放行，5 行删得只剩 2 行； `PRAGMA user_version = 7` 也写进去了。调用方只看到一个 ResourceClosedError—— pysqlite 只在 INSERT/UPDATE/DELETE 开头的语句前隐式开事务，WITH 开头的直接 自动提交，连接关闭时的回滚兜不住。Excel/CSV 上传生成的正是这种 SQLite 源。 两层一起收： 守卫：以查询开头的语句（SELECT / WITH / EXPLAIN）要把正文扫一遍，先抹掉注释 和引号里的内容，再找写关键字（DELETE、UPDATE、INTO、FOR UPDATE、EXPLAIN ANALYZE 里的语句……），关键字后紧跟 ( 的是同名函数不算（MySQL 的 INSERT()、 TRUNCATE()）。PRAGMA 改成只读白名单，`= 值` 和 `(值)` 两种赋值写法都拦。 另收一小撮有副作用的函数，其中 set_config 必须拦：PG 真库实测，它能把连接层 的只读关掉，同一连接的下一个事务就能写。函数名加引号、dblink 另开连接这两种 绕法也补上了。is_write 与只读判定共用一份判定，免得审批那道门另有一套口径。 连接层：只读源连接时就只读（engine_args）。SQLite 以 mode=ro 打开（file: URI， 路径带空格、中文、# 都测过）；PostgreSQL 设会话默认事务只读；MySQL / MariaDB 设会话事务只读。Oracle 没有会话级开关，只剩守卫——README 写明请配只读账号。 真库验证（临时实例，已清理）：PostgreSQL 18 与 MySQL 8.0 上，DELETE、WITH … DELETE、 PG 的数据修改 CTE、nextval、EXPLAIN ANALYZE 绕过守卫直连执行，都被数据库以 只读事务拒绝；可写源不受影响。 测试：守卫新增 33 例（藏在查询里的写、会写的 PRAGMA、同名函数与引号标识符不误杀）， 去掉修复重跑是 17 例失败；test_readonly_source.py 6 例（当初那条绕过、绕过守卫 直连写、中文路径、可写源不受波及、各库的会话参数）。297 → 334 passed。 顺带查到但没在这里修的：可写源上的写操作既不审批也不生效——is_dangerous_call 全仓库没人调用，而 run_query 从不 commit，写完就被回滚。两件事必须一起修 （先接审批再让写入生效，反过来就是一个无审批写库的窗口），放在审批统一那一步。 --- README.md | 7 +- backend/app/data/engine.py | 37 +++++++- backend/app/data/guard.py | 130 ++++++++++++++++++++++++-- backend/tests/test_readonly_source.py | 115 +++++++++++++++++++++++ backend/tests/test_sql_guard.py | 64 ++++++++++++- 5 files changed, 339 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_readonly_source.py diff --git a/README.md b/README.md index 6bf9b342..06e0cc29 100644 --- a/README.md +++ b/README.md @@ -340,8 +340,11 @@ Excel / CSV 若传入知识库，只能被切块检索，数字即成为模型 - 分支条件使用 AST 白名单求值器而非 `eval` - HTTP 工具解析域名后逐 IP 检查，拦截内网、回环与云元数据地址，且不跟随重定向（逐跳重新校验） - 文件工具路径 `resolve()` 后必须仍位于工作区内 -- SQL 守卫采用白名单动词判定 + 多语句拦截 + 结果集上限；`DROP` / `TRUNCATE` / `GRANT` 等 - 在可写数据源上亦一律拒绝 +- SQL 守卫采用白名单动词判定 + 正文扫描 + 多语句拦截 + 结果集上限；`DROP` / `TRUNCATE` / `GRANT` 等 + 在可写数据源上亦一律拒绝。以查询开头却会写的语句（`WITH … DELETE`、`SELECT … INTO`、 + `EXPLAIN ANALYZE`、`PRAGMA … =`、`set_config()` 等）按写操作处理 +- 只读数据源在**连接层**也只读，不只靠守卫猜：SQLite 以 `mode=ro` 打开，PostgreSQL / MySQL + 会话事务只读。**Oracle 没有会话级只读开关**，只剩守卫一道——只读数据源请配只读账号 - API Key 经 Fernet 加密落库，仅向前端返回掩码 --- diff --git a/backend/app/data/engine.py b/backend/app/data/engine.py index 79c75e1d..25c29f9e 100644 --- a/backend/app/data/engine.py +++ b/backend/app/data/engine.py @@ -107,10 +107,41 @@ def build_url(source: Any, *, reveal: bool = False) -> str: return f"{driver}://{auth}{host}/{source.database or ''}{('?' + query) if query else ''}" +def engine_args(source: Any) -> tuple[str, dict[str, Any]]: + """真正连库用的连接串和额外参数。只读源在这里拿到**连接层**的只读。 + + 守卫按关键字判定，而模型能写出什么 SQL 事前无法穷举——`WITH … DELETE` + 就曾从首关键字白名单下面钻过去，在 SQLite 上真把数据删了（pysqlite 只在 + INSERT/UPDATE/DELETE 开头的语句前隐式开事务，WITH 开头的直接自动提交）。 + 所以只读不能只靠猜，连接本身就得写不进去： + + - SQLite：以 mode=ro 打开文件，写操作由 SQLite 自己拒绝。 + - PostgreSQL：会话的默认事务只读。能关掉它的 SET / set_config() 守卫都拦。 + - MySQL / MariaDB：会话事务只读。能关掉它的只有 SET，守卫拦。 + - Oracle：没有会话级只读开关，只剩守卫这一道——只读数据源请配只读账号。 + """ + url = build_url(source, reveal=True) + if not source.readonly: + return url, {} + kind = (source.kind or "").lower() + if kind == "sqlite": + path = source.database or "" + if path and path != ":memory:": + from urllib.parse import quote + url = f"{_DRIVERS['sqlite']}:///file:{quote(path)}?mode=ro&uri=true" + return url, {} + if kind in ("postgres", "postgresql"): + return url, {"connect_args": {"server_settings": {"default_transaction_read_only": "on"}}} + if kind in ("mysql", "mariadb"): + return url, {"connect_args": {"init_command": "SET SESSION TRANSACTION READ ONLY"}} + return url, {} + + class EngineCache: """按数据源 id 缓存 engine。连接池的建立不便宜，不该每次查询都重来一遍。 - 配置变了要显式 invalidate——改了密码却还在用旧连接，排查起来很费神。 + 配置变了要显式 invalidate——改了密码却还在用旧连接，排查起来很费神； + 改了只读开关却还在用旧连接，只读就形同虚设。 """ def __init__(self) -> None: @@ -125,12 +156,14 @@ class EngineCache: async with self._lock: if key in self._engines: return self._engines[key] + url, extra = engine_args(source) engine = create_async_engine( - build_url(source, reveal=True), + url, pool_size=3, max_overflow=2, pool_pre_ping=True, # 长时间空闲后连接会被数据库掐掉，先探活再用 pool_recycle=1800, + **extra, ) self._engines[key] = engine return engine diff --git a/backend/app/data/guard.py b/backend/app/data/guard.py index 4b6022d1..eda0e85c 100644 --- a/backend/app/data/guard.py +++ b/backend/app/data/guard.py @@ -38,6 +38,42 @@ _ALWAYS_DENIED = frozenset({ _LINE_COMMENT = re.compile(r"--[^\n]*") _BLOCK_COMMENT = re.compile(r"/\*.*?\*/", re.S) +# 首关键字是查询、正文却能写的语句。只看首关键字的话它们全都能混过去： +# WITH k AS (SELECT 1) DELETE FROM t SQLite / MySQL / PG 都认 +# WITH x AS (DELETE … RETURNING *) SELECT … PG 的数据修改 CTE +# SELECT … INTO backup / INTO OUTFILE '…' PG 建表 / MySQL 写服务器文件 +# EXPLAIN ANALYZE DELETE … PG 的 ANALYZE 会真的执行 +# 关键字后面紧跟 ( 的不算：那是同名函数（MySQL 的 INSERT()、TRUNCATE()）。 +_HIDDEN_WRITE = re.compile( + r"\b(insert|update|delete|merge|upsert|truncate|drop|alter|create|grant|revoke|" + r"call|copy|into|attach|detach|vacuum|reindex|lock\s+tables?|replace\s+into)\b(?!\s*\()", + re.I, +) + +# 查询里一调用就有副作用的函数。函数没法白名单，这里只收一小撮要命的： +# set_config 能把 PG 连接层的只读关掉（真库实测：关掉后同一连接的下一个事务 +# 就能写），dblink 另开一条不受只读约束的连接，nextval/setval 推进序列，其余是 +# 杀连接、读写服务器文件、加载本地扩展。函数名可以加引号写（"set_config"(…)）， +# 所以名字两边允许引号，扫描时也只抹字符串、不抹标识符。 +_SIDE_EFFECT_CALL = re.compile( + r"[\"`]?\b(set_config|nextval|setval|pg_terminate_backend|pg_cancel_backend|pg_reload_conf|" + r"lo_import|lo_export|lo_unlink|dblink\w*|load_extension)\b[\"`]?\s*\(", + re.I, +) + +# SQLite 的 PRAGMA 有读有写，而且写有两种写法：`= 值` 和 `(值)`。 +# 只放行确定只读的：前一组的参数是表名/索引名，后一组不带参数。 +_PRAGMA_READ_WITH_ARG = frozenset({ + "table_info", "table_xinfo", "table_list", "index_list", "index_info", + "index_xinfo", "foreign_key_list", +}) +_PRAGMA_READ_NO_ARG = frozenset({ + "table_list", "database_list", "collation_list", "function_list", "module_list", + "pragma_list", "compile_options", "encoding", "page_count", "page_size", + "freelist_count", "schema_version", "user_version", "application_id", "data_version", +}) +_PRAGMA = re.compile(r"pragma\s+(?:\w+\s*\.\s*)?(\w+)\s*(.*)$", re.I | re.S) + class SqlRejected(ValueError): """SQL 没通过守卫。消息直接给模型看，所以要说清楚为什么被拒、该怎么改。""" @@ -101,6 +137,75 @@ def first_verb(sql: str) -> str: return match.group(0).lower() if match else "" +def _blank_quoted(sql: str, *, identifiers: bool = True) -> str: + """把引号里的内容换成空格：字符串 'please delete' 和标识符 "update" 都不是关键字。 + + identifiers=False 时只抹字符串、保留 "…" 和 `…`——找函数调用时要看得见 + 被引号括起来的函数名。 + """ + quotes = ("'", '"', "`") if identifiers else ("'",) + out: list[str] = [] + quote: str | None = None + i = 0 + while i < len(sql): + ch = sql[i] + if quote: + if ch == quote: + if i + 1 < len(sql) and sql[i + 1] == quote: # '' 是转义，不是结束 + out.append(" ") + i += 2 + continue + quote = None + out.append(ch) + else: + out.append(" ") + else: + if ch in quotes: + quote = ch + out.append(ch) + i += 1 + return "".join(out) + + +def _pragma_write_reason(stmt: str) -> str | None: + match = _PRAGMA.match(strip_comments(stmt).strip()) + if not match: + return "看不出这条 PRAGMA 读的是什么" + name, rest = match.group(1).lower(), match.group(2).strip() + if rest.startswith("="): + return f"PRAGMA {name} = … 会改数据库设置" + if name in _PRAGMA_READ_WITH_ARG or (name in _PRAGMA_READ_NO_ARG and not rest): + return None + return ( + f"PRAGMA {name}{' ' + rest if rest else ''} 不在只读白名单里" + "（看表结构用 table_info / index_list / foreign_key_list）" + ) + + +def _write_reason(stmt: str) -> str | None: + """这条语句会不会写。会写就返回一句说给模型听的原因，不会写返回 None。 + + 首关键字只是第一道：以查询开头的语句正文照样能写（见 _HIDDEN_WRITE）， + 所以 SELECT / WITH / EXPLAIN 还要把正文扫一遍。SHOW / DESCRIBE / VALUES + 结构上写不了，不扫——`SHOW CREATE TABLE` 里的 CREATE 不是在建表。 + """ + verb = first_verb(stmt) + if verb not in _READONLY_VERBS: + return f"{verb.upper()} 会修改数据" + if verb == "pragma": + return _pragma_write_reason(stmt) + if verb in ("select", "with", "explain"): + text = strip_comments(stmt) + hidden = _HIDDEN_WRITE.search(_blank_quoted(text)) + if hidden: + word = " ".join(hidden.group(1).split()).upper() + return f"语句里有 {word}——以查询开头的语句也能写数据" + call = _SIDE_EFFECT_CALL.search(_blank_quoted(text, identifiers=False)) + if call: + return f"{call.group(1)}() 有副作用" + return None + + def check(sql: str, *, readonly: bool, source_name: str = "") -> str: """校验并返回规范化后的单条 SQL。不通过就抛 SqlRejected。 @@ -130,17 +235,24 @@ def check(sql: str, *, readonly: bool, source_name: str = "") -> str: "这类操作请在数据库客户端里人工执行。" ) - if readonly and verb not in _READONLY_VERBS: - where = f"「{source_name}」" if source_name else "这个数据源" - raise SqlRejected( - f"{where}是只读的，{verb.upper()} 不被允许。" - "只能执行 SELECT / WITH / SHOW / DESCRIBE / EXPLAIN。" - "确实需要写入的话，请在设置里另建一个关闭了只读的数据源。" - ) + if readonly: + reason = _write_reason(stmt) + if reason: + where = f"「{source_name}」" if source_name else "这个数据源" + raise SqlRejected( + f"{where}是只读的：{reason}，不被允许。" + "只能执行不改数据的 SELECT / WITH / SHOW / DESCRIBE / EXPLAIN。" + "确实需要写入的话，请在设置里另建一个关闭了只读的数据源。" + ) return stmt def is_write(sql: str) -> bool: - """是不是写操作。用来决定要不要走人工审批。""" - return first_verb(sql) not in _READONLY_VERBS + """是不是写操作。用来决定要不要走人工审批。 + + 和只读判定用同一份 _write_reason：审批这道门要是只看首关键字， + `WITH … DELETE` 在可写源上就能不经审批直接写。 + """ + statements = split_statements(sql) + return not statements or any(_write_reason(s) is not None for s in statements) diff --git a/backend/tests/test_readonly_source.py b/backend/tests/test_readonly_source.py new file mode 100644 index 00000000..d381f00e --- /dev/null +++ b/backend/tests/test_readonly_source.py @@ -0,0 +1,115 @@ +"""只读数据源在连接层也只读。 + +守卫按关键字判定，漏判过一次：`WITH k AS (SELECT 1) DELETE FROM t` 首关键字是 +WITH，被当成查询放行；而 pysqlite 只在 INSERT/UPDATE/DELETE 开头的语句前隐式开 +事务，WITH 开头的直接自动提交——连接关闭时的回滚兜不住，数据真的删了，调用方 +只看到一个驱动报错。Excel/CSV 上传生成的正是这种 SQLite 源。 + +所以只读不能只靠猜。这组测试绕过守卫，直接在连接上写，看它拒不拒。 +""" +from __future__ import annotations + +import sqlite3 +import uuid +from types import SimpleNamespace + +import pytest +from sqlalchemy import text +from sqlalchemy.exc import OperationalError + +from app.data.engine import engine_args, engines, run_query +from app.data.guard import SqlRejected + + +def _db(path) -> str: + conn = sqlite3.connect(path) + conn.execute("CREATE TABLE t (x INTEGER)") + conn.executemany("INSERT INTO t VALUES (?)", [(i,) for i in range(5)]) + conn.commit() + conn.close() + return str(path) + + +def _source(path: str = "", *, readonly: bool = True, kind: str = "sqlite"): + return SimpleNamespace( + id=f"ro-{uuid.uuid4().hex[:8]}", kind=kind, database=path, readonly=readonly, + name="测试源", options={}, password=None, username="u", host="h", port=None, + ) + + +def _state(path: str) -> tuple[int, int]: + conn = sqlite3.connect(path) + try: + return (conn.execute("SELECT COUNT(*) FROM t").fetchone()[0], + conn.execute("PRAGMA user_version").fetchone()[0]) + finally: + conn.close() + + +@pytest.fixture +async def ro_source(tmp_path): + src = _source(_db(tmp_path / "shop.db")) + yield src + await engines.invalidate(src.id) + + +async def test_the_original_bypass_is_closed(ro_source): + """当初那条：守卫放行、驱动报错、数据已经没了。现在守卫就拦下，数据原样。""" + with pytest.raises(SqlRejected, match="只读"): + await run_query(ro_source, "WITH k AS (SELECT 1) DELETE FROM t WHERE x < 3") + assert _state(ro_source.database) == (5, 0) + + +async def test_readonly_sqlite_refuses_writes_even_past_the_guard(ro_source): + """守卫再漏一次也没用：连接是以 mode=ro 打开的，写由 SQLite 自己拒绝。""" + engine = await engines.get(ro_source) + async with engine.connect() as conn: + for sql in ("DELETE FROM t", + "WITH k AS (SELECT 1) DELETE FROM t", + "PRAGMA user_version = 7"): + with pytest.raises(OperationalError, match="readonly"): + await conn.execute(text(sql)) + assert _state(ro_source.database) == (5, 0) + + +async def test_readonly_sqlite_still_answers_queries(ro_source): + result = await run_query(ro_source, "SELECT COUNT(*) AS n FROM t") + assert result.rows == [[5]] + info = await run_query(ro_source, "PRAGMA table_info(t)") + assert info.row_count == 1 + + +async def test_readonly_path_with_spaces_and_cjk_still_opens(tmp_path): + """连接串走 file: URI，路径得转义——带空格、中文、# 的文件名不能打不开。""" + src = _source(_db(tmp_path / "销售 数据#1.db")) + try: + assert (await run_query(src, "SELECT COUNT(*) FROM t")).rows == [[5]] + finally: + await engines.invalidate(src.id) + + +async def test_writable_sqlite_is_not_opened_readonly(tmp_path): + """只读收紧不能波及可写源。""" + src = _source(_db(tmp_path / "rw.db"), readonly=False) + engine = await engines.get(src) + try: + async with engine.begin() as conn: + await conn.execute(text("DELETE FROM t WHERE x = 0")) + finally: + await engines.invalidate(src.id) + assert _state(src.database) == (4, 0) + + +def test_server_databases_get_session_level_readonly(): + """PG / MySQL 的只读落在会话上（真库验证见提交说明）。 + + Oracle 没有会话级只读开关，只剩守卫——这里把这一点钉住，免得有人以为它也有。 + """ + _, pg = engine_args(_source("db", kind="postgresql")) + assert pg["connect_args"]["server_settings"]["default_transaction_read_only"] == "on" + _, my = engine_args(_source("db", kind="mysql")) + assert my["connect_args"]["init_command"] == "SET SESSION TRANSACTION READ ONLY" + _, ora = engine_args(_source("db", kind="oracle")) + assert ora == {} + _, rw = engine_args(_source("db", kind="postgresql", readonly=False)) + assert rw == {} diff --git a/backend/tests/test_sql_guard.py b/backend/tests/test_sql_guard.py index 1cd5233c..2cc48b14 100644 --- a/backend/tests/test_sql_guard.py +++ b/backend/tests/test_sql_guard.py @@ -7,7 +7,7 @@ from __future__ import annotations import pytest -from app.data.guard import SqlRejected, check, first_verb, split_statements +from app.data.guard import SqlRejected, check, first_verb, is_write, split_statements def rejected(sql: str, *, readonly: bool = True) -> str | None: @@ -47,6 +47,68 @@ def test_readonly_blocks_writes(sql): assert reason and "只读" in reason +@pytest.mark.parametrize("sql", [ + # 首关键字是查询，正文在写。只看首关键字的话全都能混过去—— + # 第一条在 SQLite 上实测过：守卫放行，数据真的删了 + "WITH k AS (SELECT 1) DELETE FROM t WHERE x < 3", + "WITH x AS (DELETE FROM t RETURNING *) SELECT * FROM x", # PG 的数据修改 CTE + "WITH k AS (SELECT 1) UPDATE t SET x = 1", + "WITH k AS (SELECT 1) INSERT INTO t SELECT * FROM k", + "SELECT * INTO backup FROM orders", # PG：建一张新表 + "SELECT * FROM orders INTO OUTFILE '/tmp/o.csv'", # MySQL：写服务器上的文件 + "EXPLAIN ANALYZE DELETE FROM orders", # PG：ANALYZE 会真的执行 + "SELECT * FROM orders FOR UPDATE", # 行锁 + "SELECT set_config('default_transaction_read_only', 'off', false)", # 关掉连接层只读 + "SELECT \"set_config\"('default_transaction_read_only', 'off', false)", # 函数名加引号 + "SELECT pg_catalog.set_config('default_transaction_read_only', 'off', false)", + "SELECT * FROM dblink('dbname=x', 'DELETE FROM t RETURNING *') AS r(x int)", # 另开连接 + "SELECT nextval('order_seq')", +]) +def test_readonly_blocks_writes_hidden_behind_a_query_verb(sql): + reason = rejected(sql) + assert reason and "只读" in reason, f"漏了：{sql!r}" + + +@pytest.mark.parametrize("sql", [ + "PRAGMA user_version = 7", + "PRAGMA user_version(7)", # 函数写法同样是赋值 + "PRAGMA query_only = 0", # 想先把只读关掉 + "PRAGMA main.writable_schema = 1", + "PRAGMA optimize", # 会跑 ANALYZE，写统计表 + "PRAGMA wal_checkpoint(TRUNCATE)", +]) +def test_readonly_blocks_pragmas_that_write(sql): + reason = rejected(sql) + assert reason and "只读" in reason, f"漏了：{sql!r}" + + +@pytest.mark.parametrize("sql", [ + "PRAGMA table_info(orders)", + "PRAGMA main.table_info('orders')", + "PRAGMA index_list(orders)", + "PRAGMA foreign_key_list(orders)", + "PRAGMA user_version", + "SELECT updated_at, deleted_flag, created_by FROM orders", # 标识符里带关键字不算 + "SELECT 'please delete me' AS note", # 字符串里的不算 + 'SELECT "update" FROM t', # 引号里的标识符不算 + "SELECT REPLACE(name, 'a', 'b'), INSERT('abc', 1, 1, 'x'), TRUNCATE(price, 2) FROM t", + "SHOW CREATE TABLE orders", # MySQL 看建表语句 + "WITH t AS (SELECT 1 AS into_count) SELECT * FROM t", +]) +def test_readonly_does_not_misfire_on_harmless_queries(sql): + """正文扫描收紧了，误杀也得守住：同名函数、标识符、字符串都不是在写。""" + assert rejected(sql) is None, f"误杀了：{sql!r}" + + +def test_is_write_sees_writes_hidden_behind_a_query_verb(): + """可写源靠 is_write 决定要不要人工审批——WITH 开头的写不能被当成查询放过去。""" + assert is_write("WITH k AS (SELECT 1) DELETE FROM t") + assert is_write("SELECT * INTO backup FROM t") + assert is_write("UPDATE t SET x = 1") + assert not is_write("SELECT * FROM t") + assert not is_write("WITH k AS (SELECT 1) SELECT * FROM k") + + @pytest.mark.parametrize("sql", [ "DROP TABLE orders", "TRUNCATE TABLE orders",