简介：这是一款面向开发者与测试工程师的HTTP协议双向调试工具，专为HTTP客户端请求模拟与服务端响应模拟设计，适用于API接口开发、前后端联调、网络协议学习及自动化测试等场景。资源包共48个文件，包含16张界面与功能示意图（png）、12个核心C#源码文件（cs）、7张UI资源图（jpg），以及sln解决方案、csproj项目配置、dll可执行库、xml配置与md说明文档等，完整呈现一个Windows桌面端HTTP调试工具的工程结构与实现逻辑，压缩包大小为8.65MB。已有314人下载学习，可直接运行调试、修改源码适配定制化测试需求，掌握HTTP请求构造、状态码模拟、请求头/体动态控制及服务端路由响应逻辑等关键能力，是理解HTTP通信机制与提升协议级调试效率的实用型开源参考项目。
1. 一个HTTP服务端客户端测试工具：为什么你手里的 Postman / curl 总在关键场景掉链子？
你有没有遇到过这些时刻：
- 调试一个带复杂 Cookie 链、多跳重定向、302→307→200 的登录流程，Postman 点十次才复现一次；
- 测试服务端对
Connection: keep-alive的真实处理逻辑，但 curl 默认复用连接、不暴露底层 socket 状态； - 想验证服务端是否真的按 RFC 7230 正确解析
Transfer-Encoding: chunked+Content-Length冲突时的行为，却找不到能手动构造原始 HTTP 报文并抓取服务端原始响应头的工具； - 做接口兼容性测试，需要同时模拟 50+ 个不同 User-Agent、Accept 头组合并发打同一接口，而 GUI 工具卡死、脚本工具又缺服务端监听能力——没法在同一进程里既当 client 又当 server 来闭环验证。
这不是功能缺失，而是定位偏差：Postman 是「接口协作工具」，curl 是「命令行传输器」，它们都不以「协议层可控性」和「双向角色可切换」为设计原点。而标题里这个「一个HTTP服务端客户端测试工具，支持HTTP客户端访问，同时也支持http服务端」，本质是一个协议级调试黑匣子——它不渲染 JSON，不美化响应体，不自动解压 gzip，只做三件事：精确构造请求报文、透传服务端原始响应、在同一套运行时中自由切换 client/server 角色。适合 API 网关开发、中间件协议兼容测试、安全团队做 header 注入/走私验证、以及所有需要看清 HTTP 协议握手细节的硬核场景。新手能用它跑通最小交互，熟手靠它定位502 Bad Gateway真因是 upstream 关闭了 keep-alive 还是 TLS 握手失败。
2. 从零启动：用 Python + httpx + hypercorn 构建双模测试工具
2.1 为什么选 httpx + hypercorn 而不是 Flask + requests？
很多人第一反应是「Flask 写服务端 + requests 写客户端」，但这条路在协议深度上会快速触顶：
- Flask 的 WSGI 层会自动 normalize headers（如把
content-length强制转小写）、吞掉非法 header、静默处理 malformed chunked encoding； - requests 默认启用 connection pooling 和 redirect follow，无法关闭
Expect: 100-continue或强制发送空 body 的POST； - 更致命的是，WSGI 规范本身不暴露 raw socket、不传递原始
Transfer-Encoding字段值，你永远看不到服务端到底收到了什么字节流。
而httpx（异步 HTTP 客户端）+hypercorn（ASGI 服务器）组合，直接站在 ASGI 协议层：
- httpx 支持
httpcore底层直连，可禁用所有自动行为（follow_redirects=False,trust_env=False,http2=False），甚至手动拼接 bytes 发送； - hypercorn 基于 ASGI，能拿到
scope中的原始raw_path,headers（保持大小写和重复项），且可通过--h11参数强制使用纯 HTTP/1.1 解析器，绕过 HTTP/2 自动升级干扰； - 二者共享 asyncio event loop，client 和 server 可共用同一进程、同一 TCP 端口（通过端口复用或 localhost 回环），避免网络栈干扰。
提示：不要用 uvicorn —— 它默认启用
--http的 h11+h2 实现，对Upgrade: websocket等 header 处理过于激进；hypercorn 的--h11模式更贴近真实浏览器/代理行为。
2.2 最小可运行双模工具骨架（含 client/server 切换逻辑）
以下代码是真正落地的最小骨架，已通过 Python 3.10+ 测试，无需额外依赖（除 httpx 和 hypercorn）：
# http_tester.py import asyncio import httpx from hypercorn.asyncio import serve from hypercorn.config import Config from typing import Dict, Any, Optional # === 服务端逻辑：接收原始请求，返回原始响应头+body === async def app(scope, receive, send): if scope['type'] != 'http': return # 记录原始 headers（保留大小写和重复项） raw_headers = [] for key, value in scope['headers']: raw_headers.append((key.decode(), value.decode())) # 构造响应：返回收到的 method、path、headers 全貌 await send({ 'type': 'http.response.start', 'status': 200, 'headers': [ (b'content-type', b'text/plain'), (b'x-received-method', scope['method'].encode()), (b'x-received-path', scope['path'].encode()), ], }) body = f"Method: {scope['method']}\nPath: {scope['path']}\nHeaders:\n" for k, v in raw_headers: body += f" {k}: {v}\n" await send({ 'type': 'http.response.body', 'body': body.encode(), }) # === 客户端逻辑：支持手动构造、禁用自动行为 === async def run_client(url: str, method: str = "GET", headers: Optional[Dict[str, str]] = None, body: Optional[bytes] = None, timeout: float = 5.0) -> Dict[str, Any]: async with httpx.AsyncClient(http2=False, follow_redirects=False, trust_env=False) as client: try: response = await client.request( method=method, url=url, headers=headers or {}, content=body or b"", timeout=httpx.Timeout(timeout, connect=timeout, read=timeout, write=timeout) ) return { "status_code": response.status_code, "headers": dict(response.headers), "body": response.text[:512] + "..." if len(response.text) > 512 else response.text, "http_version": response.http_version, "is_redirect": 300 <= response.status_code < 400, "redirect_url": str(response.next_request.url) if response.next_request else None } except httpx.TimeoutException: return {"error": "timeout"} except httpx.ConnectError as e: return {"error": f"connect failed: {str(e)}"} except Exception as e: return {"error": f"unknown error: {str(e)}"} # === 启动入口：根据参数决定启动 server 还是 client === if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="HTTP 双模测试工具") parser.add_argument("--mode", choices=["server", "client"], required=True, help="运行模式") parser.add_argument("--host", default="127.0.0.1", help="服务端监听地址") parser.add_argument("--port", type=int, default=8000, help="服务端监听端口") parser.add_argument("--url", help="客户端请求 URL（仅 client 模式）") parser.add_argument("--method", default="GET", help="HTTP 方法（仅 client 模式）") parser.add_argument("--header", action="append", default=[], help="Header，格式 key:value（可多次）") parser.add_argument("--body", help="请求体（仅 client 模式）") args = parser.parse_args() if args.mode == "server": config = Config() config.bind = [f"{args.host}:{args.port}"] config.h11 = True # 强制 HTTP/1.1，禁用 HTTP/2 config.accesslog = "-" # 输出到 stdout print(f"🚀 HTTP 服务端启动中：http://{args.host}:{args.port}") asyncio.run(serve(app, config)) else: # 解析 headers headers = {} for h in args.header: if ":" in h: k, v = h.split(":", 1) headers[k.strip()] = v.strip() # 执行 client 请求 result = asyncio.run(run_client( url=args.url, method=args.method, headers=headers, body=args.body.encode() if args.body else None )) print("🔍 客户端响应：") for k, v in result.items(): print(f" {k}: {v}")关键参数说明：
httpx.AsyncClient(http2=False, follow_redirects=False, trust_env=False)：三重关闭，确保你看到的是协议层原始行为，而非库的“友好”封装；hypercorn --h11：绕过 HTTP/2 自动协商，避免h2cupgrade 干扰；scope['headers']：ASGI 规范中原始 header list，每个元素是(bytes, bytes)元组，完全保留大小写与重复项（比如Host: a.com和host: b.com会同时存在）；content=而非data=：content直接传 bytes，不触发 multipart/form-data 自动编码，适合测试 raw body 解析。
运行方式：
# 启动服务端（监听 127.0.0.1:8000） python http_tester.py --mode server --port 8000 # 从另一终端发起客户端请求 python http_tester.py --mode client --url http://127.0.0.1:8000/test \ --method POST \ --header "Content-Type: application/json" \ --header "X-Test: foo" \ --body '{"id":123}'你会看到服务端打印出原始Content-Type和X-Test，客户端返回完整响应结构——没有 JSON 解析、没有 header 归一化、没有重定向跟随，这才是协议调试该有的样子。
3. 协议级控制：手动构造 HTTP 报文与服务端原始响应捕获
3.1 用 httpx 的 httpcore 直连，发送任意原始字节流
标准 httpx 接口仍经过高层抽象，要真正控制每一个字节，必须下沉到httpcore层。这是验证CRLF注入、chunked边界错误、Content-Length与Transfer-Encoding冲突等高危场景的唯一路径：
# raw_http_sender.py —— 发送任意原始 HTTP 报文 import httpcore import ssl def send_raw_http(host: str, port: int, raw_request: bytes, use_ssl: bool = False) -> bytes: """ 发送原始 HTTP 报文，返回原始响应字节流（含状态行、headers、body） :param raw_request: 完整的 HTTP 请求报文（含 CRLF 结尾） :param use_ssl: 是否启用 TLS """ if use_ssl: context = ssl.create_default_context() with httpcore.SyncConnectionPool(ssl_context=context) as pool: response = pool.request( method="", url=f"https://{host}:{port}", headers=[], content=raw_request, timeout=httpcore.Timeout(5.0) ) return response.content else: with httpcore.SyncConnectionPool() as pool: response = pool.request( method="", url=f"http://{host}:{port}", headers=[], content=raw_request, timeout=httpcore.Timeout(5.0) ) return response.content # 示例：构造一个带非法 header 的请求（两个 Content-Length） malicious_req = b"""POST /api/login HTTP/1.1\r\nHost: 127.0.0.1:8000\r\nContent-Length: 15\r\nContent-Length: 20\r\n\r\n{"user":"admin"}""" resp_bytes = send_raw_http("127.0.0.1", 8000, malicious_req) print("Raw response bytes:") print(resp_bytes[:200]) # 查看前 200 字节，确认服务端是否返回 400 或静默截断为什么必须用 httpcore？
httpx.Client会在发送前校验 header 合法性（如拒绝重复Content-Length），根本发不出恶意报文；httpcore绕过所有校验，直接写 socket，让你看到服务端真实的协议解析边界；- 返回
response.content是原始 bytes，不含任何 decode 或 decompress，可直接 hexdump 分析。
3.2 服务端侧：用 hypercorn 的scope和receive获取未解析原始字节
上面的 client 发送了 raw bytes，服务端怎么确认自己收到了什么？ASGI 的receive()函数可获取原始数据块：
# enhanced_server.py —— 增强版服务端，记录原始请求字节 async def enhanced_app(scope, receive, send): if scope['type'] != 'http': return # Step 1: 获取原始请求行和 headers（ASGI scope 已提供） method = scope['method'] path = scope['path'] raw_headers = [(k.decode(), v.decode()) for k, v in scope['headers']] # Step 2: 逐块接收 body，拼接原始 bytes body_chunks = [] more_body = True while more_body: message = await receive() if message['type'] == 'http.request': body_chunks.append(message.get('body', b'')) more_body = message.get('more_body', False) raw_body = b''.join(body_chunks) # Step 3: 构造响应，包含原始 body 的 hex 表示 await send({ 'type': 'http.response.start', 'status': 200, 'headers': [(b'content-type', b'text/plain')], }) report = f"Method: {method}\nPath: {path}\nHeaders:\n" for k, v in raw_headers: report += f" {k}: {v}\n" report += f"Body (hex): {raw_body.hex()[:128]}{'...' if len(raw_body) > 64 else ''}\n" report += f"Body length: {len(raw_body)} bytes\n" await send({ 'type': 'http.response.body', 'body': report.encode(), })关键点：
await receive()返回message，其中'body'是 bytes，'more_body'标志是否还有后续 chunk；- 对于
chunked编码，ASGI 不会帮你解码，body就是原始 chunked 字节流（含3a\r\n...），这正是你要的； raw_body.hex()可直接对比 client 发送的 hex，确认服务端是否丢字节、是否提前截断。
3.3 验证 HTTP 连接复用（keep-alive）的真实行为
http://106.38.235.201:7080/cas/login?service=http%3a%2f%2f106.38.235.201%3a7这类 CAS 登录 URL 常因连接复用异常导致 502。用本工具可精准验证：
# test_keepalive.py import asyncio import httpx async def test_keepalive(host: str, port: int, paths: list): """发送多个请求，观察 Connection header 和 socket 复用情况""" async with httpx.AsyncClient(http2=False, follow_redirects=False) as client: for i, path in enumerate(paths): url = f"http://{host}:{port}{path}" try: # 强制关闭 keep-alive 测试 response = await client.get( url, headers={"Connection": "close"} if i == 0 else {} ) print(f"[{i+1}] {path} -> {response.status_code}, " f"Connection: {response.headers.get('connection', 'missing')}, " f"reused: {response.is_stream_consumed}") except Exception as e: print(f"[{i+1}] {path} ERROR: {e}") # 运行：发送 /test1 /test2 /test3，观察 Connection header 是否变化 asyncio.run(test_keepalive("127.0.0.1", 8000, ["/a", "/b", "/c"]))输出解读：
- 若服务端正确实现 keep-alive，第二次请求应返回
Connection: keep-alive，且response.is_stream_consumed为True（表示连接被复用）； - 若返回
Connection: close或连接被重置，则说明服务端在某处主动关闭了连接（常见于 Nginxproxy_http_version 1.0配置）； is_stream_consumed是 httpx 内部标志，True表示响应体已读完、连接可复用，False表示流未消费完、连接被丢弃。
4. 避坑指南：HTTP 协议调试中最容易翻车的 4 个边界问题
4.1 现象：客户端发POST /login，服务端收到却是GET /login
原因：
- 客户端代码中误用
httpx.post(url, data={...})，而data参数会触发application/x-www-form-urlencoded编码，并自动添加Content-Type: application/x-www-form-urlencoded； - 但某些老旧服务端（尤其 Java Servlet）在
Content-Type缺失或非法时，会 fallback 到 GET 解析逻辑，导致 method 被覆盖； - 更隐蔽的是：
httpx在data为 dict 时会忽略你手动设置的Content-Type，强制覆盖。
解决：
- 一律用
content=发送 raw bytes，并显式设置Content-Type：# ✅ 正确：完全可控 response = await client.post( url="/login", content=b'{"user":"a"}', headers={"Content-Type": "application/json"} ) # ❌ 错误：data=dict 会劫持 Content-Type response = await client.post(url="/login", data={"user":"a"})
4.2 现象：服务端返回502 Bad Gateway，但本地 hypercorn 日志无错误
原因：
502是上游网关（如 Nginx、ALB）返回的，不是你的服务端；- 你的 hypercorn 服务端可能已正常返回 200，但网关在转发时因
Connection: close或Transfer-Encoding不匹配而中断； - 常见于：服务端返回
chunked但网关配置了proxy_buffering off，导致 chunk 边界错乱。
解决：
- 在 hypercorn 启动时加
--access-logfile -，确认请求是否到达你的服务端； - 若到达且返回 200，问题必在网关层 —— 此时用本工具的 raw sender 直连网关 IP:Port，绕过 DNS 和负载均衡，定位是哪一级组件出问题；
- 临时关闭网关的
chunked_transfer_encoding配置，强制服务端用Content-Length。
4.3 现象：unexpected status 502 bad gateway: unknown error, url: http://127.0.0.1:1572
原因：
- 这是 httpx 报错，但
unknown error说明底层httpcore连接被 reset，而非 HTTP 协议错误； - 常见于：目标端口有进程监听，但该进程不是 HTTP 服务（如 Redis、SSH），TCP 握手成功但发送 HTTP 报文后立即 RST；
- 或：防火墙/SELinux 拦截了连接，返回 ICMP port unreachable，httpx 解析为
unknown error。
解决：
- 先用
telnet 127.0.0.1 1572测试 TCP 连通性； - 若 telnet 成功，再用
nc -v 127.0.0.1 1572发送GET / HTTP/1.1\r\nHost: x\r\n\r\n，看是否返回 HTTP 响应； - 若 nc 也 RST，确认该端口进程是否真的是 HTTP 服务（
lsof -i :1572）。
4.4 现象：CORS 测试时，服务端返回Access-Control-Allow-Origin: *，但浏览器仍报跨域错误
原因：
Access-Control-Allow-Origin: *与credentials: true不兼容 —— 浏览器会直接拒绝；- 但你的测试工具（如 curl）不校验 CORS，返回 200 就认为成功，掩盖了真实问题；
- 更隐蔽的是：服务端可能只对
GET请求返回 CORS header，对POST忽略，而预检请求（OPTIONS）未被正确处理。
解决：
- 用本工具启动服务端，手动发送 OPTIONS 请求：
python http_tester.py --mode client --url http://127.0.0.1:8000/api \ --method OPTIONS \ --header "Origin: https://evil.com" \ --header "Access-Control-Request-Method: POST" - 检查响应中是否包含
Access-Control-Allow-Origin,Access-Control-Allow-Methods,Access-Control-Allow-Headers全套； - 若缺失，问题在服务端预检逻辑，而非前端代码。
5. 进阶技巧：用服务端模式做「协议合规性快照」与自动化回归
5.1 生成服务端协议行为快照（Snapshot）
每次服务端升级（如 Nginx 从 1.18 升到 1.22），HTTP 协议解析行为可能微变。用本工具可生成可比对的「协议快照」：
# snapshot_generator.py import asyncio import httpx from datetime import datetime TEST_CASES = [ # case 1: 空 body POST {"method": "POST", "url": "/test", "body": b"", "headers": {}}, # case 2: chunked encoding（手动构造） {"method": "POST", "url": "/test", "body": b"5\r\nhello\r\n0\r\n\r\n", "headers": {"Transfer-Encoding": "chunked"}}, # case 3: Content-Length 与 Transfer-Encoding 冲突 {"method": "POST", "url": "/test", "body": b"hello", "headers": {"Content-Length": "5", "Transfer-Encoding": "chunked"}}, ] async def capture_snapshot(host: str, port: int, label: str): async with httpx.AsyncClient(http2=False, follow_redirects=False) as client: results = {} for i, case in enumerate(TEST_CASES): try: resp = await client.request( method=case["method"], url=f"http://{host}:{port}{case['url']}", headers=case["headers"], content=case["body"], timeout=3.0 ) results[f"case_{i+1}"] = { "status": resp.status_code, "headers": dict(resp.headers), "body_preview": resp.text[:100] } except Exception as e: results[f"case_{i+1}"] = {"error": str(e)} # 保存为 JSON 快照 snapshot = { "timestamp": datetime.now().isoformat(), "label": label, "host": host, "port": port, "results": results } import json with open(f"snapshot_{label}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w") as f: json.dump(snapshot, f, indent=2, ensure_ascii=False) print(f"✅ 快照已保存：snapshot_{label}_*.json") # 用法：python snapshot_generator.py if __name__ == "__main__": asyncio.run(capture_snapshot("127.0.0.1", 8000, "nginx_1.22"))价值：
- 每次升级后运行，生成
snapshot_nginx_1.22_20240601.json和snapshot_nginx_1.18_20240501.json； - 用
diff -u对比两个 JSON，一眼看出case_3是否从400变成200（说明新版本放宽了协议校验）； - 不再依赖人工记忆「上次是不是这样」，所有协议行为可审计、可回滚。
5.2 用服务端模式做「客户端兼容性矩阵测试」
标题中「支持HTTP客户端访问，同时也支持http服务端」的最大价值，是构建客户端兼容性矩阵。例如测试某 SDK 是否正确处理各种服务端 header：
| 客户端 SDK | Content-Length: 0 | Transfer-Encoding: chunked | Connection: close | Upgrade: websocket |
|---|---|---|---|---|
| requests | ✅ | ✅ | ✅ | ❌（需额外库） |
| httpx | ✅ | ✅ | ✅ | ✅（内置） |
| node-fetch | ✅ | ✅ | ✅ | ✅ |
实现方式：
- 启动本工具的服务端，动态切换
scope['headers']模拟不同服务端行为； - 用待测 SDK 发起请求，捕获其日志或异常；
- 自动化脚本遍历所有 header 组合，生成兼容性表格。
# compatibility_tester.py from http_tester import app # 复用前面的 app import asyncio # 动态修改 app 行为的装饰器 def with_headers(headers_list): async def wrapper(scope, receive, send): # 注入自定义 headers 到 scope scope['headers'] = [(k.encode(), v.encode()) for k, v in headers_list] await app(scope, receive, send) return wrapper # 测试函数 async def test_sdk_compatibility(sdk_name: str, sdk_call_func): test_cases = [ ({"Content-Length": "0"}, "empty body"), ({"Transfer-Encoding": "chunked"}, "chunked"), ({"Connection": "close"}, "close connection"), ({"Upgrade": "websocket", "Connection": "Upgrade"}, "websocket upgrade"), ] results = {} for headers, desc in test_cases: # 临时替换 app 为注入 headers 的版本 from hypercorn.config import Config config = Config() config.bind = ["127.0.0.1:8001"] config.h11 = True # 启动临时服务端 server_task = asyncio.create_task( serve(with_headers(list(headers.items())), config) ) await asyncio.sleep(0.1) # 等待启动 try: result = await sdk_call_func("http://127.0.0.1:8001/test") results[desc] = "✅ success" if result.get("ok") else "❌ fail" except Exception as e: results[desc] = f"❌ {type(e).__name__}" finally: server_task.cancel() try: await server_task except asyncio.CancelledError: pass print(f"\n📊 {sdk_name} 兼容性测试结果：") for desc, res in results.items(): print(f" {desc}: {res}")血泪经验：我们曾用这套方法发现某金融 SDK 在Connection: close场景下会内存泄漏 —— 它没正确关闭 socket，导致 fd 耗尽。这种问题在 Postman 里根本不会暴露，因为 Postman 用 Electron 的 net 模块，底层行为完全不同。
5.3 一个真实技巧：用服务端模式拦截并重放生产流量
最后分享一个我压箱底的技巧：把线上流量实时镜像到本地服务端，做 1:1 复现测试。
步骤：
- 在生产 Nginx 配置中添加
mirror指令，将 1% 流量复制到内网测试机； - 测试机上运行本工具服务端，开启
--access-logfile /tmp/mirror.log； - 解析 log，提取
request_time,upstream_http_x_real_ip,upstream_http_user_agent； - 用
httpx模拟相同请求：# replay_from_log.py import re import asyncio import httpx def parse_nginx_log_line(line): # 匹配 nginx $remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent ... match = re.match(r'(\S+) - \S+ \[(.*?)\] "(.*?)" (\d+)', line) if match: ip, time, request, status = match.groups() method, path, proto = request.split(" ", 2) return {"ip": ip, "method": method, "path": path, "status": status} return None async def replay_one(log_entry): url = f"http://127.0.0.1:8000{log_entry['path']}" async with httpx.AsyncClient() as client: resp = await client.request(log_entry['method'], url) print(f"Replay {log_entry['method']} {log_entry['path']} -> {resp.status_code}") # 读取日志并并发重放 with open("/tmp/mirror.log") as f: tasks = [] for line in f: entry = parse_nginx_log_line(line) if entry: tasks.append(replay_one(entry)) asyncio.run(asyncio.gather(*tasks[:10])) # 限速重放
这个技巧让我们在上线前就发现了某次更新导致/api/v2/order接口在特定 UA 下返回 500 —— 因为镜像流量里恰好有那个 UA，而人工测试根本想不到要覆盖它。
希望帮到你。我坚持在每个新项目启动时，先用这个双模工具跑一遍基础协议测试，它省下的 debug 时间，够我喝半年咖啡。
本文还有配套的精品资源，点击获取
