"""CMA 全站 401/403 语义统一测试（cma-auth-401-403-unify-20260915） 口径（全站统一） ---------------- * **未认证**（缺 Authorization 头 / 非 Bearer 格式 / 无效 token）→ **401** * **已认证但无权**（跨实体、角色不足）→ **403**（语义不变） 改前：既有 router 的鉴权依赖是 ``HTTPBearer(auto_error=True)``，缺 header 时由 FastAPI 直接返回 403；这与本批收口的 46 个端点（``require_auth_401`` → 401） 在同站并存两种语义。本文件锁定统一后的姿态。 可见行为变更（有意为之，非回归） -------------------------------- 原「无 token → 403」的既有端点（kpis / maps / users / alerts / reports / org ... 共 384 个 AUTHED 端点）现在缺 token 一律 401。「已认证但无权」仍是 403： role_checker（角色不足）、deps.assert_entity_scope（跨实体）。 """ import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session from app.models import Entity, User from tests.conftest import ( create_test_user, auth_header, TEST_USER_PASSWORD, get_token_for_user, ) BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"} # 抽样：改前「无 token → 403」的既有 router 端点（端点级 require_auth 与 router 级 require_role 各覆盖） SAMPLE_PREV_403 = [ "/api/cma/kpis", # kpis.py 端点级 require_auth "/api/cma/maps", # maps.py "/api/cma/org/tree", # org.py "/api/cma/users", # users.py router 级 require_role("ceo","it") "/api/cma/alerts", # alerts.py "/api/cma/reports/profit-summary", # reports.py router 级 require_role "/api/cma/budget/plans", # budget.py "/api/cma/action-plans", # action_plans.py "/api/cma/expenses/rules", # expenses.py "/api/cma/auth/me", # auth.py ] # 带路径参数的抽样（模板用具体 id=1：鉴权依赖先于路径/参数校验执行） SAMPLE_PREV_403_WITH_PARAM = [ "/api/cma/kpis/1", "/api/cma/maps/1/versions", "/api/cma/thresholds/suggest/1", ] # 角色不足（已认证）→ 仍 403 的抽样 SAMPLE_ROLE_DENIED = [ ("business", "/api/cma/users"), # 需 ceo/it ("it", "/api/cma/reports/profit-summary"), # 报表仅 ceo/finance/business ] # 跨实体（token entity=1 + entity_id=2）→ 仍 403 SAMPLE_CROSS_ENTITY = [ "/api/cma/products/matrix", "/api/cma/cost/dashboard", "/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, username: str = "testadmin", password: str = TEST_USER_PASSWORD) -> str: resp = client.post("/api/cma/auth/login", json={ "username": username, "password": 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): """测试用户（ceo）+ entity1/entity2""" create_test_user(db) _ensure_entity(db, 1) _ensure_entity(db, 2) db.commit() return db def _add_role_user(db: Session, username: str, role: str) -> None: import hashlib db.add(User(username=username, password_hash=hashlib.sha256(TEST_USER_PASSWORD.encode()).hexdigest(), name=username, role=role)) db.commit() # ============================================================ # 1. 未认证 → 401（含改前 403 的既有端点抽样 + 全量 AUTHED GET 巡扫） # ============================================================ class TestUnauthenticatedIs401: @pytest.mark.parametrize("path", SAMPLE_PREV_403 + SAMPLE_PREV_403_WITH_PARAM) def test_prev_403_endpoints_now_401(self, client: TestClient, seeded, path: str): """改前 403 → 改后 401（本任务核心行为变更）""" resp = client.get(path) assert resp.status_code == 401, ( f"{path} 无 token 期望 401，实际 {resp.status_code}: {resp.text[:200]}" ) def test_route_sweep_no_token_all_401(self, client: TestClient, seeded): """全量巡扫：所有 AUTHED GET 端点（无路径参数）无 token 一律 401，不得再出现 403""" from scripts.audit_api_auth_surface import collect_routes, build_app rows = collect_routes(build_app()) paths = sorted(r["path"] for r in rows if r["class"] == "AUTHED" and "GET" in r["methods"] and "{" not in r["path"]) assert len(paths) >= 100, f"AUTHED GET 端点疑似减少: {len(paths)}" bad = [] for p in paths: resp = client.get(p) if resp.status_code != 401: bad.append(f"{p} → {resp.status_code}") assert bad == [], "无 token 非 401 的 AUTHED GET 端点：\n" + "\n".join(bad) def test_invalid_token_401(self, client: TestClient, seeded): """无效/伪造 token → 401""" for path in SAMPLE_PREV_403: resp = client.get(path, headers=auth_header("deadbeef-not-a-real-token")) assert resp.status_code == 401, f"{path} 期望 401，实际 {resp.status_code}" def test_non_bearer_scheme_401(self, client: TestClient, seeded): """非 Bearer 格式的 Authorization 头 = 未认证 → 401（改前 HTTPBearer 给 403）""" for path in ["/api/cma/kpis", "/api/cma/users"]: resp = client.get(path, headers={"Authorization": "Token abc"}) assert resp.status_code == 401, f"{path} 期望 401，实际 {resp.status_code}" # ============================================================ # 2. 已认证但无权 → 仍 403（语义保留，不得被本次统一吞掉） # ============================================================ class TestAuthenticatedForbiddenStill403: def test_cross_entity_403(self, client: TestClient, seeded): """token entity=1 + entity_id=2 → 403 + 文案同 get_entity_id""" token = _login(client, 1) for path in SAMPLE_CROSS_ENTITY: resp = client.get(path, params={"entity_id": 2}, headers=auth_header(token)) assert resp.status_code == 403, ( f"{path} 跨实体期望 403，实际 {resp.status_code}: {resp.text[:200]}" ) assert "无权访问企业" in resp.text @pytest.mark.parametrize("role,path", SAMPLE_ROLE_DENIED) def test_role_denied_403(self, client: TestClient, seeded, role: str, path: str): """已认证但角色不足 → 403（401 统一不得覆盖角色判定）""" uname = f"{role}_auth401" _add_role_user(seeded, uname, role) token = _login(client, 1, username=uname) resp = client.get(path, headers=auth_header(token)) assert resp.status_code == 403, ( f"{path} 角色 {role} 期望 403，实际 {resp.status_code}: {resp.text[:200]}" ) # ============================================================ # 3. 持 token 正常（功能不回归） # ============================================================ class TestAuthenticatedWorks: def test_ceo_token_200(self, client: TestClient, seeded): token = _login(client, 1) for path in ["/api/cma/kpis", "/api/cma/maps", "/api/cma/org/tree", "/api/cma/users", "/api/cma/alerts", "/api/cma/reports/profit-summary"]: resp = client.get(path, headers=auth_header(token)) assert resp.status_code == 200, ( f"{path} 持 token 期望 200，实际 {resp.status_code}: {resp.text[:200]}" ) # ============================================================ # 4. Bot 通道（X-BOT-KEY）不受影响 # ============================================================ class TestBotChannelUnchanged: def test_bot_key_200(self, client: TestClient, seeded): for path in ["/api/cma/bot/kpis", "/api/cma/bot/overview", "/api/cma/bot/alerts"]: 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_no_key_401(self, client: TestClient, seeded): for path in ["/api/cma/bot/kpis", "/api/cma/bot/overview"]: assert client.get(path).status_code == 401, f"{path} 无 key 期望 401" # ============================================================ # 5. C 类公开端点不受影响 # ============================================================ class TestPublicUnchanged: def test_public_endpoints_200(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