sqlalchemy 默认防sql注入,因所有查询接口强制参数化绑定;危险在于字符串拼接,尤其动态字段名、表名、排序方向等结构信息须白名单校验。

SQLAlchemy 本身就能防 SQL 注入,前提是别绕过它去拼接 SQL 字符串。只要你用 session.execute() + 参数化查询,或 ORM 的 filter() / where() 方法传参,就基本安全。真正危险的是手动拼接 str.format()、% 或 f-string 构造查询语句。
为什么 ORM 查询默认安全?
SQLAlchemy(包括异步版)所有公开的查询接口都强制使用参数化绑定。比如:
await session.execute(select(User).where(User.email == email))
这里的 email 不会进 SQL 字符串,而是作为独立参数交给数据库驱动(如 asyncpg 或 aiomysql)处理,底层走的是预编译语句(prepared statement)路径。
常见错误现象:sqlalchemy.exc.ProgrammingError: (psycopg2.errors.SyntaxError) syntax error at or near "admin@example.com" —— 这往往说明你用了字符串拼接,而不是参数传递。
- ORM 层的
filter()、where()、order_by()等方法只接受表达式对象,不接受原始字符串 - 原生 SQL 查询必须用
text()包裹,并显式传参:session.execute(text("SELECT * FROM user WHERE email = :email"), {"email": email}) - 绝对不要写:
text(f"SELECT * FROM user WHERE email = '{email}'")或text("SELECT * FROM user WHERE email = '" + email + "'")
哪些地方容易漏掉防护?
最常被忽略的是动态字段名、表名、排序方向这类“元信息”。SQLAlchemy 不允许把这些当参数传,因为它们不属于值(value),而是结构(structure)。
例如想根据前端传来的 sort_field 和 sort_dir 动态排序:
FastAPI + Flask 混合部署最佳实践,解决路由定义、API 代理等常见问题,适用于同时运行 FastAPI API 与 Flask 前端的场景。
query = select(User).order_by(text(f"{sort_field} {sort_dir}"))
这就开了后门。攻击者传 sort_field=id; DROP TABLE user; -- 就可能触发注入。
- 解决方案:白名单校验字段和方向:
if sort_field not in ["id", "email", "created_at"]:直接raise HTTPException(400) - 排序方向只允许
"asc"或"desc",其他一律拒绝 - 表名、字段名永远不能来自用户输入;必须硬编码或从预定义映射中取
异步模式下有没有额外风险?
没有本质区别,但要注意会话生命周期管理不当可能间接放大问题。比如在依赖中未正确 await 关闭 AsyncSession,导致连接复用混乱,掩盖了本该报错的非法查询。
典型错误写法:
@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
result = await db.execute("SELECT * FROM user WHERE email = '" + email + "'") # ❌ 拼接
return result.scalars().all()
这段代码即使用了 AsyncSession,也完全失去防护能力。
- 务必确认
get_db依赖返回的是AsyncSession实例,且你在execute()前没做任何字符串拼接 - 异步驱动(如
asyncpg)对参数绑定的要求更严格,错误拼接更容易直接报错,反而暴露问题 - 生产环境建议开启
echo=True(仅开发)观察生成的 SQL,确认参数是否以$1、:email形式出现,而非明文嵌入
真正的难点不在“怎么写安全”,而在于“什么时候不该交由用户控制结构”。字段名、操作符、JOIN 表顺序这些,从来就不该是 API 的自由输入项。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!










