Hermes CI Fix
74dc9baff5
- bot_bridge 18数据端点全部 entity_id 隔离（Depends(get_entity_id)/body），/ping /risk-levels 豁免 - alert_rules 11端点 entity_id 隔离 + KPIAlert/DynamicThresholdCache 写入 entity_id - reports 17端点隔离 + generate_report 写 ReportHistory.entity_id + history 按 entity 过滤 - ai_analysis 移除硬编码默认key，改 _require_deepseek_key() 强制 env 缺失 503 - budget auto-decompose 硬编码 entity_id==1 改请求 entity - kpis update_kpi 加 UPDATE_KPI_WHITELIST 白名单（status/important_flag 不可越权改） - data_quality 收敛：删 MySQL JSON 版 _run_rule_checks，check-governance 复用 _run_governance_checks（SQLite 兼容） - _eval_threshold invert 参数修复（>=↔< 等取反），red 分支不传 invert 保持行为 - 新增 test_security_multitenant.py 13条（bot_bridge/alert_rules/reports 隔离 + invert + SQLite governance） - models 6表加 entity_id 列；生产库已 ALTER + 按真实归属回填（kpi_alerts 472行中216行属entity≠1）
233 lines
11 KiB
Python
233 lines
11 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
|