SQL注入攻击的核心原理是攻击者将恶意SQL代码片段插入到应用的输入参数中，当应用未对输入做严格过滤就直接拼接SQL语句执行时，恶意代码会被数据库解析执行，从而导致数据泄露、篡改甚至删除等严重后果。要防范这类攻击，首先需要掌握SQL注入的检测方法，再配套实时告警和拦截机制。
SQL注入行为的常见检测方式
基于请求参数的特征检测
大多数SQL注入攻击会通过URL参数、表单提交、HTTP请求头中的字段传入恶意内容，这些恶意内容通常包含SQL语法特征，比如单引号、注释符、UNION、SELECT、OR等关键字。我们可以通过正则匹配的方式识别这些特征，以下是Python实现的简单检测逻辑：
import re
# 定义SQL注入常见特征正则
SQL_INJECTION_PATTERNS = [
r"(%27)|(')|(--)|(%23)|(#)", # 单引号、注释符
r"((%3D)|(=))[^n]*((%27)|(')|(--)|(%3B)|(;))", # 等号后接恶意字符
r"w*((%27)|('))((%6F)|o|(%4F))((%72)|r|(%52))", # 匹配 or 关键字
r"((%27)|('))union", # 匹配 union 关键字
r"select.+from", # 匹配 select from 结构
]
def check_sql_injection(input_str):
"""检测输入字符串是否包含SQL注入特征"""
for pattern in SQL_INJECTION_PATTERNS:
if re.search(pattern, input_str, re.IGNORECASE):
return True
return False
# 测试示例
test_input1 = "id=1' union select 1,2,3 from users--"
test_input2 = "username=test"
print(check_sql_injection(test_input1)) # 输出 True
print(check_sql_injection(test_input2)) # 输出 False
基于数据库日志的检测
如果应用已经上线，还可以通过分析数据库的查询日志来识别异常的SQL语句。比如突然出现大量包含UNION查询、多表关联查询、没有WHERE条件的全表查询，或者执行了DROP、DELETE等高危操作的SQL，都可能是SQL注入攻击的结果。可以定期读取数据库日志，对异常SQL做标记和统计。
建立实时告警机制
检测到SQL注入行为后，需要第一时间触发告警，让运维和安全人员及时处理。我们可以结合消息队列和告警平台实现实时通知，以下是基于Python和Redis消息队列的告警逻辑示例：
import redis
import json
import time
# 连接Redis
redis_client = redis.Redis(host='127.0.0.1', port=6379, db=0)
def send_alert(attack_info):
"""发送SQL注入告警信息到消息队列"""
alert_data = {
"alert_type": "SQL_INJECTION",
"attack_time": time.strftime("%Y-%m-%d %H:%M:%S"),
"attack_ip": attack_info.get("ip"),
"attack_param": attack_info.get("param"),
"request_url": attack_info.get("url")
}
# 将告警信息推送到Redis队列
redis_client.lpush("security_alert_queue", json.dumps(alert_data))
print("SQL注入告警已发送")
# 模拟检测到攻击后发送告警
attack_info = {
"ip": "192.168.0.100",
"param": "id=1' union select 1,2,3 from users--",
"url": "http://ipipp.com/api/user"
}
send_alert(attack_info)
告警信息进入队列后，可以单独启动一个消费服务，将告警内容推送到企业微信、钉钉、邮件等通知渠道，确保相关人员能及时收到风险提醒。
实现风险实时拦截
除了告警，还需要在检测到SQL注入时直接拦截请求，避免攻击到达数据库。如果是Web应用，可以在请求入口处添加拦截中间件，以下是Flask框架的中间件示例：
from flask import Flask, request, abort
import re
app = Flask(__name__)
# SQL注入检测正则（和之前的检测逻辑一致）
SQL_INJECTION_PATTERNS = [
r"(%27)|(')|(--)|(%23)|(#)",
r"((%3D)|(=))[^n]*((%27)|(')|(--)|(%3B)|(;))",
r"w*((%27)|('))((%6F)|o|(%4F))((%72)|r|(%52))",
r"((%27)|('))union",
r"select.+from",
]
def is_sql_injection(input_str):
for pattern in SQL_INJECTION_PATTERNS:
if re.search(pattern, input_str, re.IGNORECASE):
return True
return False
@app.before_request
def sql_injection_intercept():
"""请求前拦截SQL注入"""
# 检查所有请求参数
for key, value in request.args.items():
if is_sql_injection(value):
abort(403, description="检测到非法请求，已拦截")
# 检查表单参数
if request.form:
for key, value in request.form.items():
if is_sql_injection(value):
abort(403, description="检测到非法请求，已拦截")
# 检查JSON请求体
if request.is_json:
json_data = request.get_json()
for key, value in json_data.items():
if isinstance(value, str) and is_sql_injection(value):
abort(403, description="检测到非法请求，已拦截")
@app.route("/api/user")
def get_user():
user_id = request.args.get("id")
return f"用户ID: {user_id}"
if __name__ == "__main__":
app.run(host="127.0.0.1", port=5000)
补充防护建议
除了上述检测和拦截机制，还可以结合以下方式提升防护效果：
- 使用参数化查询或者ORM框架，避免直接拼接SQL语句，从根源上杜绝SQL注入可能
- 对用户输入做严格的类型校验，比如数字类型的参数只允许输入数字，过滤掉其他非法字符
- 配置WAF（Web应用防火墙），添加通用的SQL注入防护规则，作为第一层防护屏障
- 定期做安全渗透测试，模拟SQL注入攻击，验证防护机制的有效性
通过以上检测、告警、拦截的组合方案，可以有效应对大部分SQL注入攻击，保障业务和数据的安全。