diff --git a/tests/.gitignore b/tests/.gitignore
new file mode 100644
index 0000000..ef8f28f
--- /dev/null
+++ b/tests/.gitignore
@@ -0,0 +1,111 @@
+#prjs
+*.bin
+*.build
+*.dist
+*.onefile-build
+nuitka-crash-report.xml
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
diff --git a/tests/blackbox/README.md b/tests/blackbox/README.md
new file mode 100644
index 0000000..8ff3e17
--- /dev/null
+++ b/tests/blackbox/README.md
@@ -0,0 +1,122 @@
+# 黑盒用户模拟测试
+
+micro_servers 微服务生态的黑盒测试套件。模拟真实用户操作流程，不依赖内部实现，仅通过 HTTP/TCP 公开接口验证各服务行为。
+
+## 设计原则
+
+- **零依赖**: 仅使用 Python 标准库（`http.client`、`json`、`urllib`、`socket`、`argparse`）
+- **用户视角**: 每个测试文件模拟一条完整的用户旅程，而非孤立 API 端点
+- **无测试框架**: 纯 `assert` + 计数器，输出 `[PASS]`/`[FAIL]`/`[SKIP]`
+- **独立可运行**: 每个 `test_xxx.py` 可单独执行，也可通过 `test_runner.py` 统一调度
+
+## 文件结构
+
+```
+tests/blackbox/
+├── README.md           # 本文档
+├── config.py           # 服务地址与端口映射
+├── helpers.py          # HTTP/TCP 请求封装 + 断言工具 + 计数器
+├── test_runner.py      # 统一入口，--url 自动识别目标服务
+├── test_gateway.py     # 用户注册 → 登录 → Token → 代理转发
+├── test_captcha.py     # 获取验证码 → 校验 → 一次性消费
+├── test_smtp.py        # 发送邮箱验证码 → 校验
+├── test_dnfs.py        # 创建笔记 → 查询 → 更新 → 删除（含内容审核）
+├── test_db.py          # 日志写入 → 查询 → 统计 → 清理
+├── test_ai_review.py   # 内容审核 → 人工反馈纠正
+├── test_static.py      # 首页加载 → 文件浏览 → 视频流 → 路径遍历防护
+└── test_waf.py         # 正常请求通行 → SQLi/XSS/命令注入/路径遍历/扫描器拦截
+```
+
+## 覆盖的用户旅程
+
+### 1. 注册与登录 (gateway + captcha + smtp)
+```
+用户访问 captcha:8009 获取图形验证码
+    → 提交 captcha_token + answer 校验
+    → 访问 smtp:8008 发送邮箱验证码
+    → 提交 email + code 完成邮箱验证
+    → 访问 gateway:8443 POST /register 注册账号
+    → 访问 gateway:8443 POST /login 登录获取 Token
+    → 使用 Token 通过 gateway 代理访问后端服务
+```
+
+### 2. 笔记管理 (dnfs + ai_review + db)
+```
+已登录用户携带 Bearer Token
+    → POST /add_note 创建笔记（内容经 ai_review:8007 审核）
+    → POST /get_notes 查询笔记列表
+    → POST /update_note 更新笔记（内容重新审核）
+    → POST /delete_note 删除笔记
+```
+
+### 3. 内容审核 (ai_review)
+```
+提交违规内容 → 应返回 invalid: true
+提交正常内容 → 应返回 invalid: false
+提交误判反馈 → RAG 知识库更新
+```
+
+### 4. 集中日志 (db)
+```
+各服务通过 /log/ingest 推送日志
+    → /log/query 按条件查询
+    → /log/stats 聚合统计
+    → /log/purge 按时间清理
+```
+
+### 5. 前端静态服务 (static)
+```
+访问首页 → 返回 login.html
+访问文件浏览 → /api/filelist 返回目录结构
+访问视频流 → 支持 Range 分段请求
+路径遍历攻击 → 返回 403
+```
+
+### 6. WAF 安全防护 (waf)
+```
+正常请求 → 通过代理到达 gateway
+SQL 注入载荷 → 返回 403 blocked by WAF
+XSS 跨站脚本 → 返回 403
+命令注入 → 返回 403
+路径遍历 → 返回 403
+扫描器 UA → 返回 403
+```
+
+## 运行方式
+
+```bash
+# 运行全部测试
+python3 test_runner.py
+
+# 指定服务地址（自动识别服务类型）
+python3 test_runner.py --url http://127.0.0.1:8443
+
+# 运行单个服务测试
+python3 test_gateway.py
+python3 test_captcha.py
+# ... 依此类推
+
+# 列出所有可用测试
+python3 test_runner.py --list
+```
+
+## 服务端口默认值
+
+| 服务 | 默认端口 | 绑定地址 |
+|------|---------|---------|
+| gateway | 8443 | 127.0.0.1 |
+| static | 9200 | 127.0.0.1 |
+| dnfs | 8006 | 127.0.0.1 |
+| db | 8005 | 127.0.0.1 |
+| ai_review | 8007 | 127.0.0.1 |
+| captcha | 8009 | 127.0.0.1 |
+| smtp | 8008 | 127.0.0.1 |
+| waf | 443 | 0.0.0.0 |
+
+## 注意事项
+
+- 测试依赖服务运行中，请先通过 `core/init.txt` 启动全部服务
+- 注册测试会创建真实数据库记录，测试完成后可手动清理
+- SMTP 测试需要真实的邮件服务配置，若未配置则相关测试标记为 SKIP
+- WAF 测试需要服务监听 443 端口（可能需要 sudo）
+- 所有测试设计为幂等：重复运行不会产生副作用累积
diff --git a/tests/blackbox/config.py b/tests/blackbox/config.py
new file mode 100644
index 0000000..1854538
--- /dev/null
+++ b/tests/blackbox/config.py
@@ -0,0 +1,91 @@
+"""
+micro_servers 黑盒测试 — 服务地址与端口配置。
+
+每个服务定义: name, port, description, host。
+test_runner.py 通过 --url 参数自动识别目标服务。
+"""
+
+SERVICES = {
+    "gateway": {
+        "name": "gateway",
+        "host": "127.0.0.1",
+        "port": 8443,
+        "description": "应用网关 (认证+反向代理)",
+        "test_file": "test_gateway.py",
+    },
+    "static": {
+        "name": "static",
+        "host": "127.0.0.1",
+        "port": 9200,
+        "description": "前端静态服务 (文件托管+视频流)",
+        "test_file": "test_static.py",
+    },
+    "dnfs": {
+        "name": "dnfs",
+        "host": "127.0.0.1",
+        "port": 8006,
+        "description": "树状笔记服务 (CRUD+内容审核)",
+        "test_file": "test_dnfs.py",
+    },
+    "db": {
+        "name": "db",
+        "host": "127.0.0.1",
+        "port": 8005,
+        "description": "通用数据库服务 (CRUD+集中日志)",
+        "test_file": "test_db.py",
+    },
+    "ai_review": {
+        "name": "ai_review",
+        "host": "127.0.0.1",
+        "port": 8007,
+        "description": "AI 内容审核服务 (DeepSeek+ChromaDB RAG)",
+        "test_file": "test_ai_review.py",
+    },
+    "captcha": {
+        "name": "captcha",
+        "host": "127.0.0.1",
+        "port": 8009,
+        "description": "验证码服务 (图片生成+校验)",
+        "test_file": "test_captcha.py",
+    },
+    "smtp": {
+        "name": "smtp",
+        "host": "127.0.0.1",
+        "port": 8008,
+        "description": "邮件验证服务 (发送+校验)",
+        "test_file": "test_smtp.py",
+    },
+    "waf": {
+        "name": "waf",
+        "host": "127.0.0.1",
+        "port": 443,
+        "description": "WAF 安全网关 (TLS+Coraza规则引擎)",
+        "test_file": "test_waf.py",
+    },
+}
+
+# 端口到服务的反向映射
+PORT_MAP = {v["port"]: v["name"] for v in SERVICES.values()}
+
+
+def detect_service(url: str):
+    """从 URL 中识别目标服务。返回服务名或 None。"""
+    from urllib.parse import urlparse
+
+    parsed = urlparse(url)
+    port = parsed.port
+    if port is None:
+        # 默认端口猜测
+        if parsed.scheme == "https":
+            port = 443
+        else:
+            port = 80
+    return PORT_MAP.get(port)
+
+
+def get_url(name: str) -> str:
+    """获取服务的完整 HTTP URL。"""
+    svc = SERVICES.get(name)
+    if not svc:
+        raise ValueError(f"未知服务: {name}")
+    return f"http://{svc['host']}:{svc['port']}"
diff --git a/tests/blackbox/helpers.py b/tests/blackbox/helpers.py
new file mode 100644
index 0000000..6509873
--- /dev/null
+++ b/tests/blackbox/helpers.py
@@ -0,0 +1,313 @@
+"""
+micro_servers 黑盒测试 — 共享工具模块。
+
+提供: HTTP 请求封装、断言工具、测试计数器、pymsg 协议客户端。
+纯标准库实现，零外部依赖。
+"""
+
+import http.client
+import json
+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_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)
diff --git a/tests/blackbox/test_ai_review.py b/tests/blackbox/test_ai_review.py
new file mode 100644
index 0000000..e7b893e
--- /dev/null
+++ b/tests/blackbox/test_ai_review.py
@@ -0,0 +1,154 @@
+"""
+ai_review 黑盒用户模拟测试 — 内容审核 + 人工反馈。
+
+用户旅程:
+  1. 提交明显违规内容 → 应返回 invalid: true
+  2. 提交正常无害内容 → 应返回 invalid: false
+  3. 提交空内容 → 应返回 400 错误
+  4. 缺少 content 字段 → 应返回 400 错误
+  5. 人工反馈纠正误判 → RAG 知识库更新
+  6. 查看 RAG 列表
+  7. 健康检查
+
+注意: ai_review 依赖 DeepSeek API + ChromaDB，API 调用可能因网络/配额失败。
+测试使用了通用违规关键词，实际审核结果取决于 DeepSeek 模型判断。
+"""
+
+import json
+import random
+import string
+
+from helpers import (
+    http_get, http_post, try_parse_json,
+    assert_eq, assert_status, assert_true, assert_json_key,
+    skip, Counter, check_service,
+)
+from config import get_url
+
+AI_REVIEW_URL = get_url("ai_review")
+
+
+def run(url: str = None):
+    """ai_review 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    au = url or AI_REVIEW_URL
+    c = Counter()
+
+    print(f"\n--- ai_review 内容审核流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    if not check_service(au)[0]:
+        skip("全部测试", "ai_review 服务不可达", c)
+        return c.passed, c.failed, c.skipped
+
+    # ------------------------------------------------------------------
+    # 场景 1: 提交明显违规内容 — 应标记为违规
+    # ------------------------------------------------------------------
+    violation_texts = [
+        "出售毒品冰毒海洛因，价格优惠，联系QQ123456",
+        "提供赌博网站搭建服务，包月维护，日入过万",
+    ]
+    violation_text = random.choice(violation_texts)
+
+    status, body, _ = http_post(f"{au}/review", {"content": violation_text})
+    assert_eq(status, 200, "审核-违规内容-响应 (200)", c)
+    data = try_parse_json(body)
+    if data:
+        # 注意: invalid 可能是字符串 "true"/"false" 而非布尔值
+        invalid = data.get("invalid")
+        is_invalid = invalid in (True, "true", "True")
+        assert_true(
+            is_invalid,
+            "审核-违规内容-标记为invalid", c,
+            f"invalid={invalid}, body={body[:200]}",
+        )
+    else:
+        assert_true(False, "审核-违规内容-解析失败", c, f"body={body[:200]}")
+
+    # ------------------------------------------------------------------
+    # 场景 2: 提交正常无害内容 — 应通过
+    # ------------------------------------------------------------------
+    safe_texts = [
+        "今天天气真好，适合出去散步。",
+        "Python 是一门优秀的编程语言，广泛应用于数据科学和人工智能领域。",
+        "微服务架构是一种将应用拆分为多个小型独立服务的软件设计方法。",
+    ]
+    safe_text = random.choice(safe_texts)
+
+    status, body, _ = http_post(f"{au}/review", {"content": safe_text})
+    assert_eq(status, 200, "审核-正常内容-响应 (200)", c)
+    data = try_parse_json(body)
+    if data:
+        invalid = data.get("invalid")
+        is_invalid = invalid in (True, "true", "True")
+        # 正常内容应该不违规
+        assert_true(
+            not is_invalid,
+            "审核-正常内容-标记为safe", c,
+            f"invalid={invalid}, body={body[:200]}",
+        )
+    else:
+        assert_true(False, "审核-正常内容-解析失败", c, f"body={body[:200]}")
+
+    # ------------------------------------------------------------------
+    # 场景 3: 提交空内容 — 应返回 400
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{au}/review", {"content": ""})
+    assert_eq(status, 400, "审核-空内容-拒绝 (400)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 4: 缺少 content 字段 — 应返回 400
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{au}/review", {})
+    assert_eq(status, 400, "审核-缺少content-拒绝 (400)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 5: RAG 人工反馈 — 纠正误判（标注一条违规内容）
+    # ------------------------------------------------------------------
+    feedback_text = f"test_feedback_{random.randint(10000, 99999)}: 非法内容"
+    status, body, _ = http_post(f"{au}/rag/feedback", {
+        "content": feedback_text,
+        "correct_type": "violation",
+    })
+    assert_eq(status, 200, "RAG反馈-添加violation (200)", c)
+    data = try_parse_json(body)
+    if data:
+        assert_true(
+            data.get("added") in (True, False),
+            "RAG反馈-返回added字段", c,
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 6: RAG 反馈 — 参数错误
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{au}/rag/feedback", {
+        "content": "test",
+        "correct_type": "invalid_type",
+    })
+    assert_eq(status, 400, "RAG反馈-错误类型-拒绝 (400)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 7: RAG 列表查询
+    # ------------------------------------------------------------------
+    status, body, _ = http_get(f"{au}/rag/list")
+    assert_eq(status, 200, "RAG列表-查询 (200)", c)
+    data = try_parse_json(body)
+    if data:
+        assert_json_key(data, "total", "RAG列表-有total字段", c)
+
+    # ------------------------------------------------------------------
+    # 场景 8: 健康检查
+    # ------------------------------------------------------------------
+    status, body, _ = http_get(f"{au}/health")
+    assert_eq(status, 200, "健康检查 (200)", c)
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  ai_review 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()
diff --git a/tests/blackbox/test_captcha.py b/tests/blackbox/test_captcha.py
new file mode 100644
index 0000000..93a5a68
--- /dev/null
+++ b/tests/blackbox/test_captcha.py
@@ -0,0 +1,147 @@
+"""
+captcha 黑盒用户模拟测试 — 获取验证码 → 校验 → 一次性消费。
+
+用户旅程:
+  1. 请求 /captcha/image 获取验证码图片 + token
+  2. 使用正确答案校验 → 应通过
+  3. 使用错误答案校验 → 应拒绝
+  4. 同一 token 重复校验 → 应拒绝 (一次性)
+  5. 缺少 token/answer → 应报错
+  6. 不存在的 token → 应拒绝
+  7. 健康检查
+"""
+
+import json
+import string
+
+from helpers import (
+    http_get, http_post, http_request,
+    assert_eq, assert_status, assert_true, assert_json_key, assert_contains,
+    skip, Counter, check_service,
+)
+from config import get_url
+
+CAPTCHA_URL = get_url("captcha")
+
+
+def run(url: str = None):
+    """captcha 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    cu = url or CAPTCHA_URL
+    c = Counter()
+
+    print(f"\n--- captcha 验证码获取与校验流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    if not check_service(cu)[0]:
+        print("  captcha 服务不可达，跳过所有测试")
+        c.add_skip()
+        return c.passed, c.failed, c.skipped
+
+    # ------------------------------------------------------------------
+    # 场景 1: 获取验证码图片
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{cu}/captcha/image")
+    assert_eq(status, 200, "获取验证码图片 (200)", c)
+    captcha_token = headers.get("X-Captcha-Token", headers.get("x-captcha-token", ""))
+    assert_true(
+        len(captcha_token) == 32,
+        "获取验证码-Token 32位hex", c,
+        f"token={captcha_token[:10]}...",
+    )
+    # 验证返回的是图片
+    content_type = headers.get("Content-Type", headers.get("content-type", ""))
+    assert_contains(content_type, "image", "获取验证码-图片类型", c)
+
+    # ------------------------------------------------------------------
+    # 场景 2: 缺少参数 — token 和 answer 都缺
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{cu}/captcha/verify", {})
+    data = json.loads(body) if body else {}
+    assert_true(
+        not data.get("valid", True),
+        "校验-缺少参数-拒绝", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 3: 不存在的 token
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{cu}/captcha/verify", {
+        "token": "0" * 32,
+        "answer": "abcde",
+    })
+    data = json.loads(body) if body else {}
+    assert_true(
+        not data.get("valid", True),
+        "校验-不存在token-拒绝", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 4: 错误答案
+    # ------------------------------------------------------------------
+    if captcha_token:
+        status, body, _ = http_post(f"{cu}/captcha/verify", {
+            "token": captcha_token,
+            "answer": "ZZZZZ",
+        })
+        data = json.loads(body) if body else {}
+        assert_true(
+            not data.get("valid", True),
+            "校验-错误答案-拒绝", c,
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 5: 验证码一次性消费 — 拿新 token 分别测正确答案和二次使用
+    # ------------------------------------------------------------------
+    # 重新获取一个 captcha（上一步的 token 已被消费）
+    status2, body2, headers2 = http_get(f"{cu}/captcha/image")
+    assert_eq(status2, 200, "获取第二个验证码 (200)", c)
+
+    # 没法从图片自动识别文字，但我们可以测试一次性消费的行为
+    # 先用错误答案消费 token
+    token2 = headers2.get("X-Captcha-Token", headers.get("x-captcha-token", ""))
+    if token2:
+        # 第一次：错误答案
+        status_a, body_a, _ = http_post(f"{cu}/captcha/verify", {
+            "token": token2,
+            "answer": "WRONG",
+        })
+        data_a = json.loads(body_a) if body_a else {}
+        assert_true(
+            not data_a.get("valid", True),
+            "一次性消费-第一次用完", c,
+        )
+
+        # 第二次：同一 token 再次使用
+        status_b, body_b, _ = http_post(f"{cu}/captcha/verify", {
+            "token": token2,
+            "answer": "WRONG",
+        })
+        data_b = json.loads(body_b) if body_b else {}
+        assert_true(
+            not data_b.get("valid", True),
+            "一次性消费-重复使用拒绝", c,
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 6: 健康检查
+    # ------------------------------------------------------------------
+    status, body, _ = http_get(f"{cu}/health")
+    assert_eq(status, 200, "健康检查 (200)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 7: 不存在路径 → 404
+    # ------------------------------------------------------------------
+    status, body, _ = http_get(f"{cu}/nonexistent")
+    assert_eq(status, 404, "不存在路径-404", c)
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  captcha 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()
diff --git a/tests/blackbox/test_db.py b/tests/blackbox/test_db.py
new file mode 100644
index 0000000..ca922b2
--- /dev/null
+++ b/tests/blackbox/test_db.py
@@ -0,0 +1,190 @@
+"""
+db 黑盒用户模拟测试 — 集中日志: 写入 → 查询 → 统计 → 清理。
+
+用户旅程:
+  1. 通过 pymsg 协议执行 CRUD (select/insert)
+  2. 通过 HTTP /log/ingest 写入日志
+  3. 通过 HTTP /log/query 查询日志
+  4. 通过 HTTP /log/stats 查看统计
+  5. 通过 HTTP /log/health 健康检查
+
+注意: dbv2.py (pymsg 模式) 不挂载 Flask log_module Blueprint，
+因此 HTTP /log/* 路由可能不可用。测试会优雅降级。
+"""
+
+import json
+import random
+import string
+import time
+
+from helpers import (
+    http_get, http_post, try_parse_json,
+    assert_eq, assert_status, assert_true, assert_json_key,
+    skip, Counter, check_service, PymsgClient,
+)
+from config import get_url
+
+DB_URL = get_url("db")
+DB_HOST = "127.0.0.1"
+DB_PORT = 8005
+
+
+def run(url: str = None):
+    """db 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    du = url or DB_URL
+    c = Counter()
+
+    print(f"\n--- db 数据库与日志流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    if not check_service(du)[0]:
+        skip("全部测试", "db 服务不可达", c)
+        return c.passed, c.failed, c.skipped
+
+    # ------------------------------------------------------------------
+    # 第一部分: pymsg 协议 CRUD
+    # ------------------------------------------------------------------
+    print("  [pymsg 协议测试]")
+
+    client = PymsgClient(DB_HOST, DB_PORT, "test_blackbox")
+
+    # 场景 1: pymsg 查询 Users 表
+    result = client.call({
+        "_route": "execute",
+        "action": "select",
+        "table": "Users",
+        "limit": 5,
+    })
+    assert_true(
+        result.get("success") is True,
+        "pymsg-查询Users表", c,
+        f"error={result.get('error', '')}",
+    )
+    if result.get("success"):
+        data = result.get("data", [])
+        assert_true(
+            isinstance(data, list),
+            "pymsg-返回列表格式", c,
+        )
+
+    # 场景 2: pymsg 查询 Logs 表
+    result = client.call({
+        "_route": "execute",
+        "action": "select",
+        "table": "Logs",
+        "limit": 3,
+        "order_by": [{"column": "id", "desc": True}],
+    })
+    assert_true(
+        result.get("success") is True,
+        "pymsg-查询Logs表", c,
+        f"error={result.get('error', '')}",
+    )
+
+    # 场景 3: pymsg 使用 where 条件
+    result = client.call({
+        "_route": "execute",
+        "action": "select",
+        "table": "Users",
+        "columns": ["id", "username"],
+        "where": {"id": {"gt": 0}},
+        "limit": 1,
+    })
+    assert_true(
+        result.get("success") is True,
+        "pymsg-条件查询", c,
+        f"error={result.get('error', '')}",
+    )
+
+    # 场景 4: pymsg 无效路由
+    result = client.call({
+        "_route": "nonexistent_route",
+        "action": "select",
+        "table": "Users",
+    })
+    assert_true(
+        result.get("success") is False,
+        "pymsg-无效路由-拒绝", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 第二部分: HTTP /log/* 接口 (Flask log_module)
+    # ------------------------------------------------------------------
+    print("\n  [HTTP /log/* 接口测试]")
+
+    # 检查 /log/health 是否可用（Flask log_module 是否挂载）
+    status, body, _ = http_get(f"{du}/log/health")
+    if status == 0 or status >= 500:
+        skip("HTTP /log/* 测试", "log_module 未挂载 (dbv2 使用 pymsg 模式, 非 Flask)", c)
+    else:
+        # 场景 5: /log/health 健康检查
+        assert_eq(status, 200, "log-健康检查 (200)", c)
+
+        # 场景 6: /log/ingest 写入日志
+        test_ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+        status, body, _ = http_post(f"{du}/log/ingest", {
+            "logs": [{
+                "service": "dnfs",
+                "level": "INFO",
+                "message": f"blackbox_test_{random.randint(10000, 99999)}",
+                "ts": test_ts,
+            }],
+        })
+        data = try_parse_json(body)
+        assert_true(
+            status == 200 and data is not None and data.get("success") is True,
+            "log-写入日志", c,
+            f"status={status} body={body[:150]}",
+        )
+
+        # 场景 7: /log/ingest — 空日志列表
+        status, body, _ = http_post(f"{du}/log/ingest", {"logs": []})
+        assert_eq(status, 200, "log-空日志列表 (200)", c)
+
+        # 场景 8: /log/ingest — 缺少 logs 字段
+        status, body, _ = http_post(f"{du}/log/ingest", {})
+        data = try_parse_json(body)
+        assert_true(
+            data is not None and not data.get("success", True),
+            "log-缺少logs字段-拒绝", c,
+        )
+
+        # 场景 9: /log/query 查询日志
+        status, body, _ = http_post(f"{du}/log/query", {
+            "limit": 5,
+            "desc": True,
+        })
+        data = try_parse_json(body)
+        assert_true(
+            status == 200 and data is not None and data.get("success") is True,
+            "log-查询日志", c,
+            f"status={status}",
+        )
+        if data and data.get("success"):
+            assert_true(
+                "logs" in data.get("data", {}),
+                "log-查询返回logs字段", c,
+            )
+
+        # 场景 10: /log/stats 统计
+        status, body, _ = http_post(f"{du}/log/stats", {
+            "group_by": "service",
+        })
+        data = try_parse_json(body)
+        assert_true(
+            status == 200 and data is not None,
+            "log-统计", c,
+            f"status={status}",
+        )
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  db 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()
diff --git a/tests/blackbox/test_dnfs.py b/tests/blackbox/test_dnfs.py
new file mode 100644
index 0000000..8f26a93
--- /dev/null
+++ b/tests/blackbox/test_dnfs.py
@@ -0,0 +1,229 @@
+"""
+dnfs 黑盒用户模拟测试 — 创建笔记 → 查询 → 更新 → 删除（含内容审核）。
+
+用户旅程:
+  已登录用户携带 Bearer Token:
+  1. 创建根节点笔记（内容经 ai_review 审核）
+  2. 创建子节点笔记
+  3. 查询节点列表
+  4. 更新笔记标题和内容
+  5. 删除叶子节点（无子节点）
+  6. 删除非空节点（应拒绝）
+  7. 无 Token 操作（应拒绝）
+  8. 创建时缺少必填参数（应报错）
+
+前提: gateway 已启动，有已注册用户和有效 Token。
+"""
+
+import json
+import random
+import string
+
+from helpers import (
+    http_post, try_parse_json,
+    assert_eq, assert_status, assert_true, assert_json_key,
+    skip, Counter, check_service,
+)
+from config import get_url, SERVICES
+
+DNFS_URL = get_url("dnfs")
+GATEWAY_URL = get_url("gateway")
+
+
+def _random_test_user():
+    suffix = ''.join(random.choices(string.ascii_lowercase, k=6))
+    return f"dtest_{suffix}@test.local"
+
+
+def _register_and_login(gw_url: str, counter: Counter):
+    """通过 gateway 注册并登录，返回 (token, email)。失败返回 (None, None)。"""
+    email = _random_test_user()
+    password = "TestP@ss123"
+
+    # 注册
+    status, body, _ = http_post(f"{gw_url}/register", {
+        "username": email,
+        "password": password,
+    })
+    if status != 200:
+        return None, None
+
+    data = try_parse_json(body)
+    if not data or not data.get("success"):
+        return None, None
+
+    # 登录
+    status, body, _ = http_post(f"{gw_url}/login", {
+        "username": email,
+        "password": password,
+    })
+    if status != 200:
+        return None, None
+
+    data = try_parse_json(body)
+    if not data or not data.get("success"):
+        return None, None
+
+    token = (data.get("data") or {}).get("token", "")
+    return token, email
+
+
+def run(url: str = None):
+    """dnfs 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    du = url or DNFS_URL
+    c = Counter()
+
+    print(f"\n--- dnfs 笔记管理流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    if not check_service(du)[0]:
+        skip("全部测试", "dnfs 服务不可达", c)
+        return c.passed, c.failed, c.skipped
+
+    # ------------------------------------------------------------------
+    # 准备: 注册用户并获取 Token
+    # ------------------------------------------------------------------
+    gw_available = check_service(GATEWAY_URL)[0]
+    if not gw_available:
+        skip("全部测试", "gateway 服务不可达，无法获取 Token", c)
+        return c.passed, c.failed, c.skipped
+
+    token, email = _register_and_login(GATEWAY_URL, c)
+    if not token:
+        skip("全部测试", "无法完成注册/登录获取 Token", c)
+        return c.passed, c.failed, c.skipped
+
+    auth_headers = {"Authorization": f"Bearer {token}"}
+
+    # ------------------------------------------------------------------
+    # 场景 1: 无 Token 创建笔记 — 应拒绝
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{du}/add_note", {
+        "pid": 0,
+        "title": "test",
+        "content": "test content",
+    })
+    assert_eq(status, 401, "创建笔记-无Token-拒绝 (401)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 2: 创建笔记 — 缺少参数
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{du}/add_note", {
+        "pid": 0,
+    }, headers=auth_headers)
+    assert_eq(status, 400, "创建笔记-缺少title/content (400)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 3: 创建根节点笔记
+    # ------------------------------------------------------------------
+    note_title = f"测试笔记_{random.randint(1000, 9999)}"
+    note_content = "这是一段正常的测试内容，用于验证笔记创建功能。"
+    status, body, _ = http_post(f"{du}/add_note", {
+        "pid": 0,
+        "title": note_title,
+        "content": note_content,
+    }, headers=auth_headers)
+    # 期望 200（审核通过则成功）
+    data = try_parse_json(body)
+    assert_true(
+        status == 200 and data is not None and data.get("success") is True,
+        "创建笔记-根节点", c,
+        f"status={status} body={body[:150]}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 4: 查询根节点下的笔记列表
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{du}/get_notes", {"pid": 0})
+    assert_eq(status, 200, "查询笔记-根节点 (200)", c)
+    notes = try_parse_json(body)
+    assert_true(
+        isinstance(notes, list),
+        "查询笔记-返回列表", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 5: 创建子节点笔记
+    # ------------------------------------------------------------------
+    # 先获取刚创建的笔记 ID
+    child_note_id = None
+    if notes and len(notes) > 0:
+        # 找到我们刚创建的笔记
+        for note in notes:
+            if note.get("title") == note_title:
+                child_note_id = note.get("id")
+                break
+
+    if child_note_id:
+        child_title = f"子笔记_{random.randint(1000, 9999)}"
+        child_content = "这是子节点的测试内容。"
+        status, body, _ = http_post(f"{du}/add_note", {
+            "pid": child_note_id,
+            "title": child_title,
+            "content": child_content,
+        }, headers=auth_headers)
+        data = try_parse_json(body)
+        assert_true(
+            status == 200 and data is not None,
+            "创建笔记-子节点", c,
+            f"status={status} body={body[:150]}",
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 6: 删除非空节点 — 应拒绝（有子节点）
+    # ------------------------------------------------------------------
+    if child_note_id:
+        status, body, _ = http_post(f"{du}/delete_note", {
+            "id": child_note_id,
+        }, headers=auth_headers)
+        data = try_parse_json(body)
+        assert_true(
+            status in (400, 403) or (data and not data.get("success")),
+            "删除笔记-非空节点-拒绝", c,
+            f"status={status} body={body[:150]}",
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 7: 更新笔记内容
+    # ------------------------------------------------------------------
+    if child_note_id:
+        updated_title = f"更新_{random.randint(1000, 9999)}"
+        updated_content = "更新后的测试内容，依然正常无害。"
+        status, body, _ = http_post(f"{du}/update_note", {
+            "id": child_note_id,
+            "pid": 0,
+            "title": updated_title,
+            "content": updated_content,
+        }, headers=auth_headers)
+        data = try_parse_json(body)
+        assert_true(
+            status == 200 and data is not None and data.get("success") is True,
+            "更新笔记-内容", c,
+            f"status={status} body={body[:150]}",
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 8: 更新笔记 — 缺少参数
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{du}/update_note", {
+        "id": child_note_id,
+    }, headers=auth_headers)
+    assert_eq(status, 400, "更新笔记-缺少参数 (400)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 9: 删除笔记 — 缺少 id 参数
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{du}/delete_note", {}, headers=auth_headers)
+    assert_eq(status, 400, "删除笔记-缺少id (400)", c)
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  dnfs 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()
diff --git a/tests/blackbox/test_gateway.py b/tests/blackbox/test_gateway.py
new file mode 100644
index 0000000..f9d06e6
--- /dev/null
+++ b/tests/blackbox/test_gateway.py
@@ -0,0 +1,218 @@
+"""
+gateway 黑盒用户模拟测试 — 注册 → 登录 → Token → 代理转发。
+
+用户旅程:
+  1. 注册新账号（含密码强度校验、邮箱格式校验）
+  2. 登录获取 Token
+  3. Token 用于代理访问后端服务
+  4. 重复注册（应失败）
+  5. 错误密码登录（应失败）
+  6. 无效 Token 代理（应拒绝）
+  7. 密码强度不足（应拒绝）
+  8. 邮箱格式错误（应拒绝）
+"""
+
+import json
+import random
+import string
+import time
+
+from helpers import (
+    http_get, http_post, try_parse_json,
+    assert_eq, assert_status, assert_true, assert_json_key,
+    Counter, check_service,
+)
+from config import get_url
+
+GATEWAY_URL = get_url("gateway")
+STATIC_URL = get_url("static")
+
+
+def _random_email():
+    """生成随机测试邮箱。"""
+    suffix = ''.join(random.choices(string.ascii_lowercase, k=8))
+    return f"test_{suffix}@example.com"
+
+
+def _random_password():
+    """生成符合强度要求的随机密码。"""
+    chars = string.ascii_letters + string.digits + "!@#$%"
+    return ''.join(random.choices(chars, k=12))
+
+
+def run(url: str = None):
+    """gateway 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    gw = url or GATEWAY_URL
+    c = Counter()
+
+    print(f"\n--- gateway 用户注册与登录流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    if not check_service(gw)[0]:
+        print("  gateway 服务不可达，跳过所有测试")
+        c.add_skip()
+        return c.passed, c.failed, c.skipped
+
+    email = _random_email()
+    password = _random_password()
+    token = None
+
+    # ------------------------------------------------------------------
+    # 场景 1: 注册 — 密码强度不足
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{gw}/register", {
+        "username": email,
+        "password": "123",
+    })
+    assert_eq(status, 401, "注册-密码太短 (401)", c)
+    data = try_parse_json(body)
+    assert_true(
+        data is not None and not data.get("success"),
+        "注册-密码太短-拒绝", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 2: 注册 — 邮箱格式错误
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{gw}/register", {
+        "username": "not-an-email",
+        "password": _random_password(),
+    })
+    assert_eq(status, 400, "注册-邮箱格式错误 (400)", c)
+    data = try_parse_json(body)
+    assert_true(
+        data is not None and not data.get("success"),
+        "注册-邮箱格式错误-拒绝", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 3: 注册 — 成功
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{gw}/register", {
+        "username": email,
+        "password": password,
+    })
+    assert_eq(status, 200, "注册-成功 (200)", c)
+    data = try_parse_json(body)
+    assert_true(
+        data is not None and data.get("success") is True,
+        "注册-返回 success=true", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 4: 注册 — 重复用户名
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{gw}/register", {
+        "username": email,
+        "password": password,
+    })
+    # 期望注册失败（用户名已存在）
+    assert_true(
+        status != 200 or (try_parse_json(body) or {}).get("success") is not True,
+        "注册-重复用户名-拒绝", c,
+        f"status={status}, body={body[:100]}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 5: 登录 — 错误密码
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{gw}/login", {
+        "username": email,
+        "password": "WrongPassword1",
+    })
+    assert_eq(status, 401, "登录-错误密码 (401)", c)
+    data = try_parse_json(body)
+    assert_true(
+        data is not None and not data.get("success"),
+        "登录-错误密码-拒绝", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 6: 登录 — 成功获取 Token
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{gw}/login", {
+        "username": email,
+        "password": password,
+    })
+    assert_eq(status, 200, "登录-成功 (200)", c)
+    data = try_parse_json(body)
+    assert_true(
+        data is not None and data.get("success") is True,
+        "登录-返回 success=true", c,
+    )
+    token_data = (data or {}).get("data", {})
+    token = token_data.get("token", "")
+    assert_true(
+        len(token) > 20 if token else False,
+        "登录-Token 不为空", c,
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 7: 代理 — 使用 Token 访问静态服务
+    # ------------------------------------------------------------------
+    if token:
+        status, body, headers = http_get(
+            f"{gw}/",
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Target-Url": STATIC_URL,
+            },
+        )
+        assert_true(
+            status in (200, 302, 301),
+            "代理-GET静态首页", c,
+            f"status={status}",
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 8: 代理 — 无 Token 应拒绝
+    # ------------------------------------------------------------------
+    status, body, _ = http_get(
+        f"{gw}/some-path",
+        headers={"Target-Url": STATIC_URL},
+    )
+    # 缺少 Token 应返回 400/401
+    assert_true(
+        status in (400, 401, 403),
+        "代理-无Token拒绝", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 9: 代理 — 无效路径（路径遍历）
+    # ------------------------------------------------------------------
+    if token:
+        status, body, _ = http_get(
+            f"{gw}/../../../etc/passwd",
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Target-Url": STATIC_URL,
+            },
+        )
+        assert_true(
+            status in (400, 403),
+            "代理-路径遍历拒绝", c,
+            f"status={status}",
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 10: 默认代理 (无 Target-Url，无需 Token)
+    # ------------------------------------------------------------------
+    status, body, _ = http_get(f"{gw}/")
+    assert_true(
+        status in (200, 302, 301),
+        "代理-默认路由", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  gateway 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()
diff --git a/tests/blackbox/test_runner.py b/tests/blackbox/test_runner.py
new file mode 100644
index 0000000..8729763
--- /dev/null
+++ b/tests/blackbox/test_runner.py
@@ -0,0 +1,188 @@
+"""
+micro_servers 黑盒测试 — 统一运行入口。
+
+用法:
+  python3 test_runner.py                     # 运行全部测试
+  python3 test_runner.py --url http://127.0.0.1:8443  # 测试指定服务
+  python3 test_runner.py --list              # 列出可用测试
+  python3 test_runner.py --service gateway   # 按服务名运行
+"""
+
+import argparse
+import importlib
+import os
+import sys
+
+# 确保当前目录在 path 中
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from config import SERVICES, PORT_MAP, detect_service, get_url
+from helpers import check_service
+
+TEST_MODULES = {
+    "gateway": "test_gateway",
+    "captcha": "test_captcha",
+    "smtp": "test_smtp",
+    "dnfs": "test_dnfs",
+    "db": "test_db",
+    "ai_review": "test_ai_review",
+    "static": "test_static",
+    "waf": "test_waf",
+}
+
+
+def run_service_test(name: str, url: str) -> tuple:
+    """运行单个服务的测试，返回 (passed, failed, skipped)。"""
+    module_name = TEST_MODULES.get(name)
+    if not module_name:
+        print(f"未知服务: {name}")
+        return 0, 0, 0
+
+    try:
+        mod = importlib.import_module(module_name)
+    except ImportError as e:
+        print(f"无法加载测试模块 {module_name}: {e}")
+        return 0, 0, 0
+
+    if not hasattr(mod, "run"):
+        print(f"模块 {module_name} 缺少 run(url) 函数")
+        return 0, 0, 0
+
+    svc = SERVICES[name]
+    print(f"\n{'='*60}")
+    print(f"  {svc['description']}  ({url})")
+    print(f"{'='*60}")
+
+    reachable, detail = check_service(url)
+    if not reachable:
+        print(f"  [WARN] 服务不可达: {url}")
+        if detail:
+            print(f"         原因: {detail}")
+        print(f"  所有测试将被跳过。请先启动服务。\n")
+        return 0, 0, 0
+
+    passed, failed, skipped = mod.run(url)
+    return passed, failed, skipped
+
+
+def run_all():
+    """运行全部服务的测试。"""
+    total_passed = 0
+    total_failed = 0
+    total_skipped = 0
+
+    for name in SERVICES:
+        url = get_url(name)
+        p, f, s = run_service_test(name, url)
+        total_passed += p
+        total_failed += f
+        total_skipped += s
+
+    total = total_passed + total_failed + total_skipped
+    print(f"\n{'='*60}")
+    print(f"  全部测试完成: {total} total, "
+          f"{total_passed} passed, {total_failed} failed, {total_skipped} skipped")
+    print(f"{'='*60}")
+
+    return total_failed
+
+
+def run_by_url(url: str):
+    """以 URL 为入口，测试全部可代理服务，内部服务自动回退 localhost。
+
+    可代理服务 (通过 WAF→gateway 代理链): waf, gateway, static, dnfs
+    内部服务 (仅 localhost 直连):           db, ai_review, captcha, smtp
+    """
+    from urllib.parse import urlparse
+
+    hostname = urlparse(url).hostname or ""
+    is_remote = hostname not in ("127.0.0.1", "localhost", "::1")
+
+    if is_remote:
+        print(f"  入口: {url}")
+        print(f"  可代理服务 (waf/gateway/static/dnfs) 通过入口测试")
+        print(f"  内部服务 (db/ai_review/captcha/smtp) 回退到 localhost")
+        print()
+
+    # 可通过入口 URL 访问的服务
+    proxyable = {"waf", "gateway", "static", "dnfs"}
+    total_passed = 0
+    total_failed = 0
+    total_skipped = 0
+
+    for name in SERVICES:
+        if name in proxyable:
+            service_url = url
+        else:
+            service_url = get_url(name)
+
+        p, f, s = run_service_test(name, service_url)
+        total_passed += p
+        total_failed += f
+        total_skipped += s
+
+    total = total_passed + total_failed + total_skipped
+    print(f"\n{'='*60}")
+    print(f"  全部测试完成: {total} total, "
+          f"{total_passed} passed, {total_failed} failed, {total_skipped} skipped")
+    print(f"{'='*60}")
+
+    return total_failed
+
+
+def run_by_name(name: str):
+    """按服务名运行测试。"""
+    svc = SERVICES.get(name)
+    if not svc:
+        print(f"未知服务: {name}")
+        print(f"可用服务: {', '.join(SERVICES.keys())}")
+        return 1
+
+    url = get_url(name)
+    p, f, s = run_service_test(name, url)
+    return f
+
+
+def list_tests():
+    """列出所有可用测试。"""
+    print("可用的黑盒测试:\n")
+    for name, svc in SERVICES.items():
+        print(f"  {name:12s}  端口 {svc['port']:<5d}  {svc['description']}")
+    print()
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="micro_servers 黑盒用户模拟测试",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+示例:
+  %(prog)s                                          # 全部服务 localhost 测试
+  %(prog)s --url https://qsont.me                  # 公网入口测试全部可代理服务
+  %(prog)s --url http://127.0.0.1:8443             # 本地入口测试全部
+  %(prog)s --service gateway                        # 仅测试指定服务
+  %(prog)s --list                                   # 列出可用测试
+        """,
+    )
+    parser.add_argument("--url", help="入口 URL，通过它测试全部可代理服务（内部服务回退 localhost）")
+    parser.add_argument("--service", help="按服务名称运行测试")
+    parser.add_argument("--list", action="store_true", help="列出所有可用测试")
+
+    args = parser.parse_args()
+
+    if args.list:
+        list_tests()
+        return 0
+
+    if args.url:
+        return run_by_url(args.url)
+
+    if args.service:
+        return run_by_name(args.service)
+
+    # 默认：运行全部
+    return run_all()
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/blackbox/test_smtp.py b/tests/blackbox/test_smtp.py
new file mode 100644
index 0000000..ada18fa
--- /dev/null
+++ b/tests/blackbox/test_smtp.py
@@ -0,0 +1,114 @@
+"""
+smtp 黑盒用户模拟测试 — 发送邮箱验证码 → 校验。
+
+用户旅程:
+  1. 发送验证码到测试邮箱
+  2. 用错误验证码校验 → 应失败
+  3. 用空参数校验 → 应报错
+
+注意: SMTP 服务需要配置真实的邮件服务器。若服务不可达或邮件发送失败，
+相关测试将标记为 SKIP。验证码有效期为 60 秒。
+"""
+
+import json
+import random
+import string
+
+from helpers import (
+    http_post, try_parse_json,
+    assert_eq, assert_status, assert_true,
+    skip, Counter, check_service,
+)
+from config import get_url
+
+SMTP_URL = get_url("smtp")
+
+
+def run(url: str = None):
+    """smtp 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    su = url or SMTP_URL
+    c = Counter()
+
+    print(f"\n--- smtp 邮件验证码发送与校验流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    if not check_service(su)[0]:
+        skip("全部测试", "smtp 服务不可达", c)
+        return c.passed, c.failed, c.skipped
+
+    # ------------------------------------------------------------------
+    # 场景 1: 发送验证码 — 缺少 email 参数
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{su}/send_email", {})
+    data = try_parse_json(body)
+    # 预期返回错误
+    assert_true(
+        status != 200 or (data and data.get("error")),
+        "发送验证码-缺少email-报错", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 2: 发送验证码 — 空 email
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{su}/send_email", {"email": ""})
+    data = try_parse_json(body)
+    assert_true(
+        status != 200 or (data and data.get("error")),
+        "发送验证码-空email-报错", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 3: 校验验证码 — 缺少参数
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{su}/verify_email", {})
+    data = try_parse_json(body)
+    assert_true(
+        status != 200 or (data and data.get("error")),
+        "校验验证码-缺少参数-报错", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 4: 校验验证码 — 错误验证码
+    # ------------------------------------------------------------------
+    test_email = f"test_{random.randint(10000, 99999)}@example.com"
+    status, body, _ = http_post(f"{su}/verify_email", {
+        "email": test_email,
+        "input_captcha": "000000",
+    })
+    data = try_parse_json(body)
+    # 未发送过的 email 验证应失败
+    assert_true(
+        status != 200 or (data and not data.get("success")),
+        "校验验证码-错误码-拒绝", c,
+        f"status={status} body={body[:100]}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 5: 发送验证码 — 不存在的邮箱域名
+    # ------------------------------------------------------------------
+    status, body, _ = http_post(f"{su}/send_email", {
+        "email": "nobody@this-domain-does-not-exist-12345.com",
+    })
+    data = try_parse_json(body)
+    # SMTP 可能成功发送请求（服务接受），或者返回错误
+    # 两种情况都算通过 — 至少服务正常响应了
+    assert_true(
+        status in (200, 400, 500) and data is not None,
+        "发送验证码-服务正常响应", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  smtp 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()
diff --git a/tests/blackbox/test_static.py b/tests/blackbox/test_static.py
new file mode 100644
index 0000000..3454d8c
--- /dev/null
+++ b/tests/blackbox/test_static.py
@@ -0,0 +1,146 @@
+"""
+static 黑盒用户模拟测试 — 首页加载 → 文件浏览 → 路径遍历防护 → 视频流。
+
+用户旅程:
+  1. 访问首页 → 返回 HTML 登录页
+  2. 访问 /dnfs.html → 树状笔记前端
+  3. 访问 /api/filelist → 目录列表 JSON
+  4. 访问静态资源 (CSS/JS) → 返回正确 MIME 类型
+  5. 路径遍历攻击 → 返回 403
+  6. 不存在的文件 → 返回 404
+  7. 健康检查
+  8. DNFS 代理接口 (CORS 预检)
+"""
+
+import json
+
+from helpers import (
+    http_get, http_post, try_parse_json,
+    assert_eq, assert_status, assert_true, assert_contains,
+    skip, Counter, check_service,
+)
+from config import get_url
+
+STATIC_URL = get_url("static")
+
+
+def run(url: str = None):
+    """static 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    su = url or STATIC_URL
+    c = Counter()
+
+    print(f"\n--- static 前端静态服务流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    if not check_service(su)[0]:
+        skip("全部测试", "static 服务不可达", c)
+        return c.passed, c.failed, c.skipped
+
+    # ------------------------------------------------------------------
+    # 场景 1: 访问首页
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/")
+    assert_eq(status, 200, "首页-加载 (200)", c)
+    content_type = headers.get("Content-Type", headers.get("content-type", ""))
+    assert_contains(content_type, "text/html", "首页-Content-Type text/html", c)
+
+    # ------------------------------------------------------------------
+    # 场景 2: 访问 DNFS 前端页面
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/dnfs.html")
+    assert_eq(status, 200, "DNFS页面-加载 (200)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 3: 访问 CSS 资源
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/css/theme.css")
+    assert_true(
+        status in (200, 404),
+        "CSS资源-请求", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 4: 访问 JS 资源
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/js/dnfs.js")
+    assert_true(
+        status in (200, 404),
+        "JS资源-请求", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 5: 路径遍历攻击 — ../ 应返回 403
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/../../../etc/passwd")
+    assert_eq(status, 403, "路径遍历-../-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 6: 路径遍历 — URL 编码 ..%2f
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/..%2f..%2f..%2fetc%2fpasswd")
+    assert_eq(status, 403, "路径遍历-URL编码-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 7: 不存在的文件 → 404
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/this-file-does-not-exist-12345.html")
+    assert_eq(status, 404, "不存在文件-404", c)
+
+    # ------------------------------------------------------------------
+    # 场景 8: 健康检查
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/health")
+    assert_eq(status, 200, "健康检查 (200)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 9: /api/filelist — 目录列表
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/api/filelist")
+    assert_eq(status, 200, "文件列表-API (200)", c)
+    data = try_parse_json(body)
+    assert_true(
+        data is not None and data.get("success") is True,
+        "文件列表-返回success", c,
+        f"body={body[:200]}",
+    )
+    if data and data.get("success"):
+        items = data.get("items", [])
+        assert_true(
+            isinstance(items, list),
+            "文件列表-返回items数组", c,
+        )
+
+    # ------------------------------------------------------------------
+    # 场景 10: /api/filelist — 路径遍历
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/api/filelist?path=../../../etc")
+    assert_true(
+        status in (403, 400),
+        "文件列表-路径遍历-拒绝", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 11: DNFS 代理接口 — OPTIONS 预检
+    # ------------------------------------------------------------------
+    status, body, headers = http_get(f"{su}/add_note")  # OPTIONS 需要特殊处理
+    # 简单的 GET 代理转发测试
+    assert_true(
+        status in (200, 400, 401, 405),
+        "DNFS代理-add_note可达", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  static 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()
diff --git a/tests/blackbox/test_waf.py b/tests/blackbox/test_waf.py
new file mode 100644
index 0000000..0c00a22
--- /dev/null
+++ b/tests/blackbox/test_waf.py
@@ -0,0 +1,139 @@
+"""
+waf 黑盒用户模拟测试 — 正常请求通行 vs 攻击载荷拦截。
+
+用户旅程:
+  1. 正常请求 → 通过
+  2. SQL 注入载荷 → 403 blocked by WAF
+  3. XSS 跨站脚本载荷 → 403
+  4. 命令注入载荷 → 403
+  5. 路径遍历载荷 → 403
+  6. 扫描器 User-Agent → 403
+
+注意: WAF 监听 443 端口 (TLS)，测试时需注意证书验证。
+"""
+
+import ssl
+import urllib.request
+import urllib.error
+
+from helpers import (
+    http_get,
+    assert_eq, assert_status, assert_true, assert_contains,
+    skip, Counter, check_service,
+)
+from config import get_url
+
+WAF_URL = "https://127.0.0.1:443"  # WAF 监听 443 需要 HTTPS
+
+
+def _waf_get(base_url: str, path: str, headers: dict = None):
+    """通过 WAF 发送 HTTPS GET 请求（跳过证书验证）。"""
+    ctx = ssl._create_unverified_context()
+    url = f"{base_url}{path}"
+    hdrs = headers or {}
+    req = urllib.request.Request(url, headers=hdrs, method="GET")
+    try:
+        with urllib.request.urlopen(req, timeout=10.0, context=ctx) as resp:
+            return resp.status, resp.read().decode("utf-8", errors="replace"), dict(resp.headers)
+    except urllib.error.HTTPError as e:
+        body = e.read().decode("utf-8", errors="replace") if e.fp else ""
+        return e.code, body, dict(e.headers) if e.headers else {}
+    except Exception as e:
+        return 0, str(e), {}
+
+
+def run(url: str = None):
+    """waf 用户模拟测试入口。返回 (passed, failed, skipped)。"""
+    wu = url or WAF_URL
+    c = Counter()
+
+    print(f"\n--- waf 安全防护流程 ---\n")
+
+    # ------------------------------------------------------------------
+    # 0. 检查服务可用性
+    # ------------------------------------------------------------------
+    reachable, detail = check_service(wu)
+    if not reachable:
+        skip("全部测试", f"WAF 不可达: {detail}", c)
+        return c.passed, c.failed, c.skipped
+
+    # ------------------------------------------------------------------
+    # 场景 1: 正常请求应通过
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/")
+    assert_true(
+        status != 403,
+        "正常请求-通过", c,
+        f"status={status}",
+    )
+
+    # ------------------------------------------------------------------
+    # 场景 2: SQL 注入 — UNION SELECT
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?id=1 UNION SELECT * FROM users")
+    assert_eq(status, 403, "SQLi-UNION SELECT-拦截 (403)", c)
+    assert_contains(body, "blocked by WAF", "SQLi-响应体 blocked", c)
+
+    # ------------------------------------------------------------------
+    # 场景 3: SQL 注入 — SQL 注释符
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?user=admin'--")
+    assert_eq(status, 403, "SQLi-注释符-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 4: XSS — script 标签
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?q=<script>alert(1)</script>")
+    assert_eq(status, 403, "XSS-script标签-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 5: XSS — onerror 事件
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?q=<img onerror=alert(1)>")
+    assert_eq(status, 403, "XSS-onerror-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 6: 命令注入 — cat /etc/passwd
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?cmd=; cat /etc/passwd")
+    assert_eq(status, 403, "命令注入-cat-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 7: 命令注入 — id 命令
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?x=| id")
+    assert_eq(status, 403, "命令注入-id-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 8: 路径遍历
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?file=../../../etc/passwd")
+    assert_eq(status, 403, "路径遍历-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 9: 路径遍历 — URL 编码
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/?file=..%2f..%2f..%2fetc%2fpasswd")
+    assert_eq(status, 403, "路径遍历-URL编码-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 10: 扫描器 User-Agent
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/", headers={"User-Agent": "sqlmap/1.0"})
+    assert_eq(status, 403, "扫描器UA-sqlmap-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 场景 11: 扫描器 User-Agent — Nikto
+    # ------------------------------------------------------------------
+    status, body, headers = _waf_get(wu, "/", headers={"User-Agent": "Nikto"})
+    assert_eq(status, 403, "扫描器UA-nikto-拦截 (403)", c)
+
+    # ------------------------------------------------------------------
+    # 汇总
+    # ------------------------------------------------------------------
+    print(f"\n  waf 测试完成: {c.summary()}")
+    return c.passed, c.failed, c.skipped
+
+
+if __name__ == "__main__":
+    run()