pydantic不防sql注入,真正起作用的是参数化查询、白名单校验和零字符串拼接;asyncpg需显式用execute()/fetch()配合占位符,动态标识符须硬校验,路径参数和搜索字段需额外约束。

Pydantic 不防 SQL 注入,asyncpg 本身也不防——真正起作用的只有参数化查询 + 白名单校验 + 零字符串拼接。
asyncpg 的参数化查询必须显式用 execute() 或 fetch() 配合命名/位置占位符
asyncpg 不会自动把 Pydantic 模型字段塞进 SQL;你写错一句,就等于开了一扇后门。
- 错误写法:
await conn.fetch(f"SELECT * FROM users WHERE name = '{user.name}'")—— 单引号包裹、f-string 拼接,' OR 1=1 --直接穿透 - 正确写法(命名参数):
await conn.fetch("SELECT * FROM users WHERE name = $1", user.name)或await conn.fetch("SELECT * FROM users WHERE name = :name", {"name": user.name}) - 动态列名/表名无法参数化,必须硬校验:
if order_by not in ["created_at", "name"]: raise HTTPException(400) - IN 子句要展开占位符:
placeholders = ','.join(f'${i}' for i in range(1, len(ids)+1)),再拼成f"WHERE id IN ({placeholders})",最后传(*ids)
Pydantic 模型只做“预筛”,不是防火墙
它拦不住合法格式的恶意 payload,比如 "admin'::text || (SELECT password FROM users LIMIT 1)::text--" —— 这仍是 str,也符合 max_length=50。
- 路径参数(如
/user/{id})根本不会过 Pydantic 校验,id: int是 FastAPI 自动转的,但若你手动用str(id)拼 SQL,照样中招 - 搜索字段(
q: str)建议加Field(max_length=200, strip_whitespace=True),再手动截断:q = q[:200] - 白名单优先于长度限制:用户名用
pattern=r'^[a-zA-Z0-9_]{2,30}$',但邮箱、中文名等业务字段不能这么干,得靠后续参数化兜底
asyncpg 连接池 + 依赖注入场景下,最容易漏掉的两个注入点
连接池本身不改变 SQL 安全性,但会让错误更隐蔽、更难排查。
- 在
get_db依赖里,如果用了yield conn但没 await 获取连接(比如写成conn = pool.acquire()),FastAPI 注入的是协程对象,后续conn.fetch()会静默失败或报TypeError,导致你误以为“没查到数据”而补上 fallback 字符串拼接逻辑 - 事务块中混用同步操作:
await session.execute(text("UPDATE logs SET msg = '" + user_input + "'"))——text()不等于安全,没用命名参数就是裸拼 - JSON 字段存用户输入前未
json.dumps()转义,可能带入控制字符或破坏结构,虽不属传统 SQL 注入,但可触发解析异常或服务端模板渲染漏洞
最常被忽略的一点:SQL 注入防护和异步无关。asyncpg 的 $1 占位符机制和 psycopg2 的 %s 在原理上完全一致,区别只在并发模型。别因为用了 async 就放松对字符串拼接的警惕——那才是真正的单点崩塌口。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











