1. SQL注入攻击的本质与危害SQL注入SQL Injection是Web应用中最常见的安全漏洞之一攻击者通过在用户输入中插入恶意SQL代码欺骗后端数据库执行非预期的操作。在Python开发的Web应用中这个问题尤为突出因为Python的灵活性使得开发者容易忽视输入验证。SQL注入的核心原理是混淆代码与数据边界。正常情况下用户输入应该被当作数据处理但如果没有正确隔离用户输入可能被解释为可执行代码。举个例子# 危险写法 query SELECT * FROM users WHERE username username AND password password cursor.execute(query)如果攻击者输入admin --作为用户名生成的SQL会变成SELECT * FROM users WHERE username admin -- AND password --在SQL中是注释符号这相当于直接绕过了密码验证。常见的SQL注入攻击类型包括联合查询注入UNION-based布尔盲注Boolean-based blind时间盲注Time-based blind报错注入Error-based堆叠查询Stacked queries这些攻击可能导致数据泄露、数据篡改、权限提升甚至整个数据库被删除。2023年OWASP Top 10中注入类漏洞仍位居前三其中SQL注入占比超过60%。2. Python中SQL注入的典型场景2.1 字符串拼接式查询这是最危险的写法也是新手最容易犯的错误def unsafe_login(username, password): conn sqlite3.connect(users.db) cursor conn.cursor() # 直接拼接SQL语句 - 高危 query fSELECT * FROM users WHERE username{username} AND password{password} cursor.execute(query) return cursor.fetchone()攻击者只需输入 OR 11 --作为用户名就能绕过认证。我在早期项目审计中就发现过某电商平台使用这种写法导致可以任意修改商品价格。2.2 动态表名/列名拼接有时开发者会动态构造表名或列名def get_records(table, column, value): query fSELECT * FROM {table} WHERE {column} {value} cursor.execute(query)这种场景下参数化查询对表名/列名无效必须采用白名单验证。我在一个CMS系统中就遇到过通过table参数注入的情况。2.3 二次注入即使第一次存储时做了转义如果从数据库取出后再次拼接使用仍可能产生注入# 第一次安全地存储评论 def save_comment(content): cursor.execute(INSERT INTO comments VALUES (?), (content,)) # 但后续使用时危险 def show_comment(comment_id): comment cursor.execute(SELECT content FROM comments WHERE id?, (comment_id,)).fetchone() # 取出后直接拼接使用 cursor.execute(fUPDATE stats SET viewsviews1 WHERE item{comment[0]})3. Python中的防御方案3.1 参数化查询最佳实践所有主流Python数据库接口都支持参数化查询# SQLite示例 cursor.execute(SELECT * FROM users WHERE username? AND password?, (username, password)) # MySQL示例使用%s cursor.execute(SELECT * FROM users WHERE username%s AND password%s, (username, password)) # PostgreSQL示例使用%s或%(name)s cursor.execute( SELECT * FROM users WHERE username%(user)s AND password%(pass)s , {user: username, pass: password})参数化查询的原理是将SQL语句与数据分开发送数据库能明确区分代码和数据。我在处理某金融项目时将300多处拼接查询改为参数化后安全扫描漏洞降为零。3.2 ORM框架的安全使用ORM如SQLAlchemy、Django ORM通常自动处理参数化# SQLAlchemy示例 session.query(User).filter(User.username username, User.password password).first() # Django ORM示例 User.objects.get(usernameusername, passwordpassword)但要注意ORM的raw()或extra()方法可能绕过安全机制# 危险用法 - 部分ORM允许原生SQL User.objects.raw(fSELECT * FROM auth_user WHERE username{username})3.3 输入验证与过滤对于必须动态拼接的场景如表名应采用白名单验证ALLOWED_TABLES {users, products, orders} def query_table(table): if table not in ALLOWED_TABLES: raise ValueError(Invalid table name) query fSELECT * FROM {table} LIMIT 10 # 注意这里仍需要确保table不包含危险字符对于搜索功能建议限制通配符使用def safe_search(term): term term.replace(%, \\%).replace(_, \\_) # 转义通配符 cursor.execute(SELECT * FROM products WHERE name LIKE ?, (f%{term}%,))4. 进阶防御措施4.1 最小权限原则数据库用户应遵循最小权限应用账户不应有DROP、CREATE等权限为只读操作创建单独用户限制可访问的IP范围# 生产环境连接配置示例 DATABASES { default: { ENGINE: django.db.backends.mysql, USER: app_readonly, # 只读用户 PASSWORD: complex_password, HOST: db.internal, PORT: 3306, OPTIONS: { init_command: SET sql_modeSTRICT_TRANS_TABLES, } } }4.2 使用Web应用防火墙(WAF)配置WAF规则拦截常见注入特征检测SQL关键字SELECT, UNION等检测特殊字符, --, /*等检测异常长的参数# 伪代码示例 - 简单的中间件检查 class SQLInjectionMiddleware: BLACKLIST [, --, /*, xp_, exec] def process_request(self, request): for param in request.GET: if any(bad in request.GET[param] for bad in self.BLACKLIST): return HttpResponse(可疑输入被拦截, status400)4.3 定期安全扫描使用工具自动化检测sqlmap专业但需授权使用OWASP ZAPBanditPython代码静态分析# 使用bandit扫描Python代码 bandit -r myproject/ -lll5. 实战修复易受攻击的代码让我们修复一个真实案例 - 博客系统的搜索功能原始危险代码app.route(/search) def search(): keyword request.args.get(q, ) query fSELECT * FROM posts WHERE content LIKE %{keyword}% posts db.execute(query).fetchall() return render_template(results.html, postsposts)修复步骤改用参数化查询app.route(/search) def search(): keyword request.args.get(q, ) keyword f%{keyword}% posts db.execute( SELECT * FROM posts WHERE content LIKE ?, (keyword,) ).fetchall() return render_template(results.html, postsposts)添加输入长度限制MAX_SEARCH_LENGTH 100 app.route(/search) def search(): keyword request.args.get(q, ) if len(keyword) MAX_SEARCH_LENGTH: abort(400, 搜索关键词过长) # 其余代码...记录可疑请求import re SQL_KEYWORDS re.compile(r\b(UNION|SELECT|INSERT|DELETE|UPDATE|DROP)\b, re.I) app.before_request def log_suspicious(): if SQL_KEYWORDS.search(request.query_string.decode()): app.logger.warning(f可疑SQL关键词: {request.remote_addr} - {request.url})6. 常见误区与注意事项6.1 转义不是万能的许多开发者误以为简单的字符串转义就足够# 不安全的转义尝试 def escape_sql(value): return value.replace(, ) query fSELECT * FROM users WHERE name{escape_sql(name)}这种方法不同数据库转义规则不同无法防御数字型注入如id1 OR 11可能被嵌套转义绕过6.2 存储过程也有风险错误认为存储过程绝对安全# 不安全的存储过程调用 cursor.callproc(get_user, (f{username},))正确做法是参数化调用cursor.callproc(get_user, (username,))6.3 ORM不等于免疫即使使用ORM以下情况仍可能出问题使用extra()或raw()方法批量操作时拼接字符串动态生成Q对象时未过滤输入# Django中危险的extra用法 Book.objects.extra(where[ftitle{user_input}])6.4 防御深度建议代码层面所有数据库操作使用参数化查询动态表名/列名使用白名单验证避免直接拼接SQL片段架构层面实现分层防御WAF代码防护生产环境禁用错误详情展示定期进行安全审计运维层面数据库备份隔离存储实施最小权限原则监控异常SQL查询我在实际项目中发现综合采用这些措施后SQL注入漏洞可以减少99%以上。特别是在金融和电商领域这种深度防御策略至关重要。