qing
bbea1ee9d8
- tests/blackbox/test_gateway.py 重写为 6 段 34 断言（输入校验/whoami+admin.verify/ captcha-image 代理/路径遍历矩阵/限流随机XFF隔离/CORS+脱敏回归），实测 32 passed/2 skipped - helpers.py 新增 random_xff/assert_no_internal_leak/db_execute - docs/security/gateway_SECURITY_AUDIT_20260809.md：A1-A8 增量复审（已全部修复） - docs/reports/libhttp_remote_addr_dev_doc.md：A4 根治方案（X-Remote-Addr 可信头注入） - docs/INDEX.md：security 29 篇/reports 12 篇同步
377 lines
13 KiB
Python
377 lines
13 KiB
Python
"""
|
||
micro_servers 黑盒测试 — 共享工具模块。
|
||
|
||
提供: HTTP 请求封装、断言工具、测试计数器、pymsg 协议客户端。
|
||
纯标准库实现，零外部依赖。
|
||
"""
|
||
|
||
import http.client
|
||
import json
|
||
import random
|
||
import socket
|
||
import struct
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
import urllib.parse
|
||
import ssl
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试计数器
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class Counter:
|
||
"""每个测试模块一个计数器实例，统计 passed/failed/skipped。"""
|
||
def __init__(self):
|
||
self.passed = 0
|
||
self.failed = 0
|
||
self.skipped = 0
|
||
self._failures = []
|
||
|
||
def add_pass(self):
|
||
self.passed += 1
|
||
|
||
def add_fail(self, name: str, reason: str = ""):
|
||
self.failed += 1
|
||
self._failures.append((name, reason))
|
||
|
||
def add_skip(self):
|
||
self.skipped += 1
|
||
|
||
@property
|
||
def total(self):
|
||
return self.passed + self.failed + self.skipped
|
||
|
||
def summary(self) -> str:
|
||
return f"{self.total} tests: {self.passed} passed, {self.failed} failed, {self.skipped} skipped"
|
||
|
||
def failures(self) -> list:
|
||
return self._failures
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 断言工具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def assert_eq(actual, expected, name: str, counter: Counter):
|
||
"""断言 actual == expected。"""
|
||
if actual == expected:
|
||
print(f" [PASS] {name}")
|
||
counter.add_pass()
|
||
return True
|
||
else:
|
||
print(f" [FAIL] {name}: expected {expected!r}, got {actual!r}")
|
||
counter.add_fail(name, f"expected {expected!r}, got {actual!r}")
|
||
return False
|
||
|
||
|
||
def assert_contains(haystack: str, needle: str, name: str, counter: Counter):
|
||
"""断言 needle in haystack（不区分大小写）。"""
|
||
if needle.lower() in haystack.lower():
|
||
print(f" [PASS] {name}")
|
||
counter.add_pass()
|
||
return True
|
||
else:
|
||
snippet = haystack[:200] if len(haystack) > 200 else haystack
|
||
print(f" [FAIL] {name}: '{needle}' not found in response")
|
||
counter.add_fail(name, f"'{needle}' not found in response")
|
||
return False
|
||
|
||
|
||
def assert_status(status: int, expected: int, name: str, counter: Counter):
|
||
"""断言 HTTP 状态码。"""
|
||
if status == expected:
|
||
print(f" [PASS] {name}")
|
||
counter.add_pass()
|
||
return True
|
||
else:
|
||
print(f" [FAIL] {name}: expected status {expected}, got {status}")
|
||
counter.add_fail(name, f"expected status {expected}, got {status}")
|
||
return False
|
||
|
||
|
||
def assert_true(condition: bool, name: str, counter: Counter, detail: str = ""):
|
||
"""断言条件为真。"""
|
||
if condition:
|
||
print(f" [PASS] {name}")
|
||
counter.add_pass()
|
||
return True
|
||
else:
|
||
msg = f" [FAIL] {name}"
|
||
if detail:
|
||
msg += f": {detail}"
|
||
print(msg)
|
||
counter.add_fail(name, detail)
|
||
return False
|
||
|
||
|
||
def assert_json_key(data: dict, key: str, name: str, counter: Counter):
|
||
"""断言 JSON 对象包含指定 key。"""
|
||
if key in data:
|
||
print(f" [PASS] {name}")
|
||
counter.add_pass()
|
||
return True
|
||
else:
|
||
print(f" [FAIL] {name}: key '{key}' not found in {json.dumps(data)[:200]}")
|
||
counter.add_fail(name, f"key '{key}' not found")
|
||
return False
|
||
|
||
|
||
def skip(name: str, reason: str, counter: Counter):
|
||
"""标记测试为跳过。"""
|
||
print(f" [SKIP] {name}: {reason}")
|
||
counter.add_skip()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HTTP 请求封装
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def http_request(method: str, url: str, *,
|
||
body: dict = None,
|
||
headers: dict = None,
|
||
timeout: float = 10.0,
|
||
raw_body: bytes = None,
|
||
raw_body_content_type: str = None):
|
||
"""发送 HTTP 请求，返回 (status, body_str, resp_headers)。
|
||
|
||
支持:
|
||
- JSON body (body 参数)
|
||
- 原始二进制 body (raw_body + raw_body_content_type)
|
||
- 任意自定义 headers
|
||
- 超时控制
|
||
"""
|
||
parsed = urllib.parse.urlparse(url)
|
||
data = None
|
||
hdrs = {}
|
||
|
||
if headers:
|
||
hdrs.update(headers)
|
||
|
||
if raw_body is not None:
|
||
data = raw_body
|
||
if raw_body_content_type:
|
||
hdrs["Content-Type"] = raw_body_content_type
|
||
elif body is not None:
|
||
data = json.dumps(body).encode("utf-8")
|
||
hdrs["Content-Type"] = "application/json"
|
||
|
||
ctx = ssl._create_unverified_context() if parsed.scheme == "https" else None
|
||
|
||
req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
|
||
resp_body = resp.read().decode("utf-8", errors="replace")
|
||
resp_headers = dict(resp.headers)
|
||
return resp.status, resp_body, resp_headers
|
||
except urllib.error.HTTPError as e:
|
||
resp_body = e.read().decode("utf-8", errors="replace") if e.fp else ""
|
||
resp_headers = dict(e.headers) if e.headers else {}
|
||
return e.code, resp_body, resp_headers
|
||
except Exception as e:
|
||
return 0, str(e), {}
|
||
|
||
|
||
def http_get(url: str, headers: dict = None, timeout: float = 10.0):
|
||
"""HTTP GET 请求。"""
|
||
return http_request("GET", url, headers=headers, timeout=timeout)
|
||
|
||
|
||
def http_post(url: str, body: dict = None, headers: dict = None,
|
||
timeout: float = 10.0):
|
||
"""HTTP POST 请求（JSON body）。"""
|
||
return http_request("POST", url, body=body, headers=headers, timeout=timeout)
|
||
|
||
|
||
def http_post_raw(url: str, raw_body: bytes, content_type: str,
|
||
headers: dict = None, timeout: float = 10.0):
|
||
"""HTTP POST 请求（原始二进制 body）。"""
|
||
return http_request("POST", url, raw_body=raw_body,
|
||
raw_body_content_type=content_type,
|
||
headers=headers, timeout=timeout)
|
||
|
||
|
||
def try_parse_json(text: str):
|
||
"""尝试解析 JSON，失败返回 None。"""
|
||
try:
|
||
return json.loads(text)
|
||
except (json.JSONDecodeError, ValueError):
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# pymsg 协议客户端
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class PymsgClient:
|
||
"""pymsg TCP 协议客户端（用于直接与 db:8005 通信）。
|
||
|
||
协议格式: [4字节大端长度][UTF-8 JSON payload]
|
||
"""
|
||
|
||
def __init__(self, host: str = "127.0.0.1", port: int = 8005,
|
||
name: str = "test_client"):
|
||
self.host = host
|
||
self.port = port
|
||
self.name = name
|
||
self._sock = None
|
||
|
||
def _ensure_connected(self):
|
||
if self._sock is None:
|
||
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
self._sock.settimeout(10)
|
||
self._sock.connect((self.host, self.port))
|
||
|
||
def _close(self):
|
||
if self._sock:
|
||
try:
|
||
self._sock.close()
|
||
except Exception:
|
||
pass
|
||
self._sock = None
|
||
|
||
def _send_frame(self, data: str):
|
||
"""发送一帧 pymsg 数据。"""
|
||
payload = data.encode("utf-8")
|
||
header = struct.pack(">I", len(payload))
|
||
self._sock.sendall(header + payload)
|
||
|
||
def _recv_frame(self, timeout: float = 10.0) -> str:
|
||
"""接收一帧 pymsg 数据。"""
|
||
self._sock.settimeout(timeout)
|
||
header = self._recv_exact(4, timeout)
|
||
if header is None:
|
||
raise ConnectionError("Failed to receive pymsg header")
|
||
length = struct.unpack(">I", header)[0]
|
||
payload = self._recv_exact(length, timeout)
|
||
if payload is None:
|
||
raise ConnectionError("Failed to receive pymsg payload")
|
||
return payload.decode("utf-8")
|
||
|
||
def _recv_exact(self, n: int, timeout: float = 10.0):
|
||
"""精确接收 n 字节。"""
|
||
self._sock.settimeout(timeout)
|
||
data = b""
|
||
while len(data) < n:
|
||
try:
|
||
chunk = self._sock.recv(n - len(data))
|
||
except socket.timeout:
|
||
return None
|
||
if not chunk:
|
||
return None
|
||
data += chunk
|
||
return data
|
||
|
||
def call(self, payload: dict, timeout: float = 10.0) -> dict:
|
||
"""发送请求并接收响应（短连接模式）。
|
||
|
||
payload 必须包含 "_route": "execute"。
|
||
返回 {"success": true/false, "data": [...], "error": "..."}
|
||
"""
|
||
try:
|
||
self._ensure_connected()
|
||
msg = json.dumps(payload)
|
||
self._send_frame(msg)
|
||
resp_text = self._recv_frame(timeout)
|
||
return json.loads(resp_text)
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
finally:
|
||
self._close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 服务可用性检查
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def check_tcp_service(host: str, port: int, timeout: float = 5.0) -> tuple:
|
||
"""检查纯 TCP 服务（如 pymsg 协议服务 db/captcha/smtp）是否可达。
|
||
返回 (reachable: bool, detail: str)。
|
||
"""
|
||
try:
|
||
with socket.create_connection((host, port), timeout=timeout):
|
||
return True, "TCP 连接成功"
|
||
except ConnectionRefusedError:
|
||
return False, "连接被拒绝 (服务未启动或端口未监听)"
|
||
except socket.timeout:
|
||
return False, f"TCP 连接超时 ({timeout}s)"
|
||
except OSError as e:
|
||
return False, f"网络错误: {e}"
|
||
|
||
|
||
def check_service(url: str, timeout: float = 8.0) -> tuple:
|
||
"""检查服务是否可达（HTTP 请求检测，与测试实际行为一致）。
|
||
返回 (reachable: bool, detail: str)。
|
||
任何 HTTP 响应（包括 200/302/401/403 等）均视为可达。
|
||
"""
|
||
parsed = urllib.parse.urlparse(url)
|
||
ctx = ssl._create_unverified_context() if parsed.scheme == "https" else None
|
||
try:
|
||
req = urllib.request.Request(url, method="HEAD")
|
||
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
|
||
return True, f"HTTP {resp.status}"
|
||
except urllib.error.HTTPError as e:
|
||
# 任何 HTTP 错误码都说明服务在运行
|
||
return True, f"HTTP {e.code}"
|
||
except urllib.error.URLError as e:
|
||
reason = str(e.reason) if hasattr(e, 'reason') else str(e)
|
||
return False, reason
|
||
except socket.timeout:
|
||
return False, f"HTTP 请求超时 ({timeout}s)"
|
||
except ConnectionRefusedError:
|
||
return False, "连接被拒绝 (服务未启动或端口未监听)"
|
||
except socket.gaierror:
|
||
return False, "DNS 解析失败"
|
||
except OSError as e:
|
||
return False, f"网络错误: {e.strerror if hasattr(e, 'strerror') else str(e)}"
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# gateway 测试专用工具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def random_xff() -> str:
|
||
"""生成随机 X-Forwarded-For 值，用于隔离 gateway 的 IP 级限流桶。
|
||
|
||
背景：gateway 直连模式下信任客户端 XFF 头（审计发现 A4），每个测试场景
|
||
使用独立随机 XFF，可避免限流用例之间互相消耗配额（如登录 5 次/5 分钟）。
|
||
"""
|
||
return f"10.{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}"
|
||
|
||
|
||
# 响应体泄露特征：源码文件名/路径/C++ 类型签名（P0-2/P0-3 脱敏回归基线）
|
||
_LEAK_PATTERNS = ("main.cpp", ".hpp", ".cpp:", "/opt/", "boost::", "std::",
|
||
"source_location", "typeid", "__cxa")
|
||
|
||
|
||
def assert_no_internal_leak(body: str, name: str, counter: Counter):
|
||
"""断言响应体不含内部实现信息（源码路径/C++ 类型名等，P0-2/P0-3 回归）。"""
|
||
hits = [p for p in _LEAK_PATTERNS if p in body]
|
||
if not hits:
|
||
print(f" [PASS] {name}")
|
||
counter.add_pass()
|
||
return True
|
||
else:
|
||
print(f" [FAIL] {name}: 响应泄露内部信息 {hits}")
|
||
counter.add_fail(name, f"leak patterns {hits}")
|
||
return False
|
||
|
||
|
||
def db_execute(action: str, table: str, *, where: dict = None,
|
||
values: dict = None, host: str = "127.0.0.1",
|
||
port: int = 8005, timeout: float = 10.0) -> dict:
|
||
"""通过 pymsg 调 db execute 路由（测试数据准备/清理用）。
|
||
|
||
返回 db 完整响应 {"success": ..., "data": ...}；db 不可达时
|
||
返回 {"success": False, "error": ...}（PymsgClient 内部兜底）。
|
||
"""
|
||
payload = {"_route": "execute", "action": action, "table": table}
|
||
if where is not None:
|
||
payload["where"] = where
|
||
if values is not None:
|
||
payload["values"] = values
|
||
return PymsgClient(host, port, "test_blackbox").call(payload, timeout)
|