fix(auth): unify site-wide 401/403 semantics (cma-auth-401-403-unify-20260915)
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
This commit is contained in:
Hermes CI Fix
@@ -176,10 +176,26 @@ def extract_bearer_token(request) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
# ── 统一全站鉴权语义（cma-auth-401-403-unify-20260915）────────────────────
|
||||
# 口径：**未认证（缺 token / 无效 token）→ 401**；**已认证但无权（跨实体 / 角色不足）→ 403**。
|
||||
# FastAPI 的 HTTPBearer(auto_error=True) 对「缺 Authorization 头」返回 403，与
|
||||
# require_auth_401 的 401 语义在同站并存（同一种「没带 token」两种状态码）。
|
||||
# 现统一为 auto_error=False 自行判空：缺 header / 非 Bearer 格式 → 显式 401；
|
||||
# 有 token → 仍按原逻辑校验身份（无效/过期 token → 401）。
|
||||
# 403 语义保留：require_role（角色不足）、deps.assert_entity_scope（跨实体无权）。
|
||||
_optional_bearer = HTTPBearer(auto_error=False)
|
||||
|
||||
|
||||
def require_auth(
|
||||
credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=True)),
|
||||
credentials: Optional[HTTPAuthorizationCredentials] = Depends(_optional_bearer),
|
||||
db: Session = Depends(get_db),
|
||||
) -> User:
|
||||
"""认证依赖（全站统一）：缺 token / 非 Bearer 格式 → 401；无效 token → 401。
|
||||
|
||||
已认证但无权（角色/实体）由 require_role、deps.assert_entity_scope 另行返回 403。
|
||||
"""
|
||||
if credentials is None or not credentials.credentials:
|
||||
raise HTTPException(status_code=401, detail="未认证：缺少 Bearer token")
|
||||
token = credentials.credentials
|
||||
user_id = _resolve_user_id(token)
|
||||
if user_id is None:
|
||||
@@ -200,20 +216,20 @@ def require_role(*roles: str):
|
||||
return role_checker
|
||||
|
||||
|
||||
# ── 401 语义的认证依赖（cma-api-auth-surface-audit-20260914）──────────────
|
||||
# FastAPI 的 HTTPBearer(auto_error=True) 对"缺 Authorization 头"返回 403；
|
||||
# 安全语义上「未认证」应为 401（401=未认证 / 403=无权）。
|
||||
# 本依赖用 auto_error=False 自行判空 → 缺 token 显式 401，有 token 仍完全委托 require_auth。
|
||||
_optional_bearer = HTTPBearer(auto_error=False)
|
||||
|
||||
|
||||
# ── 401 语义的认证依赖（cma-api-auth-surface-audit-20260914）
|
||||
# 2026-09-15（cma-auth-401-403-unify-20260915）：语义与 _optional_bearer 已统一，
|
||||
# 本入口保留作兼容别名（既有 router 与审计器 AUTH_STRICT 均引用此名）。
|
||||
def require_auth_401(
|
||||
credentials: Optional[HTTPAuthorizationCredentials] = Depends(_optional_bearer),
|
||||
db: Session = Depends(get_db),
|
||||
) -> User:
|
||||
"""用户端点统一鉴权入口：缺 Bearer → 401；有 token → require_auth 校验（无效 → 401）"""
|
||||
if credentials is None or not credentials.credentials:
|
||||
raise HTTPException(status_code=401, detail="未认证：缺少 Bearer token")
|
||||
"""用户端点统一鉴权入口：缺 Bearer → 401；有 token → require_auth 校验（无效 → 401）
|
||||
|
||||
2026-09-15（cma-auth-401-403-unify-20260915）起与 require_auth 语义完全一致
|
||||
（require_auth 也改为 auto_error=False 自判空 → 缺 token 401）。
|
||||
保留本名称：既有 router 的 dependencies=[Depends(require_auth_401)] 与
|
||||
backend/scripts/audit_api_auth_surface.py 的 AUTH_STRICT 分类仍引用它。
|
||||
"""
|
||||
return require_auth(credentials=credentials, db=db)
|
||||
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
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（对照，不回归）
|
||||
4. 既有 require_auth 端点（kpis/maps/org-tree）无 token → **401**（2026-09-15 语义统一：原 403）
|
||||
5. 持 token 正常 200；跨实体（token entity=1 + entity_id=2）仍 403
|
||||
6. 结构性不变量：全量路由巡检无 B 类缺口（新增用户端点漏挂鉴权 → 本用例 FAIL）
|
||||
"""
|
||||
@@ -221,11 +221,12 @@ class TestPublicAndExistingBaseline:
|
||||
"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）不回归"""
|
||||
def test_existing_403_endpoints_now_401(self, client: TestClient, seeded):
|
||||
"""语义统一（cma-auth-401-403-unify-20260915）：原 403 端点（kpis/maps/org-tree）
|
||||
无 token 现返回 401（未认证语义）；持 token 仍 200"""
|
||||
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).status_code == 401, f"{path} 无 token 期望 401"
|
||||
assert client.get(path, headers=auth_header(token)).status_code == 200
|
||||
|
||||
|
||||
|
||||
@@ -61,7 +61,7 @@ class TestAuth:
|
||||
def test_me_without_token(self, client: TestClient):
|
||||
"""无token访问需要认证的接口"""
|
||||
resp = client.get("/api/cma/auth/me")
|
||||
assert resp.status_code == 403 # HTTPBearer auto_error
|
||||
assert resp.status_code == 401 # 语义统一(2026-09-15)：未认证=401（原 HTTPBearer auto_error 给 403）
|
||||
|
||||
def test_register(self, client: TestClient, db: Session):
|
||||
"""注册新用户"""
|
||||
|
||||
@@ -0,0 +1,206 @@
|
||||
"""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
|
||||
@@ -482,4 +482,5 @@ class TestStats:
|
||||
|
||||
def test_stats_no_token(self, client: TestClient, db: Session):
|
||||
resp = client.get(self.BASE)
|
||||
assert resp.status_code == 403
|
||||
# 语义统一(2026-09-15)：未认证=401（原 HTTPBearer auto_error 给 403）
|
||||
assert resp.status_code == 401
|
||||
|
||||
@@ -348,6 +348,6 @@ class TestPermissions:
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_no_token_denied(self, client: TestClient):
|
||||
"""无token → 403（router级 require_role 直接拒绝，与 test_reports 约定一致）"""
|
||||
"""无token → 401（语义统一 2026-09-15：未认证=401；router级 require_role 仍负责角色 403）"""
|
||||
resp = client.get(BASE)
|
||||
assert resp.status_code == 403
|
||||
assert resp.status_code == 401
|
||||
|
||||
@@ -878,6 +878,6 @@ class TestReportsPermissions:
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_no_token_denied(self, client: TestClient, db: Session):
|
||||
"""无token → 403（router级 require_role 直接拒绝）"""
|
||||
"""无token → 401（语义统一 2026-09-15：未认证=401；原 router级 require_role 经 HTTPBearer 给 403）"""
|
||||
resp = client.get(f"{self.BASE}/profit-summary")
|
||||
assert resp.status_code == 403
|
||||
assert resp.status_code == 401
|
||||
|
||||
Reference in New Issue
Block a user