sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(security): 多租户隔离全量修复 security-fix multi-tenant (OpenCode审查P0)

Browse Source 
- 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）
This commit is contained in:
Hermes CI Fix
2026-08-31 10:14:22 +08:00
parent 72072dda8c
commit 74dc9baff5
11 changed files with 560 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/alert_rules.py
+70 -46
View File
@@ -41,6 +41,7 @@ class DynamicThresholdCache(Base):
"""动态阈值缓存 — 存储近3个月历史统计"""
__tablename__ = "dynamic_threshold_cache"
id = Column(Integer, primary_key=True, index=True)
entity_id = Column(Integer, default=1, comment="企业ID (多租户隔离 2026-08-31 安全修复)")
kpi_id = Column(Integer, nullable=False, comment="关联KPI ID")
period = Column(String(20), nullable=False, comment="计算期间 2026-07")
mean_value = Column(Float, nullable=True, comment="近3月均值")
@@ -91,15 +92,16 @@ def list_alert_rules(
@router.get("/kpi/{kpi_id}")
def get_kpi_rules(kpi_id: int, db: Session = Depends(get_db)):
"""获取单个KPI的所有预警规则"""
rules = db.query(AlertRule).filter(AlertRule.kpi_id == kpi_id).order_by(AlertRule.id).all()
def get_kpi_rules(kpi_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""获取单个KPI的所有预警规则（账套隔离: 按token企业）"""
rules = db.query(AlertRule).filter(
AlertRule.kpi_id == kpi_id, AlertRule.entity_id == entity_id).order_by(AlertRule.id).all()
return {"data": [{c.name: getattr(r, c.name) for c in AlertRule.__table__.columns} for r in rules]}
@router.post("")
def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(require_role("ceo", "finance", "it"))):
"""创建预警规则"""
def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(require_role("ceo", "finance", "it")), entity_id: int = Depends(get_entity_id)):
"""创建预警规则（账套隔离: 写入token企业, 2026-08-31 安全修复）"""
kpi_id = data.get("kpi_id")
rule_type = data.get("rule_type", "static")
trigger_on = data.get("trigger_on", "actual")
@@ -107,10 +109,13 @@ def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(re
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
if not kpi:
raise HTTPException(404, "KPI不存在")
if kpi.entity_id != entity_id:
raise HTTPException(404, "KPI不存在")
if rule_type not in ("static", "dynamic", "trend_up", "trend_down", "forecast_deviation"): # 升级2b: 预测偏差
raise HTTPException(400, f"不支持的规则类型: {rule_type}")
rule = AlertRule(
entity_id=entity_id,
kpi_id=kpi_id,
rule_type=rule_type,
trigger_on=trigger_on,
@@ -132,9 +137,9 @@ def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(re
@router.put("/{rule_id}")
def update_alert_rule(rule_id: int, data: dict, db: Session = Depends(get_db)):
"""更新预警规则"""
rule = db.query(AlertRule).filter(AlertRule.id == rule_id).first()
def update_alert_rule(rule_id: int, data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""更新预警规则（账套隔离: 禁止跨企业修改）"""
rule = db.query(AlertRule).filter(AlertRule.id == rule_id, AlertRule.entity_id == entity_id).first()
if not rule:
raise HTTPException(404, "预警规则不存在")
@@ -147,9 +152,9 @@ def update_alert_rule(rule_id: int, data: dict, db: Session = Depends(get_db)):
@router.delete("/{rule_id}")
def delete_alert_rule(rule_id: int, db: Session = Depends(get_db)):
"""删除预警规则"""
rule = db.query(AlertRule).filter(AlertRule.id == rule_id).first()
def delete_alert_rule(rule_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""删除预警规则（账套隔离: 禁止跨企业删除）"""
rule = db.query(AlertRule).filter(AlertRule.id == rule_id, AlertRule.entity_id == entity_id).first()
if rule:
db.delete(rule)
db.commit()
@@ -157,8 +162,8 @@ def delete_alert_rule(rule_id: int, db: Session = Depends(get_db)):
@router.post("/batch")
def batch_create_rules(data: dict, db: Session = Depends(get_db)):
"""批量创建预警规则
def batch_create_rules(data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""批量创建预警规则（账套隔离: 全部写入token企业）
data.rules: [{"kpi_id": id, "rule_type": "static", "params": {...}}, ...]
"""
rules_data = data.get("rules", [])
@@ -166,14 +171,19 @@ def batch_create_rules(data: dict, db: Session = Depends(get_db)):
for rule_data in rules_data:
kpi_id = rule_data.get("kpi_id")
rule_type = rule_data.get("rule_type", "static")
# 检查是否已存在相同类型的规则
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
if not kpi or kpi.entity_id != entity_id:
continue
# 检查是否已存在相同类型的规则（同企业内）
existing = db.query(AlertRule).filter(
AlertRule.kpi_id == kpi_id,
AlertRule.rule_type == rule_type,
AlertRule.entity_id == entity_id,
).first()
if existing:
continue
rule = AlertRule(
entity_id=entity_id,
kpi_id=kpi_id,
rule_type=rule_type,
enabled=rule_data.get("enabled", 1),
@@ -186,15 +196,17 @@ def batch_create_rules(data: dict, db: Session = Depends(get_db)):
@router.post("/generate-defaults")
def generate_default_rules(db: Session = Depends(get_db)):
"""为所有尚未配置预警规则的KPI生成默认规则"""
# 找到所有active KPI
all_kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
def generate_default_rules(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""为当前企业尚未配置预警规则的KPI生成默认规则（账套隔离 2026-08-31）"""
# 找到当前企业所有active KPI
all_kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).all()
created = 0
for kpi in all_kpis:
# 检查是否已有任何规则
existing = db.query(AlertRule).filter(AlertRule.kpi_id == kpi.id).first()
# 检查是否已有任何规则（同企业内）
existing = db.query(AlertRule).filter(
AlertRule.kpi_id == kpi.id, AlertRule.entity_id == entity_id).first()
if existing:
continue
@@ -203,6 +215,7 @@ def generate_default_rules(db: Session = Depends(get_db)):
# 1. 静态阈值规则（基于kpi_definitions的阈值）
if kpi.threshold_green or kpi.threshold_yellow or kpi.threshold_red:
rule = AlertRule(
entity_id=entity_id,
kpi_id=kpi_id,
rule_type="static",
enabled=1,
@@ -217,6 +230,7 @@ def generate_default_rules(db: Session = Depends(get_db)):
# 2. 动态趋势规则（所有KPI默认加 trend_down）
rule2 = AlertRule(
entity_id=entity_id,
kpi_id=kpi_id,
rule_type="trend_down",
enabled=1,
@@ -230,9 +244,10 @@ def generate_default_rules(db: Session = Depends(get_db)):
@router.post("/check-all")
def run_all_alert_checks(db: Session = Depends(get_db)):
"""执行所有KPI的预警检查 — 生成新的预警记录"""
rules = db.query(AlertRule).filter(AlertRule.enabled == 1).all()
def run_all_alert_checks(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""执行当前企业所有KPI的预警检查 — 生成新的预警记录（账套隔离 2026-08-31）"""
rules = db.query(AlertRule).filter(
AlertRule.enabled == 1, AlertRule.entity_id == entity_id).all()
kpi_cache = {}
value_cache = {}
@@ -246,7 +261,7 @@ def run_all_alert_checks(db: Session = Depends(get_db)):
if kpi:
kpi_cache[rule.kpi_id] = kpi
if not kpi:
if not kpi or kpi.entity_id != entity_id:
continue
# 获取最新值
@@ -288,6 +303,7 @@ def run_all_alert_checks(db: Session = Depends(get_db)):
).first()
if not existing_alert:
alert = KPIAlert(
entity_id=entity_id,
kpi_id=rule.kpi_id,
kpi_value_id=latest_value.id,
alert_level=alert_level,
@@ -306,9 +322,9 @@ def run_all_alert_checks(db: Session = Depends(get_db)):
@router.get("/dynamic-thresholds")
def get_dynamic_thresholds(kpi_id: Optional[int] = None, db: Session = Depends(get_db)):
"""获取动态阈值缓存"""
query = db.query(DynamicThresholdCache)
def get_dynamic_thresholds(kpi_id: Optional[int] = None, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""获取动态阈值缓存（账套隔离 2026-08-31）"""
query = db.query(DynamicThresholdCache).filter(DynamicThresholdCache.entity_id == entity_id)
if kpi_id:
query = query.filter(DynamicThresholdCache.kpi_id == kpi_id)
cache = query.order_by(DynamicThresholdCache.id.desc()).limit(50).all()
@@ -316,9 +332,10 @@ def get_dynamic_thresholds(kpi_id: Optional[int] = None, db: Session = Depends(g
@router.post("/calculate-dynamic")
def calculate_dynamic_thresholds(db: Session = Depends(get_db)):
"""计算所有KPI的动态阈值（基于近3个月历史均值±标准差）"""
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
def calculate_dynamic_thresholds(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""计算当前企业所有KPI的动态阈值（账套隔离 2026-08-31）"""
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).all()
current_period = datetime.now().strftime("%Y-%m")
computed = 0
@@ -351,8 +368,9 @@ def calculate_dynamic_thresholds(db: Session = Depends(get_db)):
dynamic_yellow = f">={mean_val:.2f}"
dynamic_red = f"<{mean_val:.2f}"
# 检查是否已有缓存
# 检查是否已有缓存（同企业内）
existing = db.query(DynamicThresholdCache).filter(
DynamicThresholdCache.entity_id == entity_id,
DynamicThresholdCache.kpi_id == kpi.id,
DynamicThresholdCache.period == current_period,
).first()
@@ -365,6 +383,7 @@ def calculate_dynamic_thresholds(db: Session = Depends(get_db)):
existing.dynamic_red = dynamic_red
else:
cache = DynamicThresholdCache(
entity_id=entity_id,
kpi_id=kpi.id,
period=current_period,
mean_value=mean_val,
@@ -400,7 +419,8 @@ def _check_static(value: float, params: dict, kpi) -> tuple:
return ("green", f"[静态] {kpi.kpi_name}={value}, 绿灯{green}")
elif _eval_threshold(value, yellow):
return ("yellow", f"[静态] {kpi.kpi_name}={value}, 黄灯{yellow}")
elif red and _eval_threshold(value, red, invert=True):
elif red and _eval_threshold(value, red):
# red 阈值字面即命中条件（如 "<600" = 低于600触发红灯；">25" = 高于25触发红灯）
return ("red", f"[静态] {kpi.kpi_name}={value}, 红灯{red}")
return (None, None)
@@ -469,16 +489,17 @@ def _eval_threshold(value: float, threshold_str: str, invert: bool = False) -> b
try:
if threshold_str.startswith(">="):
limit = float(threshold_str[2:])
return value >= limit if not invert else value >= limit
# invert=True 时取反：命中 = 值低于阈值（低于下限触发红灯等场景）
return value < limit if invert else value >= limit
elif threshold_str.startswith("<="):
limit = float(threshold_str[2:])
return value <= limit if not invert else value <= limit
return value > limit if invert else value <= limit
elif threshold_str.startswith(">"):
limit = float(threshold_str[1:])
return value > limit if not invert else value > limit
return value <= limit if invert else value > limit
elif threshold_str.startswith("<"):
limit = float(threshold_str[1:])
return value < limit if not invert else value < limit
return value >= limit if invert else value < limit
else:
return False
except (ValueError, TypeError):
@@ -489,13 +510,14 @@ def _eval_threshold(value: float, threshold_str: str, invert: bool = False) -> b
# 预测值检查 + 情景建议
# ============================================================
def _check_forecast_alerts(db: Session) -> int:
"""检查未来7天预测值是否超限 — 针对trigger_on='forecast'和'both'的规则"""
def _check_forecast_alerts(db: Session, entity_id: int = 1) -> int:
"""检查未来7天预测值是否超限 — 针对trigger_on='forecast'和'both'的规则（账套隔离 2026-08-31）"""
from app.utils.cash_forecast_engine import forecast_cash_flow, generate_scenario_suggestion
from app.models import CashForecast
rules = db.query(AlertRule).filter(
AlertRule.enabled == 1,
AlertRule.entity_id == entity_id,
AlertRule.trigger_on.in_(["forecast", "both"]),
).all()
@@ -512,11 +534,10 @@ def _check_forecast_alerts(db: Session) -> int:
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == rule.kpi_id).first()
if kpi:
rule_kpi_cache[rule.kpi_id] = kpi
if not kpi:
if not kpi or kpi.entity_id != entity_id:
continue
entity_id = kpi.entity_id or 1
# 获取最新的预测
# 获取最新的预测（按规则所属企业）
latest_forecasts = db.query(CashForecast).filter(
CashForecast.entity_id == entity_id,
).order_by(CashForecast.forecast_date.asc()).limit(7).all()
@@ -562,6 +583,7 @@ def _check_forecast_alerts(db: Session) -> int:
).first()
if not existing:
alert = KPIAlert(
entity_id=entity_id,
kpi_id=rule.kpi_id,
alert_level=alert_level,
alert_message=alert_message,
@@ -580,18 +602,19 @@ def _check_forecast_alerts(db: Session) -> int:
@router.post("/check-forecast")
def run_forecast_alert_check(db: Session = Depends(get_db)):
"""执行预测值预警检查 — 检查未来7天预测值是否超限"""
generated = _check_forecast_alerts(db)
def run_forecast_alert_check(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""执行预测值预警检查 — 检查未来7天预测值是否超限（账套隔离 2026-08-31）"""
generated = _check_forecast_alerts(db, entity_id=entity_id)
return {"message": f"预测值预警检查完成: 生成{generated}条", "generated": generated}
@router.post("/generate-suggestions")
def generate_alert_suggestions(db: Session = Depends(get_db)):
"""为所有未处理的预警生成情景建议"""
def generate_alert_suggestions(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""为当前企业所有未处理的预警生成情景建议（账套隔离 2026-08-31）"""
from app.utils.cash_forecast_engine import generate_scenario_suggestion
pending = db.query(KPIAlert).filter(
KPIAlert.entity_id == entity_id,
KPIAlert.status == "pending",
KPIAlert.suggestion.is_(None),
).all()
@@ -681,6 +704,7 @@ def run_forecast_deviation_check(
existing.alert_level = alert_level
else:
db.add(KPIAlert(
entity_id=entity_id,
kpi_id=rule.kpi_id,
kpi_value_id=actual.id,
alert_level=alert_level,