Hermes CI Fix
b796897f8c
require_auth now uses HTTPBearer(auto_error=False) and raises 401 explicitly when the Authorization header is missing or non-Bearer, matching require_auth_401. Unauthenticated requests are 401 for every user endpoint; authenticated-but-forbidden (cross-entity / role) stays 403. - app/auth_middleware.py: single optional bearer, 401 on missing token (require_auth_401 kept as a compatibility alias) - tests: update 4 pre-existing cases that asserted 403 without a token (test_auth, test_reports, test_kpi_causality, test_expenses, test_api_auth_surface_20260914) - intentional behavior change - add tests/test_auth_401_unification_20260915.py: sample + full sweep (all AUTHED GET routes -> 401), invalid token -> 401, cross-entity and role-denied still 403, bot channel (X-BOT-KEY) and public paths unchanged
207 lines
9.1 KiB
Python
207 lines
9.1 KiB
Python
"""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
|