Python 用 pymysql 写 SQL 如何规避 SQL 注入：从参数化查询到 TaoToken 配置骨架

1. 为什么 pymysql 里拼接 SQL 这么容易出事先说结论SQL 注入不是数据库的锅也不是 pymysql 的锅而是「把用户输入当成 SQL 语法的一部分」这件事本身有问题。你写sql select * from user where name name 数据库看到的是一条已经拼好的完整语句它没法区分哪段是你写的逻辑、哪段是用户塞进来的内容。用户输入 or 11整条语句的语义就被改写了。我见过太多项目在本地跑得好好的上线后被人用一条union select把整张用户表拖走。根因往往就是登录、搜索、分页这几个高频入口用了字符串拼接。pymysql 其实早就给了标准答案——参数化查询也就是占位符%s只是很多人没搞清它和 Python 字符串格式化的区别。这篇聚焦 Python pymysql 场景把注入成因、参数化写法、拼接风险对比讲透再给一套可复制的 config 骨架把数据库连接和统一 Key/API 通道的配置收拢到一处。适合正在写后端接口、做数据脚本或者接手了老项目想补安全漏洞的同学。读完你能直接改掉手里的拼接代码并跑通一次验证请求。2. 先搞懂 pymysql 参数化查询到底做了什么2.1 占位符不是字符串替换关键认知cursor.execute(sql, args)里的%s不是 Python 的%格式化也不是 f-string。它是 pymysql 交给数据库驱动的「参数占位符」。执行时SQL 语句结构和参数值是分两条路走的语句先被数据库解析、编译成执行计划参数再作为纯数据填进去。哪怕参数里写着; drop table user; --数据库也只把它当成一个普通字符串值不会当语法执行。这就是参数化查询能防注入的本质——数据和指令分离。你不需要自己转义引号也不需要过滤关键字驱动层已经处理好了。2.2 正确写法长什么样import pymysql conn pymysql.connect( host127.0.0.1, port3306, userroot, password123, databasepooldb, charsetutf8mb4, ) cursor conn.cursor() # 正确占位符 参数元组 sql select id, name from td where id %s cursor.execute(sql, (5,)) result cursor.fetchall() print(result) cursor.close() conn.close()注意几个细节。第一%s不管字段是整数还是字符串统一用它不要因为 id 是 int 就写成%d。第二参数用元组(5,)或列表[5]都行但单元素元组那个逗号不能省否则会被当成普通括号。第三execute的第二个参数是独立传入的绝不能自己先拼进 sql 字符串。2.3 多参数和 IN 查询怎么写# 多条件 sql select * from user where name %s and pwd %s cursor.execute(sql, (name, pwd)) # IN 查询占位符数量要动态生成但值仍走参数 ids [1, 2, 3, 4] placeholders ,.join([%s] * len(ids)) sql fselect * from td where id in ({placeholders}) cursor.execute(sql, ids)这里f-string只用来拼占位符的个数拼进去的是%s这种符号不是用户数据所以安全。真正危险的永远是「把变量值拼进 SQL 文本」。3. 错误拼接示例这些写法必须改掉3.1 三种典型翻车写法# 翻车一直接字符串拼接 name input(name: ) sql select * from user where name name cursor.execute(sql) # 输入 or 11 直接绕过 # 翻车二Python % 格式化 sql select * from user where name %s % name cursor.execute(sql) # 翻车三f-string 拼接 sql fselect * from user where name {name} cursor.execute(sql)这三种写法在数据库眼里完全一样一条已经成型的语句。用户输入admin --后面的条件就被注释掉了密码校验形同虚设。3.2 表名、字段名不能参数化怎么办参数化只能用于「值」不能用于表名、字段名、order by后面的列名。如果这些来自用户输入必须用白名单校验ALLOWED_ORDER {id, name, created_at} order_by request.args.get(order_by, id) if order_by not in ALLOWED_ORDER: order_by id sql fselect * from td order by {order_by} limit %s cursor.execute(sql, (20,))白名单是这里唯一靠谱的做法别想着用正则去过滤绕过方式太多。3.3 批量插入也要走参数化rows [(alice, 20), (bob, 22)] sql insert into user(name, age) values(%s, %s) cursor.executemany(sql, rows) conn.commit()executemany同样支持占位符比循环拼接快得多也安全得多。4. TaoToken 配置骨架把 Key 和连接收拢到一处4.1 为什么要在项目里加这层写脚本时把数据库密码、API Key 硬编码在代码里是另一个高频坑。一旦代码进仓库密钥就泄露了。我的做法是搞一个统一的 config 模块数据库连接参数和模型调用的 Key 都从环境变量读代码里只留结构。TaoToken 提供统一的 Key/API 通道正好适合放进这个骨架里后面接模型对话、写 SQL 辅助脚本都走同一个入口。官网入口在这里https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content4.2 config.py 骨架import os import pymysql class Config: # 数据库 DB_HOST os.getenv(DB_HOST, 127.0.0.1) DB_PORT int(os.getenv(DB_PORT, 3306)) DB_USER os.getenv(DB_USER, root) DB_PASSWORD os.getenv(DB_PASSWORD, ) DB_NAME os.getenv(DB_NAME, pooldb) DB_CHARSET utf8mb4 # TaoToken 统一通道 TAOTOKEN_API_BASE os.getenv(TAOTOKEN_API_BASE, https://taotoken.net/api) TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY, ) def get_conn(): return pymysql.connect( hostConfig.DB_HOST, portConfig.DB_PORT, userConfig.DB_USER, passwordConfig.DB_PASSWORD, databaseConfig.DB_NAME, charsetConfig.DB_CHARSET, cursorclasspymysql.cursors.DictCursor, )cursorclass设成DictCursor后fetchall返回的是字典列表字段名直接可读写接口时省一层转换。4.3 用上下文管理器管连接from contextlib import contextmanager contextmanager def db_cursor(): conn get_conn() try: with conn.cursor() as cursor: yield cursor conn.commit() except Exception: conn.rollback() raise finally: conn.close()调用时with db_cursor() as cur: cur.execute(select * from td where id %s, (5,)) print(cur.fetchall())连接自动提交、异常自动回滚、用完自动关闭比手写close()稳。4.4 环境变量怎么设export DB_HOST127.0.0.1 export DB_USERroot export DB_PASSWORDyour_password export DB_NAMEpooldb export TAOTOKEN_API_KEYyour_key_hereKey 的获取和接入细节可以看接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content5. 验证请求确认参数化真的生效5.1 用注入字符串做一次自测改完代码别急着上线先自己打一枪。准备一个测试接口或脚本把参数传成注入串with db_cursor() as cur: payload or 11 cur.execute(select * from user where name %s, (payload,)) rows cur.fetchall() print(命中行数:, len(rows))如果参数化生效数据库会把 or 11当成一个普通用户名去匹配命中行数应该是 0除非真有人叫这名字。如果返回了全表数据说明你的代码某处还在拼接赶紧回头查。5.2 对比拼接写法的结果# 危险写法仅用于本地验证别上线 with db_cursor() as cur: sql select * from user where name %s % payload cur.execute(sql) print(拼接写法命中行数:, len(cur.fetchall()))本地跑一次你会直观看到拼接写法返回全表参数化返回 0 行。这个对比比任何文档都有说服力。5.3 验证 TaoToken 通道连通import requests resp requests.get( f{Config.TAOTOKEN_API_BASE}/models, headers{Authorization: fBearer {Config.TAOTOKEN_API_KEY}}, timeout10, ) print(resp.status_code) print(resp.json())返回 200 且能看到模型列表说明 Key 和通道都通了。后续想让模型帮你审查 SQL 写法、生成参数化模板都从这个入口走。模型对话入口https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content6. 本篇常见错排查6.1 报错 “not enough arguments for format string”原因通常是 SQL 里出现了%但没传够参数或者你用了%格式化又叠加了占位符。比如like %s%这种写法pymysql 会把%当占位符解析。正确做法是把通配符放进参数值里keyword %python% cur.execute(select * from td where name like %s, (keyword,))6.2 单元素元组漏逗号cur.execute(sql, (5))里的(5)是整数不是元组会报参数数量不匹配。写成(5,)才对。这个坑我踩过不止一次。6.3 中文乱码连接时charset设成utf8mb4别用utf8。utf8在 MySQL 里只支持 3 字节存 emoji 会报错。建表时字符集也统一成utf8mb4。6.4 连接超时或断连长时间运行的脚本连接可能被数据库回收。加ping或改用连接池conn pymysql.connect(..., autocommitTrue) conn.ping(reconnectTrue)生产环境建议上DBUtils的PooledDB避免频繁建连。6.5 事务没提交pymysql 默认不开 autocommitinsert、update后忘了conn.commit()数据不会落库。用前面那个db_cursor上下文管理器就不会漏。6.6 把 Key 写进代码提交了一旦提交密钥就等于公开。立刻去控制台轮换 Key然后改用环境变量。控制台入口https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content7. 把防注入变成习惯顺手接上统一通道参数化查询这件事说穿了就一句话SQL 文本里只放结构值永远走execute的第二个参数。表名、字段名这类没法参数化的用白名单。把这两条守住pymysql 场景下 90% 的注入风险就没了。配置层面把数据库连接和 TaoToken 的 Key 都收进 config 模块从环境变量读代码里不出现明文密钥。这样无论是写数据脚本、做接口还是让模型辅助生成 SQL都走同一个入口改一处全局生效。API Key 管理页面在这里https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content如果你在写长期跑的编码任务或者 Agent 类项目需要稳定的模型调用通道可以看看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content最后留个实操建议把你项目里所有cursor.execute调用搜一遍凡是第二个参数为空的逐个改成参数化写法。改完用第 5 节的注入串自测一遍命中行数为 0 才算过关。