大模型安全（三十二）：核心技术栈之Requests/httpx高级应用与并发实战

摘要：在Python安全工具开发中，requests库是绝对的标准，但它在异步IO和HTTP/2支持上的缺失，使其在面对大规模并发扫描（如百万级Prompt测试）时显得力不从心。本文将引入新一代全能HTTP客户端——httpx。我们将对比两者在同步与异步模式下的性能差异，深入探讨HTTP/2对绕过WAF和提升传输效率的意义。通过实战代码，我们将构建一个基于asyncio + httpx的高并发扫描核心，实现单进程下每秒发送数千个请求的能力，并展示如何通过连接池复用和客户端指纹伪造来提升扫描的隐蔽性与稳定性。

关键词：httpx, requests, 异步IO, asyncio, HTTP/2, 高并发扫描, 连接池, Python网络编程

正文

1. 时代的眼泪：为什么要从 Requests 迁移到 HTTPX？

requests 是 Python 的国宝级库，但在大模型安全（特别是红队测试）场景下，它有两个致命短板：

不支持异步（Async）：它是阻塞的。要实现并发，必须上多线程/多进程，资源消耗大，且受GIL限制。

不支持 HTTP/2：现代网关（如gRPC接口）和WAF越来越依赖HTTP/2特性。只用HTTP/1.1不仅效率低，还容易被指纹识别为“旧时代的脚本”。

httpx 的优势：

API兼容：99% 兼容 requests 的 API，迁移成本极低。

原生异步：完美配合 asyncio，单核即可处理万级并发。

HTTP/2支持：原生支持，且无需复杂配置。

2. 高并发的核心：Event Loop 与 Connection Pooling

在扫描大模型API时，我们通常需要发送成千上万个Prompt。

错误做法：for 循环里 await client.post()。这依然是串行的，不仅慢，而且每次都建立新TCP连接（握手开销巨大）。

正确做法：使用 asyncio.gather + httpx.AsyncClient 的连接池。

连接池（Connection Pooling）：保持与服务器的TCP连接常开，复用连接发送后续请求。对于HTTPS来说，这省去了昂贵的TLS握手时间。

3. Python实战：构建万级并发扫描核心

我们将编写一个脚本，对比 requests（多线程）和 httpx（异步）在处理大量请求时的性能差异，并展示如何优雅地处理速率限制（Rate Limiting）。

代码实现：

Python



import asyncio
import httpx
import requests
import time
import threading
from concurrent.futures import ThreadPoolExecutor
 
# 模拟目标：一个响应稍微有点慢的 API
TARGET_URL = "https://httpbin.org/delay/0.5" # 延迟0.5秒
TOTAL_REQUESTS = 50
 
# --- 方案 A: Requests + 多线程 (传统派) ---
 
def requests_worker(session, url):
    try:
        resp = session.get(url)
        return resp.status_code
    except Exception as e:
        return str(e)
 
def run_requests_multithread():
    print(f"[*] [Requests] 启动多线程扫描 ({TOTAL_REQUESTS} 请求)...")
    start = time.time()
    
    # 使用 Session 开启连接池
    with requests.Session() as session:
        # 调整连接池大小，防止阻塞
        adapter = requests.adapters.HTTPAdapter(pool_connections=50, pool_maxsize=50)
        session.mount("https://", adapter)
        
        with ThreadPoolExecutor(max_workers=20) as executor:
            futures = [executor.submit(requests_worker, session, TARGET_URL) for _ in range(TOTAL_REQUESTS)]
            results = [f.result() for f in futures]
            
    duration = time.time() - start
    print(f"[+] [Requests] 完成。耗时: {duration:.2f}s | TPS: {TOTAL_REQUESTS/duration:.2f}")
 
# --- 方案 B: HTTPX + AsyncIO (现代派) ---
 
async def httpx_worker(client, url, semaphore):
    # 使用信号量 (Semaphore) 限制最大并发数，防止撑爆本地端口或被封IP
    async with semaphore:
        try:
            resp = await client.get(url)
            return resp.status_code
        except Exception as e:
            return str(e)
 
async def run_httpx_async():
    print(f"[*] [HTTPX] 启动异步扫描 ({TOTAL_REQUESTS} 请求)...")
    start = time.time()
    
    # 限制最大并发数为 50
    limits = httpx.Limits(max_keepalive_connections=20, max_connections=50)
    semaphore = asyncio.Semaphore(50)
    
    # 开启 http2=True
    async with httpx.AsyncClient(limits=limits, http2=True, verify=False) as client:
        tasks = [httpx_worker(client, TARGET_URL, semaphore) for _ in range(TOTAL_REQUESTS)]
        results = await asyncio.gather(*tasks)
        
    duration = time.time() - start
    print(f"[+] [HTTPX] 完成。耗时: {duration:.2f}s | TPS: {TOTAL_REQUESTS/duration:.2f}")
 
# --- 性能对比 ---
 
if __name__ == "__main__":
    # 1. 跑 Requests
    run_requests_multithread()
    
    print("-" * 40)
    
    # 2. 跑 HTTPX
    asyncio.run(run_httpx_async())

代码运行结果解析（预估）：

Requests：虽然开了20个线程，但由于线程切换开销和GIL锁，加上0.5s的硬等待，耗时会较长（约 1.5s – 2s）。

HTTPX：所有请求几乎同时发出，同时等待。耗时将无限接近于目标服务器的单次延迟（约 0.55s – 0.7s）。

性能提升：在网络IO密集型场景下，异步通常比多线程快 3-10倍，且内存占用极低。

4. 进阶技巧：HTTP/2 与指纹伪造 (Syllabus 8.1.1)

在攻击防护森严的LLM API（如OpenAI）时，默认的Python User-Agent会被瞬间拦截。除了修改Header，我们还需要关注TLS指纹。

httpx 结合 HTTP/2 可以规避部分基于协议版本的检测。

Python



async def advanced_stealth_request(url):
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)...",
        "Accept-Language": "en-US,en;q=0.9"
    }
    # 启用 HTTP/2，这会让流量看起来更像现代浏览器
    async with httpx.AsyncClient(http2=True, headers=headers) as client:
        resp = await client.get(url)
        print(f"Protocol used: {resp.http_version}") # 输出 HTTP/2