diff --git a/tests/unit/api/test_permission_enforcement.py b/tests/unit/api/test_permission_enforcement.py new file mode 100644 index 0000000..8937115 --- /dev/null +++ b/tests/unit/api/test_permission_enforcement.py @@ -0,0 +1,209 @@ +"""端点权限判定的 HTTP 层用例（不连数据库）。 + +## 为什么单开这一组 + +既有用例（含 `tests/integration/test_customer_service_handover_admin_mysql.py`）都通过 +`app.dependency_overrides[build_request_context]` 注入**已经带好权限**的上下文， +所以只覆盖了"有权限能通"，**覆盖不到"缺权限必须被拒"**。 + +而 2026-09-12 出过的那类事故（客服二期三个权限码 `memory:candidate:confirm` / +`memory:candidate:review` / `handover:read` 没随代码合并进环境）恰好只会在这一层暴露： + + 权限判定发生在身份解析之后（`app/api/dependencies/auth.py` → `IdentityService.resolve`）， + 而服务层测试是自己构造 `RequestContext` 的、权限字段由测试塞进去， + 所以"权限码在库里根本不存在"这类问题，单元/集成测试全绿也照样漏。 + +## 替换了什么、保留了哪些真实部分 + +- **保留**：真实路由、真实 `build_request_context` 依赖注入位、真实 + `AuthorizationService.require` 判定与 `ForbiddenAgentError` → 403 信封。 +- **替换**（只替换两处边界）： + 1. `build_request_context` → 直接返回指定权限集的上下文（跳过 JWT 与身份库查询）； + 2. `AuthorizationService` 的审计落库 → 内存替身（拒绝时会写一条 `permission.denied`）。 + +## 为什么正反两个方向都要断言 + +- **反向**（权限集里没有该权限 → 必须 403）：防"端点忘了做权限校验"。 +- **正向**（把该权限放进去 → **不能**再是 403）：把"端点要求的权限码"钉住 —— + 若有人改了端点要的权限码而没同步这里，正向会立刻变红。 + 正向之后的下游（MySQL / Redis）在单元环境不可用，返回 5xx 属正常； + 本用例只关心"不再因权限被拒"，因此断言 `status_code != 403` 而不是 `== 200`。 +""" + +from __future__ import annotations + +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from app.api.dependencies.auth import build_request_context +from app.core.contracts import RequestContext +from app.main import create_app +from app.service import authorization_service + +PERMISSION_DENIED_CODE = "AGENT_PERMISSION_DENIED" + +#: (HTTP 方法, 路径, 该端点要求的权限码, 角色, 请求体) +CASES: tuple[tuple[str, str, str, tuple[str, ...], dict[str, Any] | None], ...] = ( + ( + "GET", + "/api/v1/admin/customer-service/handover-tickets", + "handover:read", + ("admin",), + None, + ), + ( + "GET", + "/api/v1/admin/customer-service/handover-tickets/T-0001", + "handover:read", + ("admin",), + None, + ), + ( + "GET", + "/api/v1/admin/customer-profile-candidates", + "memory:candidate:review", + ("admin",), + None, + ), + ( + "POST", + "/api/v1/admin/customer-profile-candidates/1/reviews", + "memory:candidate:review", + ("admin",), + {"decision": "approved"}, + ), + ( + "POST", + "/api/v1/users/me/memory-candidates/1/decisions", + "memory:candidate:confirm", + ("customer",), + {"decision": "confirmed"}, + ), +) + + +class _FakeSession: + """只承载"拒绝时写一条审计"这一步，不碰数据库。""" + + def add(self, _instance: object) -> None: + return None + + def begin(self) -> _FakeSession: + return self + + async def __aenter__(self) -> _FakeSession: + return self + + async def __aexit__(self, *_exc: object) -> bool: + return False + + +class _FakeSessionFactory: + def __call__(self) -> _FakeSession: + return _FakeSession() + + +def _contains_code(payload: Any, code: str) -> bool: + """在响应 JSON 里递归找错误码，不依赖信封的具体层级。""" + if isinstance(payload, dict): + return any( + (isinstance(value, str) and value == code) or _contains_code(value, code) + for value in payload.values() + ) + if isinstance(payload, list): + return any(_contains_code(item, code) for item in payload) + return False + + +def _request( + monkeypatch: pytest.MonkeyPatch, + *, + method: str, + path: str, + permissions: tuple[str, ...], + roles: tuple[str, ...], + body: dict[str, Any] | None, +) -> Any: + app = create_app() + + async def override_context() -> RequestContext: + return RequestContext( + user_id="9003", + trace_id="permission-enforcement-trace", + roles=roles, + permissions=permissions, + ) + + app.dependency_overrides[build_request_context] = override_context + monkeypatch.setattr(authorization_service, "SessionFactory", _FakeSessionFactory()) + try: + # 正向用例会走到下游（MySQL / Redis），单元环境不可用会抛异常； + # 这里让它变成 5xx 响应，用例只关心"不再因权限被拒"。 + with TestClient(app, raise_server_exceptions=False) as client: + return client.request(method, path, json=body) + finally: + app.dependency_overrides.clear() + + +@pytest.mark.parametrize(("method", "path", "permission", "roles", "body"), CASES) +def test_endpoint_denies_request_without_required_permission( + monkeypatch: pytest.MonkeyPatch, + method: str, + path: str, + permission: str, + roles: tuple[str, ...], + body: dict[str, Any] | None, +) -> None: + """反向：权限集里没有该权限时，必须 403 且错误码是 `AGENT_PERMISSION_DENIED`。""" + response = _request( + monkeypatch, + method=method, + path=path, + permissions=(), + roles=roles, + body=body, + ) + + assert response.status_code == 403, ( + f"{method} {path} 缺少 {permission} 时预期 403，实际 {response.status_code}：" + f"{response.text[:200]}" + ) + assert _contains_code(response.json(), PERMISSION_DENIED_CODE), response.text + + +@pytest.mark.parametrize(("method", "path", "permission", "roles", "body"), CASES) +def test_endpoint_accepts_request_with_required_permission( + monkeypatch: pytest.MonkeyPatch, + method: str, + path: str, + permission: str, + roles: tuple[str, ...], + body: dict[str, Any] | None, +) -> None: + """正向：放入该权限后不能再是 403 —— 同时把端点要求的权限码钉住。""" + response = _request( + monkeypatch, + method=method, + path=path, + permissions=(permission,), + roles=roles, + body=body, + ) + + assert response.status_code != 403, ( + f"{method} {path} 带上 {permission} 后仍被拒；" + f"端点要求的权限码可能已改动：{response.text[:200]}" + ) + assert not _contains_code(response.json(), PERMISSION_DENIED_CODE), response.text + + +def test_cases_cover_every_phase2_permission_code() -> None: + """这组用例必须覆盖客服二期新增的三个权限码，缺一个就失去意义。""" + covered = {case[2] for case in CASES} + assert covered == { + "handover:read", + "memory:candidate:confirm", + "memory:candidate:review", + } diff --git a/tests/unit/tools/test_docs_endpoint_ids.py b/tests/unit/tools/test_docs_endpoint_ids.py new file mode 100644 index 0000000..850a4a7 --- /dev/null +++ b/tests/unit/tools/test_docs_endpoint_ids.py @@ -0,0 +1,109 @@ +"""`docs/05-接口文档.md` §19 端点编号唯一性（只读，不连库）。 + +规则与动机见 `tools/check_docs_endpoint_ids.py` 的模块说明：`check_authoritative_docs.py` +只校验 `docs/` 的**文件名编号**，不校验 §19 的**端点编号** —— 2026-09-12 两条线各自 +新增端点时都占了 `A034`/`A035`，合并后同时存在两个 `A034` 与两个 `A035` 却仍然通过守卫。 + +这里额外守住两件事： + +1. **不能"空扫通过"**：真实文档必须真的扫到足量编号，否则脚本坏掉了也会显示"无重复"； +2. **覆盖范围要可见**：扫描不得依赖前缀白名单 —— 同一件事上曾因正则写成 `[AMKCS]` + 而漏掉 `O`/`R` 两段，把 62 个端点报成 55 个；漏掉的号段一旦被复用，脚本仍会报 + "重复 0"。所以覆盖报告必须列出每个号段的数量。 +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from types import ModuleType + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parents[3] +#: 真实 §19 的规模下限：远小于实际数量，只为拦住"扫到 0 条也算通过"。 +MIN_REAL_ENDPOINTS = 40 + + +def _load_tool_module() -> ModuleType: + path = PROJECT_ROOT / "tools" / "check_docs_endpoint_ids.py" + spec = importlib.util.spec_from_file_location("check_docs_endpoint_ids", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _write_doc(tmp_path: Path, body: str) -> Path: + doc = tmp_path / "05-接口文档.md" + doc.write_text( + "# 接口文档\n\n## 19. 接口总目录\n\n" + "| 编号 | 方法与路径 | 权限 | 幂等 | 成功状态 | 审计 |\n" + "|---|---|---|---|---|---|\n" + f"{body}\n\n## 20. 变更流程\n\n占位。\n", + encoding="utf-8", + ) + return doc + + +def test_real_document_has_no_duplicate_endpoint_ids() -> None: + module = _load_tool_module() + assert module.collect_findings() == [] + + +def test_real_document_actually_scans_enough_endpoints() -> None: + """防止脚本坏掉后"什么都没扫到"却报通过。""" + module = _load_tool_module() + ids, _ = module.collect_rows(module.read_section()) + assert len(ids) >= MIN_REAL_ENDPOINTS + assert len(set(ids)) == len(ids) + + +def test_duplicate_endpoint_id_is_detected(tmp_path: Path) -> None: + module = _load_tool_module() + doc = _write_doc( + tmp_path, + "| A034 | `POST /api/v1/auth/tokens` | 公开 | 否 | `200` | 否 |\n" + "| A034 | `GET /api/v1/admin/customer-profile-candidates` | `x:y` | 否 | `200` | 否 |\n", + ) + findings = module.collect_findings(doc) + assert any("A034" in item and "重复" in item for item in findings) + + +def test_unrecognized_first_column_is_reported(tmp_path: Path) -> None: + module = _load_tool_module() + doc = _write_doc(tmp_path, "| A034（临时） | `GET /x` | `a:b` | 否 | `200` | 否 |\n") + findings = module.collect_findings(doc) + assert any("无法识别" in item for item in findings) + + +def test_coverage_report_lists_every_prefix_segment(tmp_path: Path) -> None: + """覆盖报告必须暴露各号段数量 —— 漏扫一个前缀就会在这里显形。""" + module = _load_tool_module() + doc = _write_doc( + tmp_path, + "| A001 | `GET /a1` | `a:b` | 否 | `200` | 否 |\n" + "| A002 | `GET /a2` | `a:b` | 否 | `200` | 否 |\n" + "| M003 | `GET /m3` | `a:b` | 否 | `200` | 否 |\n" + "| O001 | `GET /o1` | `a:b` | 否 | `200` | 否 |\n" + "| R001 | `GET /r1` | `a:b` | 否 | `200` | 否 |\n", + ) + report = module.describe_coverage(doc) + for expected in ("5 个端点编号", "4 个号段", "A×2", "M×1", "O×1", "R×1"): + assert expected in report, report + + +def test_missing_section_raises(tmp_path: Path) -> None: + module = _load_tool_module() + doc = tmp_path / "05-接口文档.md" + doc.write_text("# 接口文档\n\n## 1. 概述\n\n没有 §19。\n", encoding="utf-8") + with pytest.raises(module.DocumentStructureError): + module.collect_findings(doc) + + +def test_empty_checklist_prepares_no_false_success(tmp_path: Path) -> None: + """§19 存在但一张表都没有时，必须报结构异常，而不是"无重复"。""" + module = _load_tool_module() + doc = _write_doc(tmp_path, "本节没有表格。") + findings = module.collect_findings(doc) + assert any("没有扫描到任何端点编号" in item for item in findings) diff --git a/tools/check_docs_endpoint_ids.py b/tools/check_docs_endpoint_ids.py new file mode 100644 index 0000000..fc8b8aa --- /dev/null +++ b/tools/check_docs_endpoint_ids.py @@ -0,0 +1,132 @@ +"""校验 `docs/05-接口文档.md` §19「接口总目录」的端点编号唯一性（只读，不连数据库）。 + +## 为什么需要它 + +`tools/check_authoritative_docs.py` 只校验 `docs/` 的**文件名编号**与权威性声明， +**不校验 §19 里每个端点的编号**。2026-09-12 出过一次真实事故：两条线各自新增端点时 +都占用了 `A034`/`A035`，合并后 §19 同时存在两个 `A034` 与两个 `A035`， +而文档守卫照样通过 —— 编号复用比"编号不够"更麻烦，而且**静默遗留**。 + +## 口径（含覆盖自检） + +- 扫描范围：§19 章节内的**所有**表格行首列； +- **不枚举前缀白名单**，前缀从数据里归纳后打印出来。同一件事上还踩过一次 + "扫描正则写成 `[AMKCS]` 就漏掉 `O`/`R` 两段，把 62 个端点报成 55 个" —— + 一旦被漏掉的号段将来被复用，脚本仍会报"重复 0"。所以这里把 + **"扫到哪几个号段、各多少条、共计多少"** 一并输出，让覆盖范围本身可核对； +- 首列不是 `XXX###` 形状的行会被显式列出来（而不是静默跳过）； +- 只读：不修改任何文件。 + +用法： + + python tools/check_docs_endpoint_ids.py + +退出码：0 = 无重复；1 = 有重复或文档结构异常。 +""" + +from __future__ import annotations + +import re +from collections import Counter +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +INTERFACE_DOC = PROJECT_ROOT / "docs" / "05-接口文档.md" + +#: §19 的起止标题（`: \\s*` 之间允许一个空格，避免依赖标题里的具体文字）。 +SECTION_START = re.compile(r"^##\s*19\.") +SECTION_END = re.compile(r"^##\s*20\.") + +#: 表格行：只取首列，容忍单元格两侧空白。 +TABLE_ROW = re.compile(r"^\|(.+)\|\s*$") +#: Markdown 分隔行：| --- | :---: | 之类。 +SEPARATOR = re.compile(r"^[\s\-:|]+$") +#: 端点编号形状：1~4 个大写字母 + 3~4 位数字（A034 / M003 / R001）。 +ENDPOINT_ID = re.compile(r"^[A-Z]{1,4}\d{3,4}$") + + +class DocumentStructureError(RuntimeError): + """`docs/05` 的 §19 结构不符合预期，无法安全校验。""" + + +def read_section(path: Path = INTERFACE_DOC) -> list[str]: + """返回 §19 的正文行（不含起止标题）。""" + if not path.exists(): + raise DocumentStructureError(f"缺少接口权威文档：{path}") + + lines = path.read_text(encoding="utf-8").splitlines() + start = next((i for i, line in enumerate(lines) if SECTION_START.match(line)), None) + if start is None: + raise DocumentStructureError("接口文档里找不到 §19 章节标题") + end = next( + (i for i in range(start + 1, len(lines)) if SECTION_END.match(lines[i])), + len(lines), + ) + return lines[start + 1 : end] + + +def collect_rows(lines: list[str]) -> tuple[list[str], list[str]]: + """返回 `(首列编号列表, 无法识别的首列列表)`。 + + 表头行、分隔行、非表格行都会被跳过；非空但形状不对的首列会被归类为 + "无法识别"，以便显式暴露而不是静默漏扫。 + """ + ids: list[str] = [] + unrecognized: list[str] = [] + for line in lines: + match = TABLE_ROW.match(line.strip()) + if match is None: + continue + first_cell = match.group(1).split("|", 1)[0].strip() + if not first_cell or SEPARATOR.match(first_cell): + continue + if first_cell == "编号": # 表头 + continue + if ENDPOINT_ID.match(first_cell): + ids.append(first_cell) + else: + unrecognized.append(first_cell) + return ids, unrecognized + + +def collect_findings(path: Path = INTERFACE_DOC) -> list[str]: + """返回问题列表；空列表表示一致。""" + lines = read_section(path) + ids, unrecognized = collect_rows(lines) + + problems: list[str] = [] + if not ids: + problems.append("§19 没有扫描到任何端点编号，文档结构可能已变化，请人工确认") + + counter = Counter(ids) + for endpoint_id in sorted(code for code, count in counter.items() if count > 1): + problems.append(f"端点编号 {endpoint_id} 重复 {counter[endpoint_id]} 次") + + for cell in sorted(set(unrecognized)): + problems.append(f"§19 首列无法识别为端点编号：{cell!r}") + + return problems + + +def describe_coverage(path: Path = INTERFACE_DOC) -> str: + """返回覆盖范围报告 —— 让"扫到了什么"可见，而不是只报"没问题"。""" + ids, _ = collect_rows(read_section(path)) + prefixes = Counter(re.match(r"[A-Z]+", code).group() for code in ids) + detail = "、".join(f"{prefix}×{count}" for prefix, count in sorted(prefixes.items())) + return f"§19 覆盖：{len(ids)} 个端点编号 / {len(prefixes)} 个号段（{detail}）" + + +def main() -> int: + problems = collect_findings() + print(describe_coverage()) + if problems: + for problem in problems: + print(f" ✗ {problem}") + print(f"docs/05 §19 端点编号校验失败：{len(problems)} 处问题") + return 1 + print("docs/05 §19 端点编号无重复") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())