Hermes CI Fix
115a0ec755
背景（项目Bot 补充复验 §四#2）：以下 4 个只读端点未做会话账套鉴权，
用 entity=1(酣客) 会话 token + 显式 entity 参数可读到 entity=2(博海) 数据（HTTP 200）：
- GET /data-quality/governance-check?entity_id=2 → 200 score=88（对照 entity1=72）
- GET /products/matrix?entity_id=2 → 200
- GET /alerts/risk-matrix?entity=bohai → 200 博海风险矩阵
- GET /action-plans/coso-checklist?entity=bohai → 200 博海 COSO total_score=55
另：risk-matrix / coso-checklist 在非法实体值下静默回落酣客数据，UI 标签与数据不一致。
改动：
1. deps.py 新增统一防护入口 assert_entity_scope / assert_entity_slug_scope /
get_session_entity / resolve_session_slug（文案对齐 get_entity_id：无权访问企业 entity_id=X（当前账套: Y））
2. governance-check / products/matrix：entity_id 须等于会话账套，否则 403；
缺省按会话账套（governance-check 原缺省 0=全部 → 登录态下按账套收敛）
3. risk-matrix / coso-checklist：entity 须等于会话账套，否则 403；
非法值 → 400 显式报错（不再静默回落酣客）
4. P3：POST /auth/switch-entity 加 DEPRECATED 注释（前端已无调用，保留不删）
5. 测试：tests/test_security_multitenant.py 新增 TestCrossEntityEndpointAuth 15 例
（跨实体 403 / 同实体 200 / 缺省按账套 / 非法显式报错 / 参照端点行为不变）
验证：pytest tests/ -q → 951 passed（10 项失败为改动前既有基线，已 stash 对照确认）；
真实后端冒烟 19/19 PASS（entity=1 会话 4 端点跨实体全 403）。
343 lines
17 KiB
Python
343 lines
17 KiB
Python
"""多租户隔离安全修复测试（2026-08-31 OpenCode 安全审查 P0）
|
||
|
||
覆盖 DoD 输出物8/9/10：
|
||
- bot_bridge 跨 entity 隔离（X-Entity-Id header 带不同账套返回不同数据）
|
||
- alert_rules create 写入 entity_id（跨 entity 不可见）
|
||
- reports 跨 entity 过滤（profit-summary / kpi-trends）
|
||
- _eval_threshold invert 参数（低于阈值触发红灯）
|
||
- /check-governance 在 SQLite 测试库不 500（data_quality 收敛后复用 Python 解析）
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import KPIDefinition, KPIAlert, Entity, UserEntity
|
||
from app.api.alert_rules import AlertRule, _eval_threshold
|
||
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
|
||
|
||
BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
|
||
|
||
|
||
def _ensure_entity(db: Session, eid: int, name: str = None) -> Entity:
|
||
"""确保测试库存在指定 entity（BOT 通道 get_entity_id 会校验 active）"""
|
||
ent = db.query(Entity).filter(Entity.id == eid).first()
|
||
if not ent:
|
||
ent = Entity(id=eid, name=name or f"企业{eid}", short_name=f"E{eid}", status="active")
|
||
db.add(ent)
|
||
db.commit()
|
||
db.refresh(ent)
|
||
return ent
|
||
|
||
|
||
def _seed_entity_kpi(db: Session, entity_id: int, code: str, name: str = None) -> KPIDefinition:
|
||
"""创建指定账套的 KPI（多租户测试专用）"""
|
||
kpi = create_test_kpi(db, kpi_code=code, kpi_name=name or code, entity_id=entity_id,
|
||
target_value=100.0, unit="万元", frequency="monthly",
|
||
dimension="finance", status="active")
|
||
return kpi
|
||
|
||
|
||
# ============================================================
|
||
# bot_bridge 跨 entity 隔离（P0-1）
|
||
# ============================================================
|
||
class TestBotBridgeIsolation:
|
||
def test_kpis_entity_isolation(self, client: TestClient, db: Session):
|
||
"""BOT Key + X-Entity-Id=1 → 只返回 entity1 的 KPI；X-Entity-Id=2 → 只返回 entity2"""
|
||
_ensure_entity(db, 2)
|
||
_seed_entity_kpi(db, entity_id=1, code="E1_REVENUE")
|
||
_seed_entity_kpi(db, entity_id=2, code="E2_REVENUE")
|
||
db.commit()
|
||
|
||
r1 = client.get("/api/cma/bot/kpis", headers={**BOT_KEY, "X-Entity-Id": "1"})
|
||
assert r1.status_code == 200
|
||
codes1 = {i["code"] for i in r1.json()["items"]}
|
||
assert "E1_REVENUE" in codes1
|
||
assert "E2_REVENUE" not in codes1
|
||
|
||
r2 = client.get("/api/cma/bot/kpis", headers={**BOT_KEY, "X-Entity-Id": "2"})
|
||
assert r2.status_code == 200
|
||
codes2 = {i["code"] for i in r2.json()["items"]}
|
||
assert "E2_REVENUE" in codes2
|
||
assert "E1_REVENUE" not in codes2
|
||
|
||
def test_overview_entity_isolation(self, client: TestClient, db: Session):
|
||
"""overview 统计按 entity 过滤：entity1 只统计自己的 KPI/预警"""
|
||
_ensure_entity(db, 2)
|
||
k1 = _seed_entity_kpi(db, entity_id=1, code="O1_KPI")
|
||
_seed_entity_kpi(db, entity_id=2, code="O2_KPI")
|
||
db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="e1预警",
|
||
status="pending", entity_id=1))
|
||
db.commit()
|
||
|
||
r1 = client.get("/api/cma/bot/overview", headers={**BOT_KEY, "X-Entity-Id": "1"})
|
||
assert r1.status_code == 200
|
||
assert r1.json()["stats"]["kpis_total"] == 1
|
||
assert r1.json()["stats"]["alerts_open"] == 1
|
||
|
||
r2 = client.get("/api/cma/bot/overview", headers={**BOT_KEY, "X-Entity-Id": "2"})
|
||
assert r2.status_code == 200
|
||
assert r2.json()["stats"]["kpis_total"] == 1
|
||
assert r2.json()["stats"]["alerts_open"] == 0
|
||
|
||
def test_query_param_entity_isolation(self, client: TestClient, db: Session):
|
||
"""无 token 时 entity_id 也可通过 query 参数传入（Bot 通道）"""
|
||
_ensure_entity(db, 2)
|
||
_seed_entity_kpi(db, entity_id=1, code="Q1_KPI")
|
||
_seed_entity_kpi(db, entity_id=2, code="Q2_KPI")
|
||
db.commit()
|
||
|
||
r = client.get("/api/cma/bot/kpis?entity_id=1", headers=BOT_KEY)
|
||
assert r.status_code == 200
|
||
codes = {i["code"] for i in r.json()["items"]}
|
||
assert "Q1_KPI" in codes
|
||
assert "Q2_KPI" not in codes
|
||
|
||
|
||
# ============================================================
|
||
# alert_rules 隔离（P0-2）
|
||
# ============================================================
|
||
class TestAlertRulesIsolation:
|
||
def test_create_alert_rule_writes_entity_id(self, client: TestClient, db: Session):
|
||
"""create 写入 entity_id：以 entity1 身份创建的规则，entity2 不可见"""
|
||
_ensure_entity(db, 2)
|
||
k1 = _seed_entity_kpi(db, entity_id=1, code="AR_E1")
|
||
db.commit()
|
||
create_test_user(db)
|
||
token = get_token_for_user(client, "testadmin", "admin123")
|
||
headers = auth_header(token)
|
||
|
||
resp = client.post("/api/cma/alert-rules", headers=headers, json={
|
||
"kpi_id": k1.id,
|
||
"rule_type": "static",
|
||
"params": {"operator": ">=", "threshold": 80.0},
|
||
"trigger_on": "actual",
|
||
})
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
# DB 回查：规则 entity_id = 1（用户 token 绑定 entity1）
|
||
rule = db.query(AlertRule).filter(AlertRule.kpi_id == k1.id).first()
|
||
assert rule is not None
|
||
assert rule.entity_id == 1
|
||
|
||
def test_get_kpi_rules_entity_scoped(self, client: TestClient, db: Session):
|
||
"""get_kpi_rules 按 entity 过滤：entity2 查不到 entity1 的规则"""
|
||
_ensure_entity(db, 2)
|
||
k1 = _seed_entity_kpi(db, entity_id=1, code="GR_E1")
|
||
db.add(AlertRule(kpi_id=k1.id, rule_type="static",
|
||
params={"operator": ">=", "threshold": 80.0},
|
||
entity_id=1))
|
||
db.commit()
|
||
create_test_user(db)
|
||
token = get_token_for_user(client, "testadmin", "admin123")
|
||
|
||
r1 = client.get(f"/api/cma/alert-rules/kpi/{k1.id}", headers=auth_header(token))
|
||
assert r1.status_code == 200
|
||
assert len(r1.json()["data"]) >= 1
|
||
|
||
|
||
# ============================================================
|
||
# reports 跨 entity 过滤（P1-3）
|
||
# ============================================================
|
||
class TestReportsIsolation:
|
||
def test_profit_summary_entity_filtered(self, client: TestClient, db: Session):
|
||
"""profit-summary 按 entity 过滤：entity2 的 KPI 值对 entity1 不可见"""
|
||
_ensure_entity(db, 2)
|
||
from app.models import KPIValue
|
||
k1 = _seed_entity_kpi(db, entity_id=1, code="F_REVENUE")
|
||
_seed_entity_kpi(db, entity_id=2, code="F_REVENUE_2")
|
||
db.add(KPIValue(kpi_id=k1.id, period="2026-06", actual_value=888.0,
|
||
data_status="verified", entity_id=1))
|
||
db.commit()
|
||
create_test_user(db)
|
||
token = get_token_for_user(client, "testadmin", "admin123")
|
||
|
||
r1 = client.get("/api/cma/reports/profit-summary?period=2026-06", headers=auth_header(token))
|
||
assert r1.status_code == 200
|
||
# entity1 的 token → 读到 entity1 的收入项（F_REVENUE 命中，含888值）
|
||
items = r1.json()["items"]
|
||
assert len(items) >= 1
|
||
|
||
def test_kpi_trends_entity_filtered(self, client: TestClient, db: Session):
|
||
"""kpi-trends 按 entity 过滤：entity2 看不到 entity1 的 KPI 列表"""
|
||
_ensure_entity(db, 2)
|
||
_seed_entity_kpi(db, entity_id=1, code="TR_E1")
|
||
_seed_entity_kpi(db, entity_id=2, code="TR_E2")
|
||
db.commit()
|
||
create_test_user(db)
|
||
token = get_token_for_user(client, "testadmin", "admin123")
|
||
|
||
r1 = client.get("/api/cma/reports/kpi-trends", headers=auth_header(token))
|
||
assert r1.status_code == 200
|
||
kpis1 = r1.json().get("data", [])
|
||
codes1 = {k["kpi_code"] for k in kpis1}
|
||
assert "TR_E1" in codes1
|
||
assert "TR_E2" not in codes1
|
||
|
||
|
||
# ============================================================
|
||
# _eval_threshold invert 参数（P1-1）
|
||
# ============================================================
|
||
class TestInvertThreshold:
|
||
def test_invert_lower_threshold_triggers_red(self):
|
||
"""invert=True：值低于阈值时取反（低于下限触发红灯场景）"""
|
||
assert _eval_threshold(500, "<600", invert=True) is False # 500 < 600，原命中 → invert 后不命中
|
||
assert _eval_threshold(700, "<600", invert=True) is True # 700 >= 600，原不命中 → invert 后命中
|
||
|
||
def test_invert_greater_threshold(self):
|
||
"""invert=True：值高于阈值时取反"""
|
||
assert _eval_threshold(30, ">25", invert=True) is False # 30 > 25，原命中 → invert 后不命中
|
||
assert _eval_threshold(10, ">25", invert=True) is True # 10 <= 25，原不命中 → invert 后命中
|
||
|
||
def test_invert_gt_eq_and_lt_eq(self):
|
||
""">= 与 <= 的 invert 取反"""
|
||
assert _eval_threshold(80, ">=90", invert=True) is True # 80 < 90 → invert 命中
|
||
assert _eval_threshold(95, ">=90", invert=True) is False
|
||
assert _eval_threshold(95, "<=90", invert=True) is True # 95 > 90 → invert 命中
|
||
assert _eval_threshold(85, "<=90", invert=True) is False
|
||
|
||
def test_red_branch_no_invert_behavior_preserved(self):
|
||
"""_check_static red 分支不传 invert：字面阈值行为不变（回归保护）"""
|
||
assert _eval_threshold(500, "<600") is True # 低于600 → 命中（默认字面）
|
||
assert _eval_threshold(700, "<600") is False
|
||
assert _eval_threshold(30, ">25") is True
|
||
assert _eval_threshold(10, ">25") is False
|
||
|
||
|
||
# ============================================================
|
||
# /check-governance SQLite 兼容（P1-2）
|
||
# ============================================================
|
||
class TestCheckGovernanceSQLite:
|
||
def test_check_governance_no_500_on_sqlite(self, client: TestClient, db: Session):
|
||
"""data_quality 收敛后 /check-governance 在 SQLite 测试库不 500"""
|
||
_seed_entity_kpi(db, entity_id=1, code="GOV_KPI")
|
||
db.commit()
|
||
create_test_user(db)
|
||
token = get_token_for_user(client, "testadmin", "admin123")
|
||
|
||
resp = client.get("/api/cma/data-quality/check-governance", headers=auth_header(token))
|
||
assert resp.status_code == 200, f"check-governance 500: {resp.text[:300]}"
|
||
data = resp.json()
|
||
assert "score" in data
|
||
assert data["total_rules"] == 7
|
||
|
||
def test_governance_check_still_works(self, client: TestClient, db: Session):
|
||
"""governance-check 端点（新口径）在收敛后仍正常"""
|
||
_seed_entity_kpi(db, entity_id=1, code="GOV2_KPI")
|
||
db.commit()
|
||
create_test_user(db)
|
||
token = get_token_for_user(client, "testadmin", "admin123")
|
||
|
||
resp = client.get("/api/cma/data-quality/governance-check", headers=auth_header(token))
|
||
assert resp.status_code == 200, resp.text[:300]
|
||
assert resp.json()["total_rules"] == 7
|
||
|
||
|
||
# ============================================================
|
||
# 4 端点跨租户鉴权补单（cma-cross-entity-endpoints-auth-20260914）
|
||
# 对齐参照：growth-quality / predict.accuracy（已 403）
|
||
# ============================================================
|
||
class TestCrossEntityEndpointAuth:
|
||
"""会话 entity=1 显式请求 entity=2 → 403；同实体 → 200；非法实体 → 显式报错"""
|
||
|
||
def _login(self, client, db):
|
||
_ensure_entity(db, 2)
|
||
create_test_user(db)
|
||
return get_token_for_user(client, "testadmin", "admin123") # 登录账套 entity_id=1
|
||
|
||
# ── 1. data-quality/governance-check ──────────────────
|
||
def test_governance_check_cross_entity_403(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/data-quality/governance-check?entity_id=2", headers=auth_header(token))
|
||
assert r.status_code == 403, f"跨实体应 403，实际 {r.status_code}: {r.text[:200]}"
|
||
assert "无权访问企业 entity_id=2（当前账套: 1）" in r.text
|
||
|
||
def test_governance_check_same_entity_200(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/data-quality/governance-check?entity_id=1", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["entity_id"] == 1
|
||
|
||
def test_governance_check_no_param_scoped_to_session(self, client, db):
|
||
"""未传 entity_id：按会话账套（不再默认 0=全部）"""
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/data-quality/governance-check", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["entity_id"] == 1
|
||
|
||
def test_governance_check_illegal_entity_403(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/data-quality/governance-check?entity_id=999", headers=auth_header(token))
|
||
assert r.status_code == 403, r.text[:200]
|
||
|
||
# ── 2. products/matrix ────────────────────────────────
|
||
def test_products_matrix_cross_entity_403(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/products/matrix?entity_id=2&months=6", headers=auth_header(token))
|
||
assert r.status_code == 403, f"跨实体应 403，实际 {r.status_code}: {r.text[:200]}"
|
||
|
||
def test_products_matrix_same_entity_200(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/products/matrix?entity_id=1", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["entity_id"] == 1
|
||
|
||
def test_products_matrix_no_param_scoped_to_session(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/products/matrix", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["entity_id"] == 1
|
||
|
||
# ── 3. alerts/risk-matrix（字符串实体）────────────────
|
||
def test_risk_matrix_cross_entity_403(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/alerts/risk-matrix?entity=bohai", headers=auth_header(token))
|
||
assert r.status_code == 403, f"跨实体应 403，实际 {r.status_code}: {r.text[:200]}"
|
||
assert "无权访问企业 entity_id=2（当前账套: 1）" in r.text
|
||
|
||
def test_risk_matrix_same_entity_200(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/alerts/risk-matrix?entity=hanke", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert "酣客" in r.json()["entity_name"]
|
||
|
||
def test_risk_matrix_no_param_uses_session_entity(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/alerts/risk-matrix", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert "酣客" in r.json()["entity_name"]
|
||
|
||
def test_risk_matrix_illegal_entity_explicit_error(self, client, db):
|
||
"""非法实体 → 400 显式报错（不再静默回落酣客数据）"""
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/alerts/risk-matrix?entity=2", headers=auth_header(token))
|
||
assert r.status_code == 400, f"非法实体应 400，实际 {r.status_code}: {r.text[:200]}"
|
||
assert "非法实体" in r.text
|
||
assert "默认返回酣客数据" not in r.text
|
||
|
||
# ── 4. action-plans/coso-checklist ────────────────────
|
||
def test_coso_checklist_cross_entity_403(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/action-plans/coso-checklist?entity=bohai", headers=auth_header(token))
|
||
assert r.status_code == 403, f"跨实体应 403，实际 {r.status_code}: {r.text[:200]}"
|
||
|
||
def test_coso_checklist_same_entity_200(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/action-plans/coso-checklist?entity=hanke", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert "酣客" in r.json()["entity_name"]
|
||
|
||
def test_coso_checklist_illegal_entity_explicit_error(self, client, db):
|
||
token = self._login(client, db)
|
||
r = client.get("/api/cma/action-plans/coso-checklist?entity=999", headers=auth_header(token))
|
||
assert r.status_code == 400, r.text[:200]
|
||
assert "非法实体" in r.text
|
||
|
||
# ── 对齐参照端点行为不变 ──────────────────────────────
|
||
def test_reference_403_endpoints_unchanged(self, client, db):
|
||
"""growth-quality / predict.accuracy 既有 403 行为不变"""
|
||
token = self._login(client, db)
|
||
r1 = client.get("/api/cma/growth-quality/periods?entity_id=2", headers=auth_header(token))
|
||
assert r1.status_code == 403, r1.text[:200]
|
||
r2 = client.get("/api/cma/predict/accuracy?entity_id=2", headers=auth_header(token))
|
||
assert r2.status_code == 403, r2.text[:200]
|