From ac645a56ee6f07f07d1a516e232488b58c9db00f Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Mon, 7 Sep 2026 10:22:24 +0800 Subject: [PATCH] =?UTF-8?q?fix(security):=20ai=5Fanalysis=E5=A4=9A?= =?UTF-8?q?=E7=A7=9F=E6=88=B7=E9=9A=94=E7=A6=BB=E7=BC=BA=E5=8F=A3=20+=20da?= =?UTF-8?q?shboard=E9=A2=84=E8=AD=A6=E8=AE=A1=E6=95=B0=20+=20bot=5Fkpis?= =?UTF-8?q?=E5=BD=92=E5=B1=9E=20+=20cost=E8=B4=A6=E5=A5=97=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=20+=20kpis=20create=E7=99=BD=E5=90=8D=E5=8D=95=20(Ope?= =?UTF-8?q?nCode=E5=AE=A1=E6=9F=A520260907)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/app/api/ai_analysis.py | 71 +++-- backend/app/api/bot_kpis.py | 5 + backend/app/api/cost.py | 56 ++-- backend/app/api/dashboard.py | 5 +- backend/app/api/kpis.py | 7 + backend/app/utils/cost_engine.py | 45 ++- backend/tests/test_multitenant_ai_20260907.py | 274 ++++++++++++++++++ 7 files changed, 410 insertions(+), 53 deletions(-) create mode 100644 backend/tests/test_multitenant_ai_20260907.py diff --git a/backend/app/api/ai_analysis.py b/backend/app/api/ai_analysis.py index 2f76c7d3..3399936a 100644 --- a/backend/app/api/ai_analysis.py +++ b/backend/app/api/ai_analysis.py @@ -156,10 +156,16 @@ def generate_rule_suggestions(db: Session, entity_id: int, "kpi_id": b.kpi_id, "period": period, "budget_value": round(actual, 2), }) - # 4. pending预警 → 建行动方案 - alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").all() + # 4. pending预警 → 建行动方案（多租户隔离 2026-09-07: 只取本账套预警） + alerts = db.query(KPIAlert).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.status == "pending", + ).all() for a in alerts: - k = db.query(KPIDefinition).filter(KPIDefinition.id == a.kpi_id).first() + k = db.query(KPIDefinition).filter( + KPIDefinition.id == a.kpi_id, + KPIDefinition.entity_id == entity_id, + ).first() kpi_name = k.kpi_name if k else f"KPI#{a.kpi_id}" title = f"处理预警：{kpi_name} {a.alert_message[:30]}" content = f"存在待处理预警（{a.alert_level}级）：{a.alert_message}。建议建立行动方案跟进。" @@ -291,8 +297,11 @@ async def dashboard_analysis(role: str = Query("ceo"), db: Session = Depends(get "unit": k.unit, }) - # 获取预警 - alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").count() + # 获取预警（多租户隔离 2026-09-07: 只统计本账套） + alerts = db.query(KPIAlert).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.status == "pending", + ).count() # 构建分析prompt kpi_text = "\n".join([f"- {k['name']}({k['code']}): 目标={k['target']}, 实际={k['actual']}({k['period']}), 维度={k['dimension']}" for k in kpi_summary if k['actual'] is not None]) @@ -339,8 +348,12 @@ async def kpi_analysis(kpi_id: int, db: Session = Depends(get_db), if cached: return cached - kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() + kpi = db.query(KPIDefinition).filter( + KPIDefinition.id == kpi_id, + KPIDefinition.entity_id == entity_id, + ).first() if not kpi: + # 跨企业 kpi_id 同样 404（多租户隔离 2026-09-07） raise HTTPException(404, "KPI不存在") values = db.query(KPIValue).filter(KPIValue.kpi_id == kpi_id).order_by(KPIValue.period.asc()).all() @@ -422,9 +435,11 @@ async def _stream_analysis(prompt: str): @router.get("/dashboard-analysis-stream") -async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depends(get_db)): +async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id)): """AI分析驾驶舱数据 — SSE流式输出""" - cache_key = f"dashboard_analysis:{role}" + # 多租户隔离 2026-09-07: cache key 补 entity_id（原跨企业串缓存） + cache_key = f"dashboard_analysis:{role}:{entity_id}" cached = cache_get("ai", cache_key) if cached: # 缓存存在，直接以流的形式一次性返回 @@ -434,7 +449,10 @@ async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depe yield "data: {\"text\": \"[DONE]\"}\n\n" return StreamingResponse(cached_stream(), media_type="text/event-stream") - kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all() + kpis = db.query(KPIDefinition).filter( + KPIDefinition.entity_id == entity_id, + KPIDefinition.status == "active", + ).all() kpi_summary = [] for k in kpis: latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id).order_by(KPIValue.period.desc()).first() @@ -443,7 +461,10 @@ async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depe "target": k.target_value, "actual": latest.actual_value if latest else None, "period": latest.period if latest else None, "unit": k.unit, }) - alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").count() + alerts = db.query(KPIAlert).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.status == "pending", + ).count() kpi_text = "\n".join([f"- {k['name']}({k['code']}): 目标={k['target']}, 实际={k['actual']}({k['period']}), 维度={k['dimension']}" for k in kpi_summary if k['actual'] is not None]) role_label = {"ceo": "CEO（总经理）", "finance": "财务部", "business": "业务部"}.get(role, "管理层") prompt = f"""我是一家公司的管理层，以下是当前管理会计系统的KPI数据和系统状态，请给出专业的分析和管理建议： @@ -468,6 +489,7 @@ async def ask_question( request: Request, db: Session = Depends(get_db), current_user: User = Depends(require_auth), + entity_id: int = Depends(get_entity_id), ): """自然语言查询 — CEO问企业经营问题""" body = await request.json() @@ -476,8 +498,11 @@ async def ask_question( if not question: raise HTTPException(400, "请输入问题") - # 收集系统数据作为上下文 - kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all() + # 收集系统数据作为上下文（多租户隔离 2026-09-07: 只取本账套） + kpis = db.query(KPIDefinition).filter( + KPIDefinition.entity_id == entity_id, + KPIDefinition.status == "active", + ).all() kpi_context = [] for k in kpis: latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id).order_by(KPIValue.period.desc()).first() @@ -487,13 +512,21 @@ async def ask_question( f"{' ⚠️' + alert.alert_level if alert else ''}" ) - # 获取改善计划 - plans = db.query(ActionPlan).order_by(ActionPlan.created_at.desc()).limit(10).all() + # 获取改善计划（多租户隔离 2026-09-07: 只取本账套） + plans = db.query(ActionPlan).filter( + ActionPlan.entity_id == entity_id, + ).order_by(ActionPlan.created_at.desc()).limit(10).all() plan_context = [f"- {p.title}({p.assignee}, {p.status}, {p.progress}%)" for p in plans] # 获取预警 - red_alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending", KPIAlert.alert_level == "red").count() - yellow_alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending", KPIAlert.alert_level == "yellow").count() + red_alerts = db.query(KPIAlert).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.status == "pending", KPIAlert.alert_level == "red", + ).count() + yellow_alerts = db.query(KPIAlert).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.status == "pending", KPIAlert.alert_level == "yellow", + ).count() system_context = f"""你是管理会计OS的AI助手，基于以下企业数据回答管理层问题。 @@ -525,9 +558,13 @@ async def ask_question( async def review_plans( db: Session = Depends(get_db), current_user: User = Depends(require_auth), + entity_id: int = Depends(get_entity_id), ): """AI复盘改善行动计划执行效果""" - plans = db.query(ActionPlan).order_by(ActionPlan.created_at.asc()).all() + # 多租户隔离 2026-09-07: ActionPlan 按 entity_id 过滤（表 2026-08-31 已加列回填） + plans = db.query(ActionPlan).filter( + ActionPlan.entity_id == entity_id, + ).order_by(ActionPlan.created_at.asc()).all() if not plans: return {"analysis": "暂无改善行动计划，无法复盘"} diff --git a/backend/app/api/bot_kpis.py b/backend/app/api/bot_kpis.py index 8a982bd8..285ec894 100644 --- a/backend/app/api/bot_kpis.py +++ b/backend/app/api/bot_kpis.py @@ -6,6 +6,7 @@ from datetime import datetime from app.database import get_db from app.auth_middleware import require_auth +from app.deps import get_entity_id from app.models import KPIDefinition, KPIValue from app.risk_levels import risk_level @@ -142,6 +143,7 @@ def update_bot_kpi_value( kpi_id: int, data: dict, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), ): """更新Bot KPI当前值""" kpi = db.query(KPIDefinition).filter( @@ -150,6 +152,9 @@ def update_bot_kpi_value( ).first() if not kpi: raise HTTPException(404, "Bot KPI不存在") + # 多租户隔离 2026-09-07: 归属校验，禁止跨企业修改他企 Bot KPI + if kpi.entity_id != entity_id: + raise HTTPException(404, "Bot KPI不存在") actual_value = data.get("actual_value") if actual_value is None: diff --git a/backend/app/api/cost.py b/backend/app/api/cost.py index c8159b9d..70cca7e7 100644 --- a/backend/app/api/cost.py +++ b/backend/app/api/cost.py @@ -170,6 +170,8 @@ def create_actual_cost(data: dict, db: Session = Depends(get_db), # ============================================================ # ABC 作业成本 +# 注：ABC 作业中心/分配为单实体演示数据（ab_activities/ab_allocations 全为 entity 1， +# 历史演示用途），OpenCode 多租户审查 2026-09-07 判定暂不强制按账套隔离；若启用多企业 ABC 需补 entity 过滤。 # ============================================================ @router.get("/abc/activities") @@ -229,34 +231,41 @@ def list_allocations(period: Optional[str] = Query(None), # ============================================================ @router.get("/overview") -def cost_overview(period: Optional[str] = Query(None)): - """成本总览（总成本、结构占比、趋势）""" +def cost_overview(period: Optional[str] = Query(None), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id)): + """成本总览（总成本、结构占比、趋势）— 多租户隔离 2026-09-07""" if period is None: period = datetime.now().strftime("%Y-%m") - return get_cost_overview(period) + return get_cost_overview(period, entity_id=entity_id) @router.get("/variance") def cost_variance(product_code: str = Query(...), - period: Optional[str] = Query(None)): - """量差价差分析""" + period: Optional[str] = Query(None), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id)): + """量差价差分析 — 多租户隔离 2026-09-07""" if period is None: period = datetime.now().strftime("%Y-%m") - return calc_product_variance(product_code, period) + return calc_product_variance(product_code, period, entity_id=entity_id) @router.get("/breakdown") def cost_breakdown(product_code: str = Query(...), - period: Optional[str] = Query(None)): - """成本构成（料/工/费占比）""" + period: Optional[str] = Query(None), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id)): + """成本构成（料/工/费占比）— 多租户隔离 2026-09-07""" if period is None: period = datetime.now().strftime("%Y-%m") - return get_cost_breakdown(product_code, period) + return get_cost_breakdown(product_code, period, entity_id=entity_id) @router.get("/comparison") def cost_method_comparison(entity: str = Query("hanke")): - """三种成本法对比分析：传统/变动/作业成本法 (CMA P1)""" + """三种成本法对比分析：传统/变动/作业成本法 (CMA P1) + 注：纯静态教学样例数据（无 DB 访问），不涉多租户泄漏——OpenCode 审查 2026-09-07 豁免标注""" if entity == "hanke": gross_revenue = 713 # 万 channel_rebate_rate = 0.828 @@ -317,35 +326,42 @@ def cost_method_comparison(entity: str = Query("hanke")): @router.get("/dashboard") -def cost_dashboard(period: Optional[str] = Query(None)): - """成本分析首页—汇总数据""" +def cost_dashboard(period: Optional[str] = Query(None), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id)): + """成本分析首页—汇总数据（多租户隔离 2026-09-07: ActualCost 按账套过滤）""" if period is None: period = datetime.now().strftime("%Y-%m") - overview = get_cost_overview(period) + overview = get_cost_overview(period, entity_id=entity_id) - # 获取所有产品列表 - db = get_db().__next__() + # 获取所有产品列表（本账套） + db_session = db if db is not None else get_db().__next__() try: - products = db.query(ActualCost.product_code, ActualCost.product_name).filter( - ActualCost.period == period + products = db_session.query(ActualCost.product_code, ActualCost.product_name).filter( + ActualCost.entity_id == entity_id, + ActualCost.period == period, ).distinct().all() product_list = [{"code": p[0], "name": p[1]} for p in products] # 各产品成本 product_costs = [] for code, name in products: - costs = db.query(ActualCost).filter( + costs = db_session.query(ActualCost).filter( + ActualCost.entity_id == entity_id, ActualCost.product_code == code, ActualCost.period == period, ).all() total = round(sum(c.actual_cost for c in costs), 2) product_costs.append({"product_code": code, "product_name": name, "total_cost": total}) finally: - db.close() + if db is None: + db_session.close() return { "period": period, "overview": overview, "products": product_list, - "total_cost": round(sum(p["total_cost"] for p in product_list) + overview.get("erp_cost", 0), 2) if product_list else 0, + # 修复既有bug(2026-09-07测试实证): 原对 product_list 求 p["total_cost"] 必 KeyError(产品列表无total_cost键) + # → 改为 product_costs（含各产品总额），有产品数据时端点不再 500 + "total_cost": round(sum(p["total_cost"] for p in product_costs) + overview.get("erp_cost", 0), 2) if product_costs else 0, } diff --git a/backend/app/api/dashboard.py b/backend/app/api/dashboard.py index 4eec9d26..f06f38dc 100644 --- a/backend/app/api/dashboard.py +++ b/backend/app/api/dashboard.py @@ -79,7 +79,10 @@ def get_dashboard_summary(role: str = Query("ceo"), period: str = Query("month") return cached kpi_total = db.query(func.count(KPIDefinition.id)).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).scalar() - alert_count = db.query(func.count(KPIAlert.id)).filter(KPIAlert.status == "pending").scalar() + alert_count = db.query(func.count(KPIAlert.id)).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.status == "pending", + ).scalar() dims = db.query(KPIDefinition.dimension, func.count(KPIDefinition.id)).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id ).group_by(KPIDefinition.dimension).all() diff --git a/backend/app/api/kpis.py b/backend/app/api/kpis.py index 8885656b..d8fb0bc6 100644 --- a/backend/app/api/kpis.py +++ b/backend/app/api/kpis.py @@ -731,6 +731,13 @@ def create_kpi(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES, enti raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs}) data["entity_id"] = entity_id # 账套隔离: 强制写入token企业 (2026-08-23 P1a) data = apply_calc_type_inference(data) + # create 侧字段白名单（OpenCode 审查残留项修复 2026-09-07）： + # 仿 update 白名单 + kpi_code（新建必填），拦截 status/important_flag/主键等敏感字段注入 + # （update 2026-08-31 已有白名单；create 此前直接 KPIDefinition(**data) 展开，可被注入） + _create_allow = UPDATE_KPI_WHITELIST | {"kpi_code"} + for k in list(data.keys()): + if k != "entity_id" and k not in _create_allow: + data.pop(k, None) kpi = KPIDefinition(**data) db.add(kpi) db.commit() diff --git a/backend/app/utils/cost_engine.py b/backend/app/utils/cost_engine.py index a624b6b7..0fece6e5 100644 --- a/backend/app/utils/cost_engine.py +++ b/backend/app/utils/cost_engine.py @@ -56,18 +56,23 @@ def calc_efficiency_variance(standard_hours: float, actual_hours: float, # 产品级差异分析 # ============================================================ -def calc_product_variance(product_code: str, period: str) -> dict: - """计算指定产品在指定期间的成本差异""" +def calc_product_variance(product_code: str, period: str, entity_id: Optional[int] = None) -> dict: + """计算指定产品在指定期间的成本差异（entity_id=None 时不过滤，API 调用必传——多租户隔离 2026-09-07）""" db = get_session_local()() try: - standards = db.query(StandardCost).filter( + q_std = db.query(StandardCost).filter( StandardCost.product_code == product_code, StandardCost.status == "active", - ).all() - actuals = db.query(ActualCost).filter( + ) + q_act = db.query(ActualCost).filter( ActualCost.product_code == product_code, ActualCost.period == period, - ).all() + ) + if entity_id is not None: + q_std = q_std.filter(StandardCost.entity_id == entity_id) + q_act = q_act.filter(ActualCost.entity_id == entity_id) + standards = q_std.all() + actuals = q_act.all() if not standards or not actuals: return {"product_code": product_code, "error": "标准成本或实际成本数据不足", "items": [], "summary": {}} @@ -191,12 +196,16 @@ def allocate_cost(activity_id: int, period: str, product_code: str, # 成本总览数据 # ============================================================ -def get_cost_overview(period: str) -> dict: - """获取成本总览数据（总成本、结构占比、趋势）""" +def get_cost_overview(period: str, entity_id: Optional[int] = None) -> dict: + """获取成本总览数据（总成本、结构占比、趋势） + entity_id=None 时不过滤（兼容存量调用）；API 调用必须传 entity_id —— 多租户隔离 2026-09-07""" db = get_session_local()() try: - # 从实际成本表汇总 - actual_costs = db.query(ActualCost).filter(ActualCost.period == period).all() + # 从实际成本表汇总（按账套过滤） + q_costs = db.query(ActualCost).filter(ActualCost.period == period) + if entity_id is not None: + q_costs = q_costs.filter(ActualCost.entity_id == entity_id) + actual_costs = q_costs.all() total_cost = sum(a.actual_cost for a in actual_costs) # 按成本类型分组 @@ -238,7 +247,10 @@ def get_cost_overview(period: str) -> dict: trend = [] for p in reversed(months_texts): - costs = db.query(ActualCost).filter(ActualCost.period == p).all() + q_trend = db.query(ActualCost).filter(ActualCost.period == p) + if entity_id is not None: + q_trend = q_trend.filter(ActualCost.entity_id == entity_id) + costs = q_trend.all() total = round(sum(c.actual_cost for c in costs), 2) trend.append({"period": p, "total_cost": total}) except Exception: @@ -256,14 +268,17 @@ def get_cost_overview(period: str) -> dict: db.close() -def get_cost_breakdown(product_code: str, period: str) -> dict: - """获取成本构成（料/工/费占比）""" +def get_cost_breakdown(product_code: str, period: str, entity_id: Optional[int] = None) -> dict: + """获取成本构成（料/工/费占比）（entity_id=None 不过滤——多租户隔离 2026-09-07，API 必传）""" db = get_session_local()() try: - actuals = db.query(ActualCost).filter( + q_actuals = db.query(ActualCost).filter( ActualCost.product_code == product_code, ActualCost.period == period, - ).all() + ) + if entity_id is not None: + q_actuals = q_actuals.filter(ActualCost.entity_id == entity_id) + actuals = q_actuals.all() material = sum(a.actual_cost for a in actuals if a.cost_type == "material") labor = sum(a.actual_cost for a in actuals if a.cost_type == "labor") diff --git a/backend/tests/test_multitenant_ai_20260907.py b/backend/tests/test_multitenant_ai_20260907.py new file mode 100644 index 00000000..789e5929 --- /dev/null +++ b/backend/tests/test_multitenant_ai_20260907.py @@ -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}"