Hermes CI Fix
40d522cd34
- Login.vue: no prefilled default creds; no anonymous entity enumeration on mount - deploy.sh: readiness polling (0.5s/30s cap) replaces fixed sleep 3; login smoke reads same-source 600 secret file, missing file = explicit abort - 19 verify/check/gen_report scripts: shared _admin_creds() loader (CMA_ADMIN_ENV -> ops .secrets -> profile copy); no hardcoded creds - tests: single TEST_USER_PASSWORD fixture constant; docs: pointer to 600 secret file; removed 4 legacy .bak + pycache residue
378 lines
19 KiB
Python
378 lines
19 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 下线（cma-residual-cross-entity-cleanup-20260914）→ 404；governance-check 保留
|
||
"""
|
||
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, TEST_USER_PASSWORD
|
||
|
||
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", TEST_USER_PASSWORD)
|
||
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", TEST_USER_PASSWORD)
|
||
|
||
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", TEST_USER_PASSWORD)
|
||
|
||
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", TEST_USER_PASSWORD)
|
||
|
||
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 下线（cma-residual-cross-entity-cleanup-20260914）
|
||
# ============================================================
|
||
class TestCheckGovernanceRetired:
|
||
def test_check_governance_endpoint_removed(self, client: TestClient, db: Session):
|
||
"""/check-governance 已下线 → 404（该端点 entity_id 参数无会话鉴权，跨租户）"""
|
||
_seed_entity_kpi(db, entity_id=1, code="GOV_KPI")
|
||
db.commit()
|
||
create_test_user(db)
|
||
token = get_token_for_user(client, "testadmin", TEST_USER_PASSWORD)
|
||
|
||
resp = client.get("/api/cma/data-quality/check-governance", headers=auth_header(token))
|
||
assert resp.status_code == 404, f"应 404，实际 {resp.status_code}: {resp.text[:200]}"
|
||
# 跨租户取数路径随之消失（同参数不再有响应体）
|
||
resp2 = client.get("/api/cma/data-quality/check-governance?entity_id=2", headers=auth_header(token))
|
||
assert resp2.status_code == 404, resp2.text[:200]
|
||
|
||
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", TEST_USER_PASSWORD)
|
||
|
||
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", TEST_USER_PASSWORD) # 登录账套 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]
|
||
|
||
|
||
# ============================================================
|
||
# reports/dupont 口径收敛（cma-residual-cross-entity-cleanup-20260914）
|
||
# 旧「跨实体对比」豁免（2026-08-31）撤销 → 分析对象=会话账套
|
||
# ============================================================
|
||
class TestDupontSessionScope:
|
||
BASE = "/api/cma/reports/dupont"
|
||
|
||
def _login(self, client, db):
|
||
_ensure_entity(db, 2)
|
||
create_test_user(db)
|
||
return get_token_for_user(client, "testadmin", TEST_USER_PASSWORD) # 登录账套 entity_id=1
|
||
|
||
def test_dupont_cross_entity_bohai_403(self, client, db):
|
||
"""酣客会话显式 entity=bohai（博海真实KPI）→ 403（旧豁免已撤销）"""
|
||
token = self._login(client, db)
|
||
r = client.get(f"{self.BASE}?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_dupont_no_param_is_session_entity(self, client, db):
|
||
"""无 entity 参数 → 按会话账套（酣客），不再默认博海"""
|
||
token = self._login(client, db)
|
||
r = client.get(self.BASE, headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["entity"] == "hanke"
|
||
assert "酣客" in r.json()["entity_name"]
|
||
|
||
def test_dupont_same_entity_200(self, client, db):
|
||
"""同实体显式参数 → 200"""
|
||
token = self._login(client, db)
|
||
r = client.get(f"{self.BASE}?entity=hanke", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["entity"] == "hanke"
|