"""CMA API 鉴权面收口测试（cma-api-auth-surface-audit-20260914） 背景：项目Bot 实测发现部分用户端点无 token 亦可取数（products/matrix、cost/*、predict/* 等）。根因：这些 router 未挂任何鉴权依赖，端点仅用 assert_entity_scope（它**不是鉴权**， 无 token 时按「Bot 服务通道」放行）。 本文件锁定收口后的鉴权姿态： 1. B 类（原缺鉴权用户端点）无 token → **401**（46 个端点逐条断言） 2. C 类（显式公开端点）无 token 仍按原语义可用 3. A 类（X-BOT-KEY Bot 通道）行为不变：带 key 200 / 无 key 401 4. 既有 require_auth 端点（kpis/maps）无 token → 403（对照，不回归） 5. 持 token 正常 200；跨实体（token entity=1 + entity_id=2）仍 403 6. 结构性不变量：全量路由巡检无 B 类缺口（新增用户端点漏挂鉴权 → 本用例 FAIL） """ import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session from app.models import Entity, UserEntity from tests.conftest import create_test_user, auth_header, TEST_USER_PASSWORD BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"} # ── B 类收口清单：原「无 token 亦可访问」的用户端点（46 个）────────────── # 无 token 必须 401。模板路径用具体 id=1（鉴权依赖先于路径/body 校验执行）。 B_CLASS_ENDPOINTS = [ ("POST", "/api/cma/admin/cache/clear"), ("GET", "/api/cma/alignment/config"), ("GET", "/api/cma/alignment/modes"), ("GET", "/api/cma/auth/roles"), ("GET", "/api/cma/cost/abc/activities"), ("POST", "/api/cma/cost/abc/activities"), ("POST", "/api/cma/cost/abc/allocate"), ("GET", "/api/cma/cost/abc/allocations"), ("GET", "/api/cma/cost/actual-costs"), ("POST", "/api/cma/cost/actual-costs"), ("GET", "/api/cma/cost/breakdown"), ("GET", "/api/cma/cost/comparison"), ("GET", "/api/cma/cost/dashboard"), ("GET", "/api/cma/cost/overview"), ("GET", "/api/cma/cost/standard-costs"), ("POST", "/api/cma/cost/standard-costs"), ("PUT", "/api/cma/cost/standard-costs/1"), ("DELETE", "/api/cma/cost/standard-costs/1"), ("GET", "/api/cma/cost/variance"), ("GET", "/api/cma/notifications/channels"), ("POST", "/api/cma/notifications/channels"), ("PUT", "/api/cma/notifications/channels/1"), ("DELETE", "/api/cma/notifications/channels/1"), ("POST", "/api/cma/notifications/channels/1/test"), ("GET", "/api/cma/notifications/logs"), ("GET", "/api/cma/permissions/config"), ("GET", "/api/cma/permissions/modules"), ("GET", "/api/cma/predict/accuracy"), ("POST", "/api/cma/predict/cash-forecast"), ("GET", "/api/cma/predict/cash-forecast/history"), ("POST", "/api/cma/predict/cvp"), ("POST", "/api/cma/predict/cvp-detailed"), ("POST", "/api/cma/predict/growth-quality"), ("POST", "/api/cma/predict/investment"), ("GET", "/api/cma/predict/kpi-forecast"), ("GET", "/api/cma/predict/kpi-forecast/finance"), ("GET", "/api/cma/predict/kpi-forecast/sensitivity"), ("POST", "/api/cma/predict/real-option"), ("POST", "/api/cma/predict/relevant-decision"), ("POST", "/api/cma/predict/scenario"), ("POST", "/api/cma/predict/scenario-suggestion/generate"), ("GET", "/api/cma/predict/scenario-suggestions"), ("POST", "/api/cma/predict/sensitivity"), ("GET", "/api/cma/products/matrix"), ("GET", "/api/cma/tenant/current"), ("POST", "/api/cma/tenant/switch"), ] # 种子端点（项目Bot 实测无 token 200 → 收口后必须 401；持 token 必须 200） SEED_NO_TOKEN_401 = [ "/api/cma/products/matrix", "/api/cma/cost/dashboard", "/api/cma/cost/overview", "/api/cma/predict/accuracy", ] def _ensure_entity(db: Session, eid: int) -> Entity: ent = db.query(Entity).filter(Entity.id == eid).first() if not ent: ent = Entity(id=eid, name=f"企业{eid}", short_name=f"E{eid}", status="active") db.add(ent) db.commit() db.refresh(ent) return ent def _login(client: TestClient, entity_id: int = 1) -> str: resp = client.post("/api/cma/auth/login", json={ "username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": entity_id, }) assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}" return resp.json().get("token") @pytest.fixture def seeded(db: Session): """测试用户 + entity1/entity2（授权走登录时的默认授予）""" create_test_user(db) _ensure_entity(db, 1) _ensure_entity(db, 2) db.commit() return db # ============================================================ # 1. B 类：无 token → 401（逐端点） # ============================================================ class TestBClassRequiresAuth: @pytest.mark.parametrize("method,path", B_CLASS_ENDPOINTS, ids=[f"{m} {p}" for m, p in B_CLASS_ENDPOINTS]) def test_no_token_401(self, client: TestClient, method: str, path: str): resp = client.request(method, path) assert resp.status_code == 401, ( f"{method} {path} 无 token 期望 401，实际 {resp.status_code}: {resp.text[:200]}" ) def test_seed_endpoints_401(self, client: TestClient): """验收种子项：4 个原「无 token 200」端点收口后 401""" for path in SEED_NO_TOKEN_401: resp = client.get(path) assert resp.status_code == 401, f"{path} 期望 401，实际 {resp.status_code}" def test_invalid_token_401(self, client: TestClient): """伪造/失效 token 亦 401（不因鉴权收口而放宽）""" for path in SEED_NO_TOKEN_401: resp = client.get(path, headers=auth_header("deadbeef-not-a-real-token")) assert resp.status_code == 401, f"{path} 期望 401，实际 {resp.status_code}" # ============================================================ # 2. B 类：持 token → 200（功能不回归） # ============================================================ class TestBClassWithTokenWorks: def test_seed_endpoints_200_with_token(self, client: TestClient, seeded): token = _login(client, 1) for path in SEED_NO_TOKEN_401: resp = client.get(path, headers=auth_header(token)) assert resp.status_code == 200, ( f"{path} 持 token 期望 200，实际 {resp.status_code}: {resp.text[:200]}" ) def test_previously_authed_endpoints_still_work(self, client: TestClient, seeded): """原已带鉴权的端点（alignment/tree、permissions/config PUT 之外）持 token 正常""" token = _login(client, 1) for path in ["/api/cma/alignment/config", "/api/cma/alignment/modes", "/api/cma/permissions/modules", "/api/cma/permissions/config", "/api/cma/notifications/channels", "/api/cma/notifications/logs", "/api/cma/tenant/current", "/api/cma/cost/standard-costs", "/api/cma/cost/abc/activities", "/api/cma/cost/abc/allocations"]: resp = client.get(path, headers=auth_header(token)) assert resp.status_code == 200, ( f"{path} 持 token 期望 200，实际 {resp.status_code}: {resp.text[:200]}" ) # ============================================================ # 3. 跨实体不回归：token entity=1 + entity_id=2 → 403（文案同 get_entity_id） # ============================================================ class TestCrossEntityStillForbidden: def test_cross_entity_403(self, client: TestClient, seeded): token = _login(client, 1) checks = [ ("/api/cma/products/matrix", {"entity_id": 2}), ("/api/cma/cost/dashboard", {"entity_id": 2}), ("/api/cma/cost/overview", {"entity_id": 2}), ("/api/cma/predict/accuracy", {"entity_id": 2}), ] for path, params in checks: resp = client.get(path, params=params, headers=auth_header(token)) assert resp.status_code == 403, ( f"{path} 跨实体期望 403，实际 {resp.status_code}: {resp.text[:200]}" ) assert "无权访问企业" in resp.text def test_same_entity_200(self, client: TestClient, seeded): token = _login(client, 1) for path in ["/api/cma/products/matrix", "/api/cma/cost/dashboard"]: resp = client.get(path, params={"entity_id": 1}, headers=auth_header(token)) assert resp.status_code == 200, f"{path} 同实体期望 200，实际 {resp.status_code}" # ============================================================ # 4. A 类 Bot 通道行为不变 # ============================================================ class TestBotChannelUnchanged: def test_bot_endpoints_with_key_200(self, client: TestClient, seeded): for path in ["/api/cma/bot/kpis", "/api/cma/bot/overview", "/api/cma/bot/cost/actual", "/api/cma/bot/cost/standard", "/api/cma/bot/alerts", "/api/cma/bot/strategic-maps"]: resp = client.get(path, headers={**BOT_KEY, "X-Entity-Id": "1"}) assert resp.status_code == 200, ( f"{path} 带 X-BOT-KEY 期望 200，实际 {resp.status_code}: {resp.text[:200]}" ) def test_bot_endpoints_without_key_401(self, client: TestClient, seeded): for path in ["/api/cma/bot/kpis", "/api/cma/bot/overview"]: resp = client.get(path) assert resp.status_code == 401, ( f"{path} 无 X-BOT-KEY 期望 401，实际 {resp.status_code}" ) # ============================================================ # 5. C 类公开端点 & 既有 403 对照 # ============================================================ class TestPublicAndExistingBaseline: def test_public_endpoints_unchanged(self, client: TestClient, seeded): assert client.get("/health").status_code == 200 assert client.get("/api/cma/bot/ping").status_code == 200 assert client.get("/api/cma/security/captcha/request").status_code == 200 assert client.get("/api/cma/auth/login-entities").status_code == 200 # 登录端点公开但密码错仍 401 r = client.post("/api/cma/auth/login", json={ "username": "testadmin", "password": "wrong", "entity_id": 1}) assert r.status_code == 401 def test_existing_403_endpoints_unchanged(self, client: TestClient, seeded): """对照：原 403 端点（kpis/maps/org-tree）不回归""" token = _login(client, 1) for path in ["/api/cma/kpis", "/api/cma/maps", "/api/cma/org/tree"]: assert client.get(path).status_code == 403, f"{path} 无 token 期望仍 403" assert client.get(path, headers=auth_header(token)).status_code == 200 # ============================================================ # 6. 结构性不变量：全量路由巡检无 B 类缺口 # ============================================================ class TestAuthSurfaceInvariant: def test_no_gap_in_route_table(self): """任何新用户端点漏挂 require_auth → 本用例 FAIL（防缺口回归）""" from scripts.audit_api_auth_surface import collect_routes, build_app rows = collect_routes(build_app()) gaps = [f'{r["methods"]} {r["path"]} ({r["module"]})' for r in rows if r["class"] == "B"] assert gaps == [], "发现无鉴权用户端点（B 类缺口）：\n" + "\n".join(gaps) def test_bot_and_auth_counts(self): """A 类 Bot 通道 / C 类公开 / 已鉴权 端点计数快照（防误删鉴权或误加公开）""" from scripts.audit_api_auth_surface import collect_routes, build_app rows = collect_routes(build_app()) by = {} for r in rows: by[r["class"]] = by.get(r["class"], 0) + 1 assert by.get("A", 0) >= 30, f"Bot 通道端点疑似减少: {by}" assert by.get("C", 0) == 9, f"公开端点集被改动（应为 9）: {by}" assert by.get("AUTHED", 0) >= 384, f"已鉴权端点疑似减少: {by}"