fix(security): ai_analysis多租户隔离缺口 + dashboard预警计数 + bot_kpis归属 + cost账套过滤 + kpis create白名单 (OpenCode审查20260907)

- ai_analysis: generate_rule_suggestions pending预警按entity过滤; kpi_analysis归属校验404; dashboard-analysis-stream/ask/review-plans 补 get_entity_id + KPI/ActionPlan/KPIAlert 全按账套过滤; stream cache key 含 entity_id (原跨企业串缓存) - dashboard: summary alert_count 补 KPIAlert.entity_id 过滤 (kpi_total 已有) - bot_kpis: update_bot_kpi_value 归属校验 404 (原任意登录可改他企 bot KPI) - cost: ActualCost/StandardCost 确认多企业共用 → cost_overview/variance/breakdown/dashboard 补 entity 过滤(engine 函数加 entity_id 透传); dashboard total_cost 既有 KeyError 修复 - kpis: create_kpi 字段白名单 (仿 UPDATE_KPI_WHITELIST+kpi_code), 拦截 status/important_flag 注入 - tests: 新增 test_multitenant_ai_20260907.py 11条越权用例 (跨entity 404/空返回断言) - 全量 pytest: 745 passed / 10 failed(与基线 HEAD 一致, 存量问题非本次引入) / 1 xfailed
2026-09-07 10:22:24 +08:00
parent d71c7fef67
commit ac645a56ee
7 changed files with 410 additions and 53 deletions
@@ -0,0 +1,274 @@
+"""OpenCode 审查 2026-09-07 修复验证 — ai_analysis/dashboard/bot_kpis/cost/kpis 多租户隔离越权用例
+
+覆盖：
+- ai_analysis kpi_analysis 跨企业 kpi_id → 404
+- ai_analysis generate_rule_suggestions 跨企业预警不进建议
+- ai_analysis review-plans / ask 按账套过滤（统计口径 + 不 500）
+- dashboard summary alert_count 按账套
+- bot_kpis update 跨企业写 → 404 / 同企业写 → 200
+- cost dashboard/overview 按账套（ActualCost 多企业共用确认）
+- kpis create 白名单：status/important_flag 注入被忽略
+"""
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan, AISuggestion, Entity, UserEntity, ActualCost
+from app.api import ai_analysis
+from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
+
+
+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 _seed_kpi(db: Session, entity_id: int, code: str) -> KPIDefinition:
+    return create_test_kpi(db, kpi_code=code, entity_id=entity_id,
+                           dimension="finance", status="active", target_value=100.0)
+
+
+def _seed_user(db: Session, entity_id: int):
+    """默认测试用户，登录指定账套"""
+    create_test_user(db)
+    _ensure_entity(db, entity_id)
+    # 无授权记录 → login 会默认授予全部 active 企业（_ensure_default_grants）
+    db.commit()
+
+
+def _token(client: TestClient, entity_id: int) -> str:
+    """登录并返回指定账套的 token（get_token_for_user 固定 entity1，这里手动登录）"""
+    resp = client.post("/api/cma/auth/login", json={
+        "username": "testadmin", "password": "admin123", "entity_id": entity_id,
+    })
+    assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
+    return resp.json().get("token")
+
+
+# ============================================================
+# ai_analysis
+# ============================================================
+class TestAiAnalysisIsolation:
+    def test_kpi_analysis_cross_entity_404(self, client: TestClient, db: Session):
+        """跨企业 kpi_id 调 kpi-analysis → 404（不泄漏他企 KPI 数据）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k_e2 = _seed_kpi(db, entity_id=2, code="MT_AI_E2")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.get(f"/api/cma/ai/kpi-analysis/{k_e2.id}", headers=auth_header(token1))
+        assert r.status_code == 404, f"跨企业 kpi 应 404: {r.text[:200]}"
+
+    def test_kpi_analysis_own_entity_ok(self, client: TestClient, db: Session):
+        """本企业 kpi_id → 200（回归保护）"""
+        _seed_user(db, 1)
+        k_e1 = _seed_kpi(db, entity_id=1, code="MT_AI_E1")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.get(f"/api/cma/ai/kpi-analysis/{k_e1.id}", headers=auth_header(token1))
+        assert r.status_code == 200, r.text[:300]
+
+    def test_generate_suggestions_cross_entity_alert_excluded(self, db: Session):
+        """generate_rule_suggestions(entity=1) 不消费 entity2 的 pending 预警"""
+        _ensure_entity(db, 2)
+        k1 = _seed_kpi(db, entity_id=1, code="MT_SUG_E1")
+        k2 = _seed_kpi(db, entity_id=2, code="MT_SUG_E2")
+        db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="e1预警", status="pending", entity_id=1))
+        a2 = KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="e2预警", status="pending", entity_id=2)
+        db.add(a2)
+        db.commit()
+
+        ai_analysis.generate_rule_suggestions(db, entity_id=1, source="test", user_id=None)
+        db.commit()
+
+        # entity1 的预警 → 生成 alert 类建议
+        sugs = db.query(AISuggestion).filter(AISuggestion.source == "test").all()
+        alert_sugs = [s for s in sugs if s.suggestion_type == "action_plan" and s.target_type == "alert"]
+        assert len(alert_sugs) >= 1, "本账套 pending 预警应生成建议"
+        # 无一建议引用 entity2 的预警 id / KPI
+        sug_alert_ids = {s.suggestion_data.get("alert_id") for s in alert_sugs}
+        sug_kpi_ids = {s.suggestion_data.get("kpi_id") for s in alert_sugs}
+        assert a2.id not in sug_alert_ids, "跨企业预警不应进入建议"
+        assert k2.id not in sug_kpi_ids, "跨企业 KPI 不应进入建议"
+
+    def test_review_plans_entity_scoped(self, client: TestClient, db: Session):
+        """review-plans 只复盘本账套 ActionPlan（stats.total 口径隔离）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k1 = _seed_kpi(db, entity_id=1, code="MT_PL_E1")
+        k2 = _seed_kpi(db, entity_id=2, code="MT_PL_E2")
+        db.add(ActionPlan(entity_id=1, alert_id=None, kpi_id=k1.id, title="e1计划", assignee="张三",
+                          priority="high", status="in_progress", progress=50, created_by="系统"))
+        db.add(ActionPlan(entity_id=2, alert_id=None, kpi_id=k2.id, title="e2计划", assignee="李四",
+                          priority="high", status="pending", progress=0, created_by="系统"))
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post("/api/cma/ai/review-plans", headers=auth_header(token1))
+        assert r.status_code == 200, r.text[:300]
+        stats = r.json()["stats"]
+        assert stats["total"] == 1, f"应只统计本账套1条，实际 {stats}"
+
+    def test_ask_question_no_crash_cross_entity(self, client: TestClient, db: Session):
+        """ask 上下文含他企 KPI 时不 500 且正常返回（entity 过滤后无泄漏）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k2 = _seed_kpi(db, entity_id=2, code="MT_ASK_E2")
+        db.add(KPIValue(kpi_id=k2.id, period="2026-06", actual_value=88.0, entity_id=2))
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post("/api/cma/ai/ask", headers=auth_header(token1),
+                        json={"question": "企业营收情况如何"})
+        assert r.status_code == 200, r.text[:300]
+        assert "answer" in r.json()
+
+
+# ============================================================
+# dashboard summary alert_count
+# ============================================================
+class TestDashboardAlertCountIsolation:
+    def test_alert_count_per_entity(self, client: TestClient, db: Session):
+        """summary alert_count 按账套：entity1 只数自己的 pending 预警"""
+        from app.utils import cache as cache_util
+        cache_util.delete("dashboard")  # 防跨测试缓存污染
+
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k1 = _seed_kpi(db, entity_id=1, code="MT_DB_E1")
+        k2 = _seed_kpi(db, entity_id=2, code="MT_DB_E2")
+        db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="e1", status="pending", entity_id=1))
+        db.add(KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="e2a", status="pending", entity_id=2))
+        db.add(KPIAlert(kpi_id=k2.id, alert_level="yellow", alert_message="e2b", status="resolved", entity_id=2))
+        db.commit()
+        token1 = _token(client, 1)
+        token2 = _token(client, 2)
+
+        r1 = client.get("/api/cma/dashboard/summary", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text[:200]
+        assert r1.json()["alert_count"] == 1, f"entity1 应只数自己的1条: {r1.json()['alert_count']}"
+
+        cache_util.delete("dashboard")
+        r2 = client.get("/api/cma/dashboard/summary", headers=auth_header(token2))
+        assert r2.status_code == 200, r2.text[:200]
+        assert r2.json()["alert_count"] == 1, f"entity2 应只数自己的 pending 1条(不含resolved): {r2.json()['alert_count']}"
+
+
+# ============================================================
+# bot_kpis 归属校验
+# ============================================================
+class TestBotKpisOwnership:
+    def _bot_kpi(self, db: Session, entity_id: int, code: str) -> KPIDefinition:
+        return create_test_kpi(db, kpi_code=code, entity_id=entity_id, status="active",
+                               bot_source="finance-bot", kpi_name=code, target_value=100.0)
+
+    def test_cross_entity_update_404(self, client: TestClient, db: Session):
+        """entity1 用户改 entity2 的 Bot KPI → 404"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k2 = self._bot_kpi(db, entity_id=2, code="MT_BOT_E2")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post(f"/api/cma/bot-kpis/{k2.id}/value", headers=auth_header(token1),
+                        json={"actual_value": 55, "period": "2026-09"})
+        assert r.status_code == 404, f"跨企业改 Bot KPI 应 404: {r.status_code} {r.text[:200]}"
+
+    def test_same_entity_update_ok(self, client: TestClient, db: Session):
+        """本企业改自己的 Bot KPI → 200（回归保护）"""
+        _seed_user(db, 1)
+        k1 = self._bot_kpi(db, entity_id=1, code="MT_BOT_E1")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post(f"/api/cma/bot-kpis/{k1.id}/value", headers=auth_header(token1),
+                        json={"actual_value": 66, "period": "2026-09"})
+        assert r.status_code == 200, r.text[:300]
+
+
+# ============================================================
+# cost 多账套（ActualCost 共用表确认 → 过滤）
+# ============================================================
+class TestCostIsolation:
+    def _seed_actual(self, db: Session, entity_id: int, code: str, amount: float, period: str = "2026-06"):
+        ac = ActualCost(entity_id=entity_id, period=period, product_code=code, product_name=code,
+                        cost_type="material", item_name="材料", actual_quantity=1.0,
+                        actual_price=amount, actual_cost=amount, source="manual")
+        db.add(ac)
+
+    def test_cost_dashboard_entity_filtered(self, client: TestClient, db: Session):
+        """cost/dashboard 产品列表/总额按账套过滤（ActualCost 多企业共用）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        self._seed_actual(db, 1, "PROD_E1", 100.0)
+        self._seed_actual(db, 2, "PROD_E2", 200.0)
+        db.commit()
+        token1 = _token(client, 1)
+        token2 = _token(client, 2)
+
+        r1 = client.get("/api/cma/cost/dashboard?period=2026-06", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text[:300]
+        codes1 = {p["code"] for p in r1.json()["products"]}
+        assert codes1 == {"PROD_E1"}, f"entity1 不应看到 entity2 产品: {codes1}"
+
+        r2 = client.get("/api/cma/cost/dashboard?period=2026-06", headers=auth_header(token2))
+        assert r2.status_code == 200, r2.text[:300]
+        codes2 = {p["code"] for p in r2.json()["products"]}
+        assert codes2 == {"PROD_E2"}, f"entity2 不应看到 entity1 产品: {codes2}"
+
+    def test_cost_overview_entity_filtered(self, client: TestClient, db: Session):
+        """cost/overview manual_cost 按账套过滤"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        self._seed_actual(db, 1, "OV_E1", 100.0)
+        self._seed_actual(db, 2, "OV_E2", 300.0)
+        db.commit()
+        token1 = _token(client, 1)
+        token2 = _token(client, 2)
+
+        r1 = client.get("/api/cma/cost/overview?period=2026-06", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text[:300]
+        assert r1.json()["manual_cost"] == 100.0, f"entity1 manual_cost 应100: {r1.json()}"
+
+        r2 = client.get("/api/cma/cost/overview?period=2026-06", headers=auth_header(token2))
+        assert r2.status_code == 200, r2.text[:300]
+        assert r2.json()["manual_cost"] == 300.0, f"entity2 manual_cost 应300: {r2.json()}"
+
+
+# ============================================================
+# kpis create 白名单（mass-assignment 残留项）
+# ============================================================
+class TestKpiCreateWhitelist:
+    def test_create_injected_sensitive_fields_ignored(self, client: TestClient, db: Session):
+        """create 带 status/important_flag 注入 → 被忽略，落库为默认值"""
+        _seed_user(db, 1)
+        token1 = _token(client, 1)
+
+        resp = client.post("/api/cma/kpis", headers=auth_header(token1), json={
+            "kpi_code": "F_MT_INJ_001",
+            "kpi_name": "注入测试指标",
+            "dimension": "finance",
+            "target_value": 100.0,
+            "unit": "%",
+            "formula": "测试公式",
+            "data_source": "测试系统",
+            "data_owner": "测试管理员",
+            "status": "disabled",      # 敏感字段：注入应被忽略 → 默认 active
+            "important_flag": 1,       # 敏感字段：注入应被忽略 → 默认 0
+            "entity_id": 999,          # 归属字段：强制覆盖为 token 企业 1
+        })
+        assert resp.status_code == 200, resp.text[:300]
+        data = resp.json()
+        assert data["status"] == "active", f"status 注入应被忽略: {data.get('status')}"
+        assert data.get("important_flag", 0) == 0, f"important_flag 注入应被忽略: {data.get('important_flag')}"
+
+        row = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_MT_INJ_001").first()
+        assert row is not None
+        assert row.entity_id == 1, f"entity_id 应强制为 token 企业1: {row.entity_id}"