Python sqlite3.Connection.set_authorizer 如何拦截 SQL：动作码判断、返回值语义与审计范围

给一段来自插件或迁移脚本的 SQL 做只读审查时，光靠执行前字符串匹配很容易漏掉别名、子查询和隐式访问。Python 的 sqlite3.Connection.set_authorizer() 可以把检查点放到 SQLite 编译语句的阶段：每遇到一次读表、写表或调用函数，回调都会收到动作码和相关对象名。把返回值设为 SQLITE_OK、SQLITE_DENY 或 SQLITE_IGNORE，就能分别放行、拒绝整条语句，或让某个列读出来变成 NULL。

最小可靠做法是先用回调记录 action、arg1、arg2 和 source，再只对明确禁止的动作返回 SQLITE_DENY；它适合做 SQLite 连接级的策略门，不等于替代业务授权。

先把拦截点放在 SQLite 编译阶段

set_authorizer() 绑定在一个 sqlite3.Connection 上。SQLite 准备一条语句时，会针对访问表、访问列、修改数据等动作调用回调。Python 文档把回调参数概括为 action、arg1、arg2、database 和 source；其中 source 能帮助定位触发动作的触发器或视图。

这也是它和正则检查的区别：查询写成多行、换了别名，甚至由视图间接读取，动作仍会进入同一个检查点。下面的示例只演示 SQLite 内置连接，不把回调伪装成通用 ORM 中间件。

import sqlite3

connection = sqlite3.connect(":memory:")
connection.execute("create table orders(id integer, amount integer)")

def authorizer(action, arg1, arg2, database, source):
    print(action, arg1, arg2, database, source)
    return sqlite3.SQLITE_OK

connection.set_authorizer(authorizer)
connection.execute("select id, amount from orders")

最小策略只拦截明确的写入动作

日志里看到动作码后，再建立窄范围规则。策略回调不需要解析 SQL 字符串，而是判断 action 与 arg1。例如只读报表连接可以拒绝 SQLITE_INSERT、SQLITE_UPDATE、SQLITE_DELETE 和 SQLITE_DROP_TABLE，普通的 SQLITE_SELECT 则返回 SQLITE_OK。

WRITE_ACTIONS = {
    sqlite3.SQLITE_INSERT,
    sqlite3.SQLITE_UPDATE,
    sqlite3.SQLITE_DELETE,
    sqlite3.SQLITE_DROP_TABLE,
}

def readonly_authorizer(action, arg1, arg2, database, source):
    if action in WRITE_ACTIONS:
        return sqlite3.SQLITE_DENY
    return sqlite3.SQLITE_OK

connection.set_authorizer(readonly_authorizer)
try:
    connection.execute("delete from orders where id = 1")
except sqlite3.DatabaseError as error:
    print(error)

可见结果是写语句在准备阶段就失败，读语句仍可正常返回。这里别急着把所有未知动作都拒绝：SQLite 版本、视图和触发器可能带来额外动作，先记录再收紧规则更容易定位兼容问题。

SQLITE_DENY 与 SQLITE_IGNORE 不是一回事

SQLITE_DENY 表示当前 SQL 不允许继续，Python 侧通常会收到 sqlite3.DatabaseError。SQLITE_IGNORE 更细：对列读取动作返回它时，该列值会以 NULL 参与当前语句，其余列仍可读。这个差异适合做脱敏式只读视图，但不适合掩盖业务层必须存在的字段。

回调还可能收到数据库名和 source。生产代码应把这些字段连同连接用途写入审计日志，并避免在回调里再次查询同一个连接，否则容易让检查逻辑变得递归且难以排错。

三个边界决定它能不能进生产

回调是连接级状态

策略只影响已经调用 set_authorizer() 的连接。连接池新建连接时必须重复绑定，关闭或替换连接后也要重新核对策略是否存在。

它检查 SQLite 动作，不识别业务身份

回调知道表、列、动作和 source，却不知道当前请求是否属于某个用户、租户或订单角色。身份判断应在业务层完成，再决定使用哪种连接或允许哪类操作。

未知动作要留证再处理

推荐先记录未知 action 与对象名，在测试数据覆盖视图、触发器和常用查询后再决定放行或拒绝。把未知值直接当成拒绝，可能让正常查询在升级后突然失败；把未知值全部放行，则会削弱只读策略。

用一条正反查询完成验收

验收至少覆盖一次读和一次写，并检查回调确实命中过目标动作。读查询应返回订单行，写查询应在 connection.execute() 处失败；如果两者都成功，先检查是否把 authorizer 绑到了另一个连接，或者规则集合漏掉了实际动作码。

connection.set_authorizer(readonly_authorizer)
rows = connection.execute("select id from orders").fetchall()
assert rows == []

try:
    connection.execute("update orders set amount = 10 where id = 1")
except sqlite3.DatabaseError:
    pass
else:
    raise AssertionError("write action was not denied")

这段检查关注的是策略结果，不是某个固定错误文案。这样在 Python 小版本或 SQLite 构建差异下，验收仍然稳定。

相关问题

set_authorizer 能替代参数化查询吗？

不能。参数化查询解决值的绑定问题，authorizer 解决连接级动作策略，两者应该同时使用。

为什么回调没有在每个字符或每个条件上触发？

它面向 SQLite 的语义动作，例如读表、读列和写入，不是 SQL 文本扫描器，所以不会按字符或布尔条件逐项报告。

可以用 SQLITE_IGNORE 隐藏整张表吗？

不应这样设计。它更适合列读取场景；整表禁止应返回 SQLITE_DENY，并在业务层限制可访问的连接。

小结

把 set_authorizer() 当作 SQLite 连接上的一道窄门：先记录动作，再对明确的写入或危险动作返回 SQLITE_DENY，需要列级隐藏时才考虑 SQLITE_IGNORE。最后用读写正反用例验证绑定范围，并把身份和租户授权留在业务层。