fix(security): close unauthenticated CMA user API surface + audit tool (cma-api-auth-surface-audit-20260914)

Root cause: main.py include_router had no global auth dependency; user routers
(products/cost/predict/...) relied on assert_entity_scope, which treats a
missing session token as the Bot service channel and lets the request through.
Result: 46 user endpoints returned data without any token on a public site.

Changes:
- auth_middleware.py: add require_auth_401 (missing Bearer -> 401, delegates to
  require_auth otherwise); 401/403 semantics are explicit.
- router-level Depends(require_auth_401) on products, cost, predict,
  notifications, permissions, tenant, alignment; endpoint-level on
  /auth/roles and /admin/cache/clear.
- deps.py: document that get_entity_id/assert_entity_scope are NOT auth.
- frontend: add the missing Bearer header to 5 tokenless fetch/axios calls that
  hit the newly protected endpoints (KPIAlignment x2, RolePermissions x2,
  Dashboard x1).
- tests: new test_api_auth_surface_20260914.py (58 cases: no-token 401,
  token 200, cross-entity 403, bot channel unchanged); adapt test_predict /
  test_kpi_forecast to send session tokens like production.
- scripts/audit_api_auth_surface.py: reusable FastAPI dependency-graph audit
  (423 endpoints; A bot / B gap / C public), exits 1 when a B-class gap exists.
This commit is contained in:
fullstack-bot
2026-09-14 16:41:15 +08:00
parent aac1bd1df1
commit b815b7f623
18 changed files with 604 additions and 51 deletions
+42 -6
View File
@@ -4,17 +4,52 @@
cvp / investment / sensitivity / scenario / cvp-detailed / cash-forecast /
cash-forecast/history / accuracy / scenario-suggestions /
scenario-suggestion/generate / real-option / growth-quality
鉴权(cma-api-auth-surface-audit-20260914):/api/cma/predict/* 已收口为「用户端点必须
带会话 Bearer」。本文件聚焦业务逻辑,`_predict_auth` autouse fixture 统一注入账套1的
测试 token 作为 client 默认头;需要其他账套的用例自行用 `_token_for_entity` 传 headers。
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import CashForecast, ForecastAccuracy
from app.models import CashForecast, ForecastAccuracy, Entity, User, UserEntity
from tests.conftest import create_test_user, get_token_for_user, auth_header
BASE = "/api/cma/predict"
def _token_for_entity(client: TestClient, db: Session, eid: int) -> str:
"""为指定账套签发测试 token(自动补实体 + 授权),供跨账套/自定义实体用例使用"""
ent = db.query(Entity).filter(Entity.id == eid).first()
if not ent:
db.add(Entity(id=eid, name=f"企业{eid}", short_name=f"E{eid}", status="active"))
db.commit()
user = db.query(User).filter(User.username == "testadmin").first()
if user is None:
user = create_test_user(db)
db.commit()
rel = db.query(UserEntity).filter(
UserEntity.user_id == user.id, UserEntity.entity_id == eid).first()
if not rel:
db.add(UserEntity(user_id=user.id, entity_id=eid))
db.commit()
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": eid})
assert resp.status_code == 200, f"登录 entity{eid} 失败: {resp.text[:200]}"
return resp.json()["token"]
@pytest.fixture(autouse=True)
def _predict_auth(client: TestClient, db: Session):
"""统一注入账套1的测试 token(显式传 headers= 的用例仍可覆盖,httpx 请求级优先)"""
if db.query(User).filter(User.username == "testadmin").first() is None:
create_test_user(db)
db.commit()
client.headers["Authorization"] = f"Bearer {get_token_for_user(client)}"
yield
class TestCvp:
def test_cvp_basic(self, client: TestClient, db: Session):
"""本量利分析"""
@@ -117,7 +152,6 @@ class TestCashForecast:
def test_cash_forecast_history_empty(self, client: TestClient, db: Session):
"""历史为空"""
create_test_user(db)
token = get_token_for_user(client)
resp = client.get(f"{BASE}/cash-forecast/history", headers=auth_header(token))
assert resp.status_code == 200
@@ -125,7 +159,6 @@ class TestCashForecast:
def test_accuracy_empty(self, client: TestClient, db: Session):
"""准确率为空"""
create_test_user(db)
token = get_token_for_user(client)
resp = client.get(f"{BASE}/accuracy", headers=auth_header(token))
assert resp.status_code == 200
@@ -219,20 +252,23 @@ class TestGrowthQuality:
assert "overall" in data or "level" in data or "score" in data
def test_growth_quality_entity2(self, client: TestClient, db: Session):
"""博海增长质量"""
resp = client.post(f"{BASE}/growth-quality", json={"entity_id": 2})
"""博海增长质量(需绑定 entity2 的会话 token,鉴权收口后不再靠 body entity_id)"""
token2 = _token_for_entity(client, db, 2)
resp = client.post(f"{BASE}/growth-quality", json={"entity_id": 2},
headers=auth_header(token2))
assert resp.status_code == 200
data = resp.json()
assert data["entity_name"] == "陕西博海网络科技"
def test_growth_quality_custom(self, client: TestClient, db: Session):
"""自定义企业数据(低分场景)"""
token99 = _token_for_entity(client, db, 99)
resp = client.post(f"{BASE}/growth-quality", json={
"entity_id": 99,
"name": "测试企业",
"rebateRate": 90, "trueGrossMargin": 5, "cashRatio": 3,
"expenseGrowthRate": 3, "revenueGrowthRate": 1, "mgmtRatio": 500,
})
}, headers=auth_header(token99))
assert resp.status_code == 200
data = resp.json()
assert data["entity_name"] == "测试企业"