FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · 严重
安全研究 · FlawOpen

Python 中的 SQL 注入（psycopg2、SQLAlchemy、Django）

在 cursor.execute() 里使用 f-string 或 % 拼接，会让请求参数改写你的 SQL，把查询单个用户变成导出整张表（CWE-89）。本文给出 Python 各数据库 API 的安全写法，以及占位符覆盖不到的两种情况：列名和 IN 列表。[1], [2], [3], [4], [5], [6], [7], [9]

💡 通俗易懂的原理解析 (ELI5)

想象一个邮政分拣台，桌上有一张固定指令卡：“分拣标签为 ___ 的包裹”。如果工作人员先把寄件人的标签抄进指令卡再阅读，多出来的文字就可能变成新的分拣指令。Python 的 f-string 或 % 格式化对 SQL 做的正是这件事：数据库驱动收到值之前，程序已经拼成了一整条命令。execute(sql, params) 保持指令卡不变，并把标签作为数据单独发送。列名属于指令本身，应从代码定义的允许列表中选择。[10], [1], [2], [9]

核心概念与专有名词

cursor.execute(sql, params)
DB-API 2.0（PEP 249）定义的调用。驱动会把 params 里的值与 SQL 文本分开发送，因此这些值永远改变不了语句本身。只有当值放在第二个参数里时才安全；在第一个参数里用 f-string 或 % 表达式，就是漏洞所在。[2], [3]
paramstyle: %s, ?, :name
每个驱动都在 paramstyle 属性中声明自己的占位符：psycopg2 和 PyMySQL 用 %s 或 %(name)s，sqlite3 用 ? 或 :name。%s 由驱动解析，而不是 Python 的 % 运算符，所以绝不要自己去格式化字符串。[2], [3], [5]
SQLAlchemy text()
SQLAlchemy 中用来包装原始 SQL 字符串的函数。当值以 :email 这类命名绑定参数的形式出现、并以字典传给 execute() 时是安全的。如果传进去的字符串本身已经是 f-string 就不安全：text() 不会转义字符串里的任何内容。[6], [9]
Django raw() / RawSQL / extra()
Django 绕开 ORM 的几个出口。raw(sql, params) 和 RawSQL(sql, params) 只有在所有值都放进 params 时才安全；插值进 sql 的内容 Django 会原样发送。Django 文档把 extra() 称为最后手段，代码评审时应直接禁止。[7], [9]
psycopg2.sql.Identifier
占位符只能承载值，不能承载表名或列名。sql.Identifier 会给名称加上双引号，使其无法跳出标识符的范围。在此之前先用白名单校验名称，防止用户选中你本不打算公开的列。[4], [9]
IN lists and expanding binds
sqlite3 无法把一个 Python 列表绑定到单个 ? 上，于是很多人又回到字符串拼接。SQLAlchemy 中请使用可展开的 bindparam（每个元素一个占位符），psycopg2 中对列表使用 id = ANY(%s)，sqlite3 则为每个元素生成一个 ?。[6], [3], [5]

框架映射

攻击执行流程分解

步骤 1

请求中的值进入数据层

请求参数以文本形式到达 get_user()。Flask 视图可以读取 request.args["id"]；Django 视图读取 request.GET["id"]。视图会把未经验证的文本传给 get_user()。[11], [12], [9]

步骤 2

f-string 把值拼进 SQL

f"... WHERE id = {user_id}" 生成一个扁平的字符串。使用无害的测试值 0 OR 1=1 时，文本变成 WHERE id = 0 OR 1=1，而且没有任何标记说明这个值本该在哪里结束。[10], [1], [9]

步骤 3

驱动没有收到任何参数

调用 execute(query) 时只传了 SQL，于是 psycopg2 原样发送文本，PostgreSQL 把 OR 1=1 解析为 WHERE 子句的一部分，而不是数据。[3], [1], [9]

步骤 4

接口返回了所有行

条件对每一行都成立，所以查询单个用户的接口返回了整张表。psycopg2 还允许在一次调用中执行多条以分号分隔的语句；sqlite3 会拒绝这种写法，这能减小损失，但阻止不了数据被窃取。[13], [5], [1], [9]

源代码对比：漏洞与安全实现

存在漏洞的实现
# repository.py：下面每个查询都把请求数据拼进了 SQL 文本
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2：驱动还没看到语句，f-string 就已经把它拼好了
    pg_cursor.execute(f"SELECT id, email FROM users WHERE id = {user_id}")
    return pg_cursor.fetchone()


def find_orders(conn: sqlite3.Connection, status):
    # sqlite3：这里的 % 是 Python 字符串格式化，不是驱动的占位符
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy：text() 不会转义字符串里已有的内容
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django：raw() 会把插值进去的值原样发给数据库
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # IN 列表：拼接后的字符串让某个元素能夹带额外的 SQL
    id_list = ", ".join(ids)
    return session.execute(text(f"SELECT id FROM orders WHERE id IN ({id_list})")).all()


def list_users(pg_cursor, sort_column):
    # 标识符：排序列直接取自查询字符串
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
加固后的安全修复
# repository.py：值以绑定参数传递，名称取自白名单
import sqlite3

from psycopg2 import sql
from sqlalchemy import bindparam, text

SORTABLE_COLUMNS = {"id", "email", "created_at"}


def get_user(pg_cursor, user_id: int):
    # psycopg2：%s 是驱动的占位符，值放在第二个参数里
    pg_cursor.execute("SELECT id, email FROM users WHERE id = %s", (user_id,))
    return pg_cursor.fetchone()


def find_orders(conn: sqlite3.Connection, status: str):
    # sqlite3 使用 qmark 风格：? 占位符，值放在元组中
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy：命名绑定参数，值在调用 execute() 时提供
    stmt = text("SELECT id FROM users WHERE email = :email")
    return session.execute(stmt, {"email": email}).all()


def find_users(User, username: str):
    # Django：raw() 单独发送 params；用 ORM 过滤器更简单
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # IN 列表：可展开的绑定为每个元素生成一个占位符
    stmt = text("SELECT id FROM orders WHERE id IN :ids").bindparams(
        bindparam("ids", expanding=True)
    )
    return session.execute(stmt, {"ids": ids}).all()


def list_users(pg_cursor, sort_column: str):
    # 标识符：先查白名单，再用 sql.Identifier 加引号
    if sort_column not in SORTABLE_COLUMNS:
        raise ValueError(f"unsupported sort column: {sort_column!r}")
    query = sql.SQL("SELECT id, email FROM users ORDER BY {}").format(sql.Identifier(sort_column))
    pg_cursor.execute(query)
    return pg_cursor.fetchall()

工程与系统安全加固清单

参考来源