Python读取MySQL数据时怎么防止SQL注入攻击？

一、SQL 注入的根源与数据读取场景下的危害

SQL 注入攻击之所以发生，是因为应用程序把不可信的外部输入直接当成了 SQL 代码的一部分去拼装。当拼接后的字符串送入 MySQL 语法解析器时，攻击者精心构造的特殊字符（例如单引号、注释符 -- 或关键字 OR、UNION）会改变原有的语法树结构，使得本应作为纯字面量比对的数据变成了控制逻辑。

很多开发者误以为“只有写入、更新或删除数据时注入才致命，读取操作只是查查数据，风险不大”。这是非常危险的误解。在数据读取场景下，SQL 注入会导致以下直接后果：

• 整表数据泄露（永真条件）：通过输入 ' OR '1'='1，让原本过滤特定用户的 WHERE 条件失效，直接返回包含所有用户账号、密码哈希或敏感隐私的完整结果集。
• 联合查询越权（UNION SELECT）：在查询语句后追加 UNION SELECT，将其他系统表（如 information_schema）或核心业务表中的敏感数据一并输出到查询结果中。
• 带外数据渗漏与盲注：即便前端界面只返回空或非空状态，攻击者也可以通过盲注技术（布尔盲注或时间延迟盲注）逐字探测出数据库内的任意数据。

# 错误示例 1：使用 f-string 拼接
cursor.execute(f"SELECT id, username, email FROM users WHERE username = '{user_input}'")

# 错误示例 2：使用加号字符串拼接
cursor.execute("SELECT id, username, email FROM users WHERE username = '" + user_input + "'")

# 错误示例 3：使用 Python 原生 % 运算符拼接字符串后再执行
sql = "SELECT id, username, email FROM users WHERE username = '%s'" % user_input
cursor.execute(sql)

二、核心解决方案：使用数据库驱动的参数化查询

在 Python 的标准数据库驱动（如 PyMySQL、mysqlclient 或官方的 mysql-connector-python）中，cursor.execute() 方法原生支持两个参数：第一个是 SQL 语句模板，第二个是绑定的参数元组或字典。

1. 占位符的语法规范

MySQL 驱动普遍遵循 Python DB-API 2.0 规范，使用 %s 作为参数占位符。注意以下关键细节：

• 不要给 %s 加引号：写成 WHERE username = %s，绝对不要写成 WHERE username = '%s'。驱动会自动根据传入变量的数据类型（字符串、数字、时间等）处理引号和转义。
• 参数必须是可迭代序列：当只有一个参数时，必须写成元组形式 (user_input,)，注意结尾的英文逗号；写成 (user_input) 会被 Python 误判为普通括号包裹的字符串，进而引发类型解构异常。

2. 规范的参数化查询代码实现

下面给出使用 pymysql 库安全读取数据的标准范例。代码结合了上下文管理器（with 语句），确保游标和连接无论查询成功或发生异常都能妥善关闭：

import pymysql
from pymysql.cursors import DictCursor

def get_user_by_username(target_username: str) -> dict | None:
    # 数据库连接配置（生产环境推荐从环境变量中读取配置）
    conn_params = {
        "host": "127.0.0.1",
        "port": 3306,
        "user": "readonly_user",       # 建议使用专用的只读用户
        "password": "StrongPassword123!",
        "database": "production_db",
        "charset": "utf8mb4",
        "cursorclass": DictCursor      # 返回字典格式的结果，便于字段读取
    }

    # 1. 静态 SQL 模板：参数位置明确使用 %s 占位符
    sql_template = """
        SELECT id, username, email, created_at, status 
        FROM users 
        WHERE username = %s AND status = %s;
    """
    
    # 2. 准备参数列表（元组形式），对应 SQL 中的每个 %s
    query_params = (target_username, "active")

    connection = pymysql.connect(**conn_params)
    try:
        with connection.cursor() as cursor:
            # 3. 关键点：将 sql_template 与 query_params 分开传递给 execute
            cursor.execute(sql_template, query_params)
            result = cursor.fetchone()
            return result
    finally:
        connection.close()

# 模拟攻击性输入测试
malicious_input = "admin' OR '1'='1"
user_record = get_user_by_username(malicious_input)
# 结果将安全地返回 None，因为数据库中不存在用户名恰好是 "admin' OR '1'='1" 的用户
print("查询结果:", user_record)

3. 特殊查询场景：IN 查询的参数化处理

很多开发者在遇到动态 IN 查询时，容易倒退回字符串拼接（如 f"WHERE id IN ({ids_str})"）。实际上，IN 查询也可以且必须使用参数化方式：

def get_users_by_ids(id_list: list[int]) -> list[dict]:
    if not id_list:
        return []

    # 1. 动态生成与列表长度一致的 %s 占位符串，例如 "%s, %s, %s"
    placeholders = ", ".join(["%s"] * len(id_list))
    sql = f"SELECT id, username FROM users WHERE id IN ({placeholders})"
    
    with connection.cursor() as cursor:
        # 2. 将 ID 列表作为整体参数传入
        cursor.execute(sql, tuple(id_list))
        return cursor.fetchall()

在这个实现中，通过 ["%s"] * len(id_list) 生成的仅仅是固定格式的占位符符号，外部数据依然作为第二个参数传递，绝不会破坏语法树结构。

三、无法使用占位符时的防御：白名单机制

参数化查询（%s）仅适用于数据值（Literal Values），无法用于 SQL 的语法结构标识符，包括：

• 表名（如 FROM {table_name}）
• 列名（如 SELECT {column_name}）
• 排序字段与方向（如 ORDER BY {sort_col} {sort_order}）

如果业务确实需要前端动态指定排序字段或排序列，绝对不能直接将参数拼入 SQL，必须建立严格的服务器端白名单映射。

动态排序列的安全代码模式

def fetch_users_with_sort(order_by: str, direction: str) -> list[dict]:
    # 1. 严格定义允许参与排序的字段白名单（映射字典）
    ALLOWED_SORT_COLUMNS = {
        "id": "id",
        "name": "username",
        "created": "created_at",
        "score": "total_score"
    }

    # 2. 严格限定排序方向
    ALLOWED_DIRECTIONS = {"asc": "ASC", "desc": "DESC"}

    # 3. 校验并提取安全列名，未命中时抛出异常或使用默认值
    safe_column = ALLOWED_SORT_COLUMNS.get(order_by.lower())
    if not safe_column:
        raise ValueError(f"非法的排序列名: {order_by}")

    safe_direction = ALLOWED_DIRECTIONS.get(direction.lower(), "ASC")

    # 4. 白名单值是受控的纯净标识符，可安全放入 SQL；数据过滤条件仍然使用 %s
    sql = f"SELECT id, username, created_at FROM users WHERE status = %s ORDER BY `{safe_column}` {safe_direction};"

    with connection.cursor() as cursor:
        cursor.execute(sql, ("active",))
        return cursor.fetchall()

四、纵深防御：数据读取场景的综合加固策略

单一维度的防护容易因疏忽产生漏洞。构建多层防线，才能确保即使个别开发人员编写了不规范的代码，也不会引发灾难性后果。

1. 数据库账号最小权限原则（Least Privilege）

用于在业务中执行数据读取的数据库连接，账号权限应当收敛至仅具备 SELECT 权限。严禁为应用的数据读取模块分配 root、拥有 DROP、ALTER、FILE 或跨库访问权限的超级账号。

-- 在 MySQL 中为只读应用分配极小权限
CREATE USER 'app_reader'@'192.168.1.%' IDENTIFIED BY 'StrongReadPwd_456!';
GRANT SELECT ON production_db.* TO 'app_reader'@'192.168.1.%';
FLUSH PRIVILEGES;

即使在极端情况下某处查询遭遇了堆叠注入攻击（Stacked Queries），只读账号也无法执行 DROP TABLE 或更新篡改数据，大大降低损失上限。

2. 输入类型强制转换与前置校验

在进入数据库操作层之前，利用 Python 的类型系统或校验框架（如 pydantic）进行强类型转换。例如分页查询的 page 和 page_size 必须强制转换为 int 类型：

try:
    page = int(request.args.get("page", 1))
    page_size = int(request.args.get("page_size", 20))
except ValueError:
    return {"error": "分页参数必须为整数"}, 400

offset = max(0, (page - 1) * page_size)
# 整数类型保证了该变量绝无包含 SQL 注入语句的可能
cursor.execute("SELECT id, username FROM users LIMIT %s, %s", (offset, page_size))

3. 关闭敏感的数据库底层报错回显

许多基于报错的 SQL 注入（Error-Based SQL Injection）依赖于数据库引擎直接抛出的具体错误信息（如表结构、字段名）。在生产环境中：

• 捕获 pymysql.MySQLError 或通用数据库异常；
• 将详细堆栈日志记录至内部日志审计系统；
• 仅向最终调用方返回通用的错误提示（如“数据查询失败，请稍后重试”），避免技术细节外泄。