Python sqlite3.Connection.set_authorizer 方法

上次修改时间：2025 年 4 月 15 日

本综合指南探讨了 Python 的 sqlite3.Connection.set_authorizer 方法，它提供了对数据库操作的细粒度控制。我们将介绍其目的、回调机制和实际安全应用。

基本定义

set_authorizer 方法注册一个授权回调，SQLite 在尝试执行数据库操作时会调用该回调。这使得可以实现自定义安全策略。

主要特点：它拦截所有 SQL 操作，允许或拒绝操作，并提供有关每个操作的详细上下文。回调返回 sqlite3.SQLITE_OK 以允许，或返回其他代码以拒绝。

基本授权示例

此示例显示了一个简单的授权器，它记录所有数据库操作而不限制任何操作。

basic_authorizer.py

import sqlite3

def authorizer_callback(action, arg1, arg2, dbname, source):
    print(f"Action: {action}, Table: {arg1}, Column: {arg2}")
    return sqlite3.SQLITE_OK

with sqlite3.connect(":memory:") as conn:
    conn.set_authorizer(authorizer_callback)
    cursor = conn.cursor()
    
    cursor.execute("CREATE TABLE users (id INTEGER, name TEXT)")
    cursor.execute("INSERT INTO users VALUES (1, 'Alice')")
    cursor.execute("SELECT * FROM users")
    
    print(cursor.fetchall())

授权器回调接收有关每个操作的详细信息。在这种情况下，它只是记录操作，同时通过返回 SQLITE_OK 来允许所有操作。

这种模式对于审计数据库活动或调试具有许多 SQL 操作的复杂应用程序很有用。

只读数据库保护

此示例通过授权器回调拒绝所有写入操作来实现只读数据库。

readonly_protection.py

import sqlite3

def read_only_authorizer(action, arg1, arg2, dbname, source):
    # Allow SELECT operations
    if action == sqlite3.SQLITE_SELECT:
        return sqlite3.SQLITE_OK
    # Deny all other operations
    return sqlite3.SQLITE_DENY

with sqlite3.connect(":memory:") as conn:
    conn.set_authorizer(read_only_authorizer)
    cursor = conn.cursor()
    
    # This will work
    cursor.execute("SELECT 1")
    print(cursor.fetchone())
    
    # This will fail
    try:
        cursor.execute("CREATE TABLE test (id INTEGER)")
    except sqlite3.DatabaseError as e:
        print(f"Operation denied: {e}")

回调检查操作类型，并且仅允许 SELECT 语句。通过返回 SQLITE_DENY 拒绝所有其他操作，这将引发 DatabaseError。

当您需要通过防止修改来确保数据完整性时，例如在报告应用程序中，此方法很有用。

表特定权限

此示例演示了通过基于表名限制对特定表的访问来实现表级别的权限。

table_permissions.py

import sqlite3

def table_authorizer(action, arg1, arg2, dbname, source):
    # Block all access to 'sensitive_data' table
    if arg1 == "sensitive_data":
        return sqlite3.SQLITE_DENY
    # Allow everything else
    return sqlite3.SQLITE_OK

with sqlite3.connect(":memory:") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE users (id INTEGER, name TEXT)")
    cursor.execute("CREATE TABLE sensitive_data (ssn TEXT, salary REAL)")
    
    conn.set_authorizer(table_authorizer)
    
    # This will work
    cursor.execute("SELECT * FROM users")
    
    # This will fail
    try:
        cursor.execute("SELECT * FROM sensitive_data")
    except sqlite3.DatabaseError as e:
        print(f"Access denied: {e}")

授权器检查表名 (arg1)，并阻止对 'sensitive_data' 表的所有访问，同时允许对其他表的操作。

此模式对于实现行级安全性或多租户应用程序很有价值，在这些应用程序中，不同的用户应该可以访问不同的数据子集。

列级保护

此示例显示了如何在允许访问其他列的同时保护表中的特定列，从而实现列级别的安全性。

column_protection.py

import sqlite3

def column_authorizer(action, arg1, arg2, dbname, source):
    # Block access to 'password' column in any table
    if arg2 == "password":
        return sqlite3.SQLITE_DENY
    # Allow everything else
    return sqlite3.SQLITE_OK

with sqlite3.connect(":memory:") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE accounts (id INTEGER, username TEXT, password TEXT)")
    cursor.execute("INSERT INTO accounts VALUES (1, 'admin', 'secret')")
    
    conn.set_authorizer(column_authorizer)
    
    # This will work
    cursor.execute("SELECT id, username FROM accounts")
    print(cursor.fetchall())
    
    # This will fail
    try:
        cursor.execute("SELECT password FROM accounts")
    except sqlite3.DatabaseError as e:
        print(f"Access denied: {e}")

回调检查列名 (arg2)，并阻止对名为 'password' 的任何列的访问，同时允许访问其他列。

此技术对于保护数据库应用程序中的敏感字段（如密码、API 密钥或个人身份信息）至关重要。

操作特定的授权

此示例为不同类型的 SQL 操作实现不同的规则，从而允许对数据库访问进行更精细的控制。

operation_authorization.py

import sqlite3

def operation_authorizer(action, arg1, arg2, dbname, source):
    # Allow all read operations
    if action in (sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ):
        return sqlite3.SQLITE_OK
    # Allow table creation but not modification
    if action == sqlite3.SQLITE_CREATE_TABLE:
        return sqlite3.SQLITE_OK
    # Deny all other operations
    return sqlite3.SQLITE_DENY

with sqlite3.connect(":memory:") as conn:
    conn.set_authorizer(operation_authorizer)
    cursor = conn.cursor()
    
    # This will work (table creation)
    cursor.execute("CREATE TABLE logs (id INTEGER, message TEXT)")
    
    # This will work (reading)
    cursor.execute("SELECT name FROM sqlite_master")
    print(cursor.fetchall())
    
    # This will fail (insert)
    try:
        cursor.execute("INSERT INTO logs VALUES (1, 'test')")
    except sqlite3.DatabaseError as e:
        print(f"Operation denied: {e}")

授权器使用 action 参数区分不同的操作类型。它允许读取操作和表创建，但阻止修改。

此方法对于实现复杂的安全策略很有用，在这种策略中，不同类型的操作需要不同的权限级别。

动态权限检查

此高级示例显示了如何实现可以基于应用程序状态在运行时更改的动态权限。

dynamic_permissions.py

import sqlite3

class DynamicAuthorizer:
    def __init__(self):
        self.read_only_mode = True
    
    def authorize(self, action, arg1, arg2, dbname, source):
        if self.read_only_mode and action != sqlite3.SQLITE_SELECT:
            return sqlite3.SQLITE_DENY
        return sqlite3.SQLITE_OK

with sqlite3.connect(":memory:") as conn:
    authorizer = DynamicAuthorizer()
    conn.set_authorizer(authorizer.authorize)
    cursor = conn.cursor()
    
    # Initial state: read-only
    try:
        cursor.execute("CREATE TABLE test (id INTEGER)")
    except sqlite3.DatabaseError:
        print("Create denied in read-only mode")
    
    # Change to write mode
    authorizer.read_only_mode = False
    cursor.execute("CREATE TABLE test (id INTEGER)")
    print("Table created successfully")
    
    # Verify
    cursor.execute("SELECT name FROM sqlite_master")
    print(cursor.fetchall())

授权器维护影响其决策的内部状态 (read_only_mode)。这允许在运行时动态更改安全策略。

这种模式对于需要在不同时间具有不同访问级别的应用程序（例如，在维护与正常操作期间）非常强大。

组合多个授权规则

此最终示例演示了如何将多个授权检查组合到一个全面的安全策略中。

combined_rules.py

import sqlite3

def comprehensive_authorizer(action, arg1, arg2, dbname, source):
    # Block all DROP operations
    if action == sqlite3.SQLITE_DROP_TABLE:
        return sqlite3.SQLITE_DENY
    
    # Block access to system tables (sqlite_ prefix)
    if arg1 and arg1.startswith("sqlite_"):
        return sqlite3.SQLITE_DENY
    
    # Block schema modifications except by direct connection
    if action in (sqlite3.SQLITE_ALTER_TABLE, sqlite3.SQLITE_CREATE_INDEX):
        if source != "main":
            return sqlite3.SQLITE_DENY
    
    # Allow everything else
    return sqlite3.SQLITE_OK

with sqlite3.connect(":memory:") as conn:
    conn.set_authorizer(comprehensive_authorizer)
    cursor = conn.cursor()
    
    # Allowed operations
    cursor.execute("CREATE TABLE users (id INTEGER)")
    cursor.execute("INSERT INTO users VALUES (1)")
    
    # Blocked operations
    try:
        cursor.execute("DROP TABLE users")
    except sqlite3.DatabaseError as e:
        print(f"Operation denied: {e}")
    
    try:
        cursor.execute("SELECT * FROM sqlite_master")
    except sqlite3.DatabaseError as e:
        print(f"Access denied: {e}")

授权器实现多个安全规则：防止表删除，阻止访问 SQLite 系统表，并限制模式修改。

这种综合方法是生产应用程序的典型方法，在这些应用程序中，必须同时解决多个安全问题。

最佳实践

保持简单：复杂的授权器会影响性能
安全失败：默认为拒绝未知操作
记录决策：记录授权结果以进行审计
彻底测试：验证所有预期的操作都能正常工作
记录策略：清楚地记录您的安全规则

资料来源

作者

我叫 Jan Bodnar，是一位热情的程序员，拥有丰富的编程经验。自 2007 年以来，我一直在撰写编程文章。迄今为止，我已经撰写了 1,400 多篇文章和 8 本电子书。我拥有超过十年的编程教学经验。

列出所有 Python 教程。