From 477515c4536812258248063687915ae0f9d342c2 Mon Sep 17 00:00:00 2001 From: wecom-fullstack Date: Wed, 23 Sep 2026 18:29:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(cma):=20=E9=9A=94=E7=A6=BB=E6=89=AB?= =?UTF-8?q?=E5=B0=BE=20=E2=80=94=20kpi=5Falert=5Fhigher=5Fbetter=20?= =?UTF-8?q?=E6=8B=86=E8=B4=A6=E5=A5=97=20+=20tenant/switch=20=E5=AE=A1?= =?UTF-8?q?=E8=AE=A1=E6=94=B6=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 任务 dama-entity-isolation-sweep-20260923（DAMA 独立回查漏项 / 任总 2026-09-23「按照建议执行」） ① 补漏同源全局键 kpi_alert_higher_better（system_configs，零 DDL） - deviation_engine: 新增 higher_better_key(entity_id)；get_higher_better_codes(db, entity_id) 改为只读本账套键、不回退旧全局键；未配置 → DEFAULT_HIGHER_BETTER（兜底，与原行为逐字一致） - budget.py: GET/PUT /alert-direction 挂 get_entity_id，只读写会话账套键；body entity_id 不信 - scripts/migrate_higher_better_key_entity_20260923.py: --check/--apply/--down + 库名保护 + 幂等 - 生产实况：库内该键 0 行 ⇒ 无原值可搬运（--apply 走 skip 零写入），e1 行为不变 ② kpi_visibility 口径裁定：角色级天然全局 = 非账套维度，不拆（关闭本项） - auth_middleware 注释 + docstring 登记「有意不做账套后缀」；口径字典 docs/global-config-keys-caliber-20260923.md（含 budget_mode/budget_driver_mode 评估结论） ③ tenant/switch 审计收口 - detail 显式 from_entity_id → to_entity_id + source + operator{id,username,name,role} - 审计失败不静默：同事务回滚 + AUDIT_WRITE_FAILED ERROR + HTTP 500（不留「改了但没审计」） - GET /current 无写路径不落审计；历史缺口不补写 证据：新增/改写用例 17 条；全量 1397 passed（去新增文件 1380 + 17）；RED 态 8 failed --- backend/app/api/budget.py | 35 +- backend/app/api/tenant.py | 64 ++- backend/app/auth_middleware.py | 18 +- backend/app/utils/deviation_engine.py | 33 +- ...inspect_entity_isolation_sweep_20260923.py | 92 +++++ ...grate_higher_better_key_entity_20260923.py | 244 +++++++++++ backend/tests/test_budget_tech_improve.py | 3 +- .../test_entity_isolation_sweep_20260923.py | 380 ++++++++++++++++++ ...audit-and-notification-caliber-20260922.md | 23 +- docs/global-config-keys-caliber-20260923.md | 106 +++++ 10 files changed, 965 insertions(+), 33 deletions(-) create mode 100644 backend/scripts/inspect_entity_isolation_sweep_20260923.py create mode 100644 backend/scripts/migrate_higher_better_key_entity_20260923.py create mode 100644 backend/tests/test_entity_isolation_sweep_20260923.py create mode 100644 docs/global-config-keys-caliber-20260923.md diff --git a/backend/app/api/budget.py b/backend/app/api/budget.py index ab82bd02..d25019d6 100644 --- a/backend/app/api/budget.py +++ b/backend/app/api/budget.py @@ -1079,17 +1079,24 @@ def get_deviation_alert_attribution( @router.get("/alert-direction") def get_alert_direction( db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), ): - """越高越好型KPI方向配置 (P2-⑤ 2026-08-28: system_configs 可维护)""" - from app.utils.deviation_engine import get_higher_better_codes, CONFIG_KEY_HIGHER_BETTER + """越高越好型KPI方向配置（**按会话账套**；2026-09-23 拆账套，零 DDL） + + 与 B1 `cash.current_balance` 同款：只读本账套键 `kpi_alert_higher_better.`， + 本账套未配置 → 走 deviation_engine 内置默认表（兜底，不跨账套借用）。 + """ + from app.utils.deviation_engine import get_higher_better_codes, higher_better_key from app.models import SystemConfig + key = higher_better_key(entity_id) cfg = db.query(SystemConfig).filter( - SystemConfig.config_key == CONFIG_KEY_HIGHER_BETTER + SystemConfig.config_key == key ).first() - codes = get_higher_better_codes(db) + codes = get_higher_better_codes(db, entity_id) return { - "config_key": CONFIG_KEY_HIGHER_BETTER, + "config_key": key, + "entity_id": entity_id, "codes": codes, "is_configured": bool(cfg and cfg.config_value), } @@ -1100,10 +1107,15 @@ def update_alert_direction( data: dict, db: Session = Depends(get_db), current_user=Depends(require_auth), + entity_id: int = Depends(get_entity_id), ): - """维护越高越好型KPI方向配置 (P2-⑤) body: {codes: ["SALES_TOTAL", ...]}""" + """维护越高越好型KPI方向配置 (P2-⑤) body: {codes: ["SALES_TOTAL", ...]} + + 账套隔离（2026-09-23）：只写本账套键 `kpi_alert_higher_better.`； + body 里的 `entity_id` 一律不信（会话账套为准，同 B1 口径）。 + """ import json as _json - from app.utils.deviation_engine import CONFIG_KEY_HIGHER_BETTER + from app.utils.deviation_engine import higher_better_key from app.models import SystemConfig codes = data.get("codes") @@ -1111,19 +1123,20 @@ def update_alert_direction( raise HTTPException(400, "codes 必须是非空数组") codes = [str(c) for c in codes] + key = higher_better_key(entity_id) cfg = db.query(SystemConfig).filter( - SystemConfig.config_key == CONFIG_KEY_HIGHER_BETTER + SystemConfig.config_key == key ).first() if cfg: cfg.config_value = _json.dumps(codes, ensure_ascii=False) else: db.add(SystemConfig( - config_key=CONFIG_KEY_HIGHER_BETTER, + config_key=key, config_value=_json.dumps(codes, ensure_ascii=False), - description="越高越好型KPI编码列表（实际低于预算才告警）", + description=f"越高越好型KPI编码列表（实际低于预算才告警）— 账套 {int(entity_id)}", )) db.commit() - return {"success": True, "config_key": CONFIG_KEY_HIGHER_BETTER, "codes": codes} + return {"success": True, "config_key": key, "entity_id": entity_id, "codes": codes} @router.put("/deviation-alerts/{alert_id}") diff --git a/backend/app/api/tenant.py b/backend/app/api/tenant.py index 727e4c04..aae235ba 100644 --- a/backend/app/api/tenant.py +++ b/backend/app/api/tenant.py @@ -24,6 +24,15 @@ 无 `entity_id` 列，无需 DDL）。两字面**刻意不合并**：一个改「用户自己的会话账套」、 一个改「平台全局状态记录」，合并后审计上不可分辨。口径登记见 `docs/audit-and-notification-caliber-20260922.md`。 + 7. **审计缺口的收口（2026-09-23 dama-entity-isolation-sweep-20260923 §二）**： + - `detail` 显式给出 `from_entity_id → to_entity_id`（原先只有 `previous` 整包， + 要人肉解析才知道「从哪来」）+ `source` + `operator`（id/username/role）。 + - **审计失败不静默**：状态键与审计行走**同一事务**，审计写入异常 → `rollback()` + （状态变更一并回滚，绝不留「改了但没审计」的窗口）+ `logger.error(AUDIT_WRITE_FAILED…)` + + HTTP 500。日志前缀 `AUDIT_WRITE_FAILED` 供运维 grep/告警链订阅。 + - `GET /current` **无写路径**（纯只读，且已限 it/ceo），故不落审计——不给读操作 + 写审计表，避免审计面被读流量稀释。 + - 历史缺口（2026-09-22 止血操作无审计行）**不补写**：审计不得事后伪造。 """ from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel @@ -32,6 +41,10 @@ from app.database import get_db from app.auth_middleware import require_auth_401, require_role from app.models import SystemConfig, Entity, OperationLog +import logging + +logger = logging.getLogger("cma.tenant") + # 用户端点：统一要求会话 Bearer 鉴权（cma-api-auth-surface-audit-20260914） # 缺口：本 router 原无鉴权——无 token 可读当前租户、并可 POST /switch 改写 # system_config.current_tenant（状态改写面）。收口为 require_auth_401。 @@ -82,11 +95,14 @@ def switch_tenant(data: TenantSwitchRequest, db: Session = Depends(get_db), raise HTTPException(404, "企业不存在") import json + from_entity_id = None previous = None cfg = db.query(SystemConfig).filter(SystemConfig.config_key == TENANT_KEY).first() if cfg and cfg.config_value: try: previous = json.loads(cfg.config_value) + if isinstance(previous, dict): + from_entity_id = previous.get("entity_id") except Exception: previous = None @@ -112,19 +128,41 @@ def switch_tenant(data: TenantSwitchRequest, db: Session = Depends(get_db), # 2026-09-22 DAMA 回查 §4b 后补：user_id 记操作人（原 None ⇒ 无法追人）； # target_type/target_id 与 /auth/switch-entity 的 switch_entity 对齐 ⇒ 按账套可过滤 # （operation_logs 无 entity_id 列，无须 DDL）。 - db.add(OperationLog( - user_id=user.id, - action="switch_tenant_state", - target_type="entity", - target_id=data.entity_id, - detail={ - "entity_id": data.entity_id, - "entity_name": ent.name, - "source": data.source, - "previous": previous, - "note": "current_tenant 状态记录变更（非数据维度）", + # 2026-09-23 收口：detail 显式 from→to / source / operator；审计失败不静默（见 §7）。 + audit_detail = { + "from_entity_id": from_entity_id, + "to_entity_id": data.entity_id, + "entity_id": data.entity_id, # 兼容既有消费方/用例（= to） + "entity_name": ent.name, + "source": data.source, + "operator": { + "user_id": user.id, + "username": getattr(user, "username", None), + "name": getattr(user, "name", None), + "role": getattr(user, "role", None), }, - )) - db.commit() + "previous": previous, + "note": "current_tenant 状态记录变更（非数据维度）", + } + try: + db.add(OperationLog( + user_id=user.id, + action="switch_tenant_state", + target_type="entity", + target_id=data.entity_id, + detail=audit_detail, + )) + # 状态键与审计行同一事务提交：审计写不进 ⇒ 状态变更一并回滚 + # （绝不留「改了但没审计」的窗口；失败必须可见，不得静默） + db.commit() + except Exception: + db.rollback() + logger.error( + "AUDIT_WRITE_FAILED tenant/switch from=%s to=%s user_id=%s source=%s " + "→ 审计写入失败，状态变更已回滚（无审计不得留变更）", + from_entity_id, data.entity_id, getattr(user, "id", None), data.source, + exc_info=True, + ) + raise HTTPException(status_code=500, detail="审计写入失败，状态变更已回滚") return {"success": True, "current": state} diff --git a/backend/app/auth_middleware.py b/backend/app/auth_middleware.py index 20340e2a..f6e8f9ae 100644 --- a/backend/app/auth_middleware.py +++ b/backend/app/auth_middleware.py @@ -244,9 +244,25 @@ def has_action(user: User, action: str, db: Session = None) -> bool: # ─── KPI可见性（按维度/分类过滤） ─── +# +# ⚠️ 口径裁定（2026-09-23 dama-entity-isolation-sweep-20260923 §一②）： +# `role_permissions.key = 'kpi_visibility'` **属「非账套维度」的全局配置，不拆账套** +# —— 理由：① 值本身是 `{role: [维度规则]}` 的**角色维度**字典，无账套语义； +# ② 角色枚举（ceo/finance/business/it）是平台级字典，`role_permissions` 表 +# 亦无 `entity_id` 列（`key` unique），「按角色看哪些维度」是集团统一口径； +# ③ 与 `route_permissions` / `action_permissions` 同族（同为角色级权限面）， +# 若只拆本键会与同表另两键口径分裂。 +# 判别口径：**同表同族的键按同一维度处理**；本键不消费任何账套状态、也不被账套写侧改写。 +# 登记：`docs/global-config-keys-caliber-20260923.md`（键维度字典）。 + def _load_kpi_visibility(db: Session = None): - """从RolePermission表加载kpi_visibility配置（独立缓存）""" + """从RolePermission表加载kpi_visibility配置（独立缓存） + + **非账套维度**（角色级天然全局，口径见上方注释与本档登记）： + 只读 `role_permissions.key = 'kpi_visibility'`，**有意不做账套后缀**—— + 这是集团共享口径，不是漏隔离。 + """ import time if not hasattr(_load_kpi_visibility, "_cache"): _load_kpi_visibility._cache = {"data": None, "ts": 0} diff --git a/backend/app/utils/deviation_engine.py b/backend/app/utils/deviation_engine.py index 39481b0a..daf152bb 100644 --- a/backend/app/utils/deviation_engine.py +++ b/backend/app/utils/deviation_engine.py @@ -32,10 +32,32 @@ DEFAULT_HIGHER_BETTER = [ CONFIG_KEY_HIGHER_BETTER = "kpi_alert_higher_better" -def get_higher_better_codes(db, entity_id: int = None) -> list: - """读取越高越好型KPI编码列表（system_configs 可维护，无配置回落默认）""" +def higher_better_key(entity_id: int) -> str: + """账套维度的方向配置键名：`kpi_alert_higher_better.`（零 DDL） + + entity_id 必传：`None` → ValueError（防「漏传 → 落默认账套」重演污染）。 + """ + if entity_id is None: + raise ValueError( + "higher_better_key 必须传 entity_id（账套隔离，2026-09-23 " + "dama-entity-isolation-sweep-20260923）" + ) + return f"{CONFIG_KEY_HIGHER_BETTER}.{int(entity_id)}" + + +def get_higher_better_codes(db, entity_id: int) -> list: + """读取**本账套**越高越好型KPI编码列表（entity_id 必传，漏传 TypeError） + + 账套隔离（2026-09-23 dama-entity-isolation-sweep-20260923 §一①，零 DDL）： + · 只读 `kpi_alert_higher_better.`，**不做「回退旧全局键」**—— + 回退分支本身就是跨账套借用通路（与 `cash.current_balance` 同源缺陷）。 + · 本账套未配置 → `DEFAULT_HIGHER_BETTER`（业务兜底：方向判定的内置默认表； + 与拆键前「全局键无行」时的行为逐字一致，故对现存账套零行为差异）。 + · 旧全局键 `kpi_alert_higher_better` 原样保留作回退杠杆 + （搬运脚本 `--down` 只删新键，回退老码即可继续读旧键）。 + """ cfg = db.query(SystemConfig).filter( - SystemConfig.config_key == CONFIG_KEY_HIGHER_BETTER + SystemConfig.config_key == higher_better_key(entity_id) ).first() if cfg and cfg.config_value: try: @@ -43,7 +65,7 @@ def get_higher_better_codes(db, entity_id: int = None) -> list: if isinstance(codes, list): return [str(c) for c in codes] except Exception: - logger.warning("system_configs[%s] 解析失败, 回落默认", CONFIG_KEY_HIGHER_BETTER) + logger.warning("system_configs[%s] 解析失败, 回落默认", higher_better_key(entity_id)) return list(DEFAULT_HIGHER_BETTER) @@ -301,7 +323,8 @@ def build_deviation_alert(db, kpi, period: str, entity_id: int = 1, min_rate: fl actual = deviation.get("actual_value") budget = deviation.get("budget_value") - # 方向性：越高越好型（配置化，system_configs.kpi_alert_higher_better） + # 方向性：越高越好型（配置化，本账套键 system_configs.kpi_alert_higher_better.； + # 2026-09-23 拆账套：entity_id 由调用方必传，不再读全局唯一键） higher_better = kpi.kpi_code in get_higher_better_codes(db, entity_id) if higher_better: diff --git a/backend/scripts/inspect_entity_isolation_sweep_20260923.py b/backend/scripts/inspect_entity_isolation_sweep_20260923.py new file mode 100644 index 00000000..8a555c72 --- /dev/null +++ b/backend/scripts/inspect_entity_isolation_sweep_20260923.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""只读侦查（生产读，本单 dama-entity-isolation-sweep-20260923 §一/§二 前置） + +只 SELECT，无任何写。用于： + 1. system_configs / role_permissions 全量键实况（含 kpi_alert_higher_better / budget_mode / + budget_driver_mode / current_tenant / cash.current_balance.* / alignment_config.*） + 2. operation_logs 实况（action 分布、switch_entity / switch_tenant_state 行数、最近行） + 3. entities 实况（账套清单） + 4. kpi_definitions 列面（是否有自身方向字段）+ 逐账套 KPI 分布 +用法: python3 [--json out.json] +""" +import json +import os +import sys +from pathlib import Path + +BACKEND = Path("/srv/hermes/shared/cma-management/backend") +os.chdir(BACKEND) +sys.path.insert(0, str(BACKEND)) + +from sqlalchemy import text # noqa: E402 +from app.database import get_engine # noqa: E402 + +out = {} +eng = get_engine() +with eng.connect() as c: + out["db"] = c.execute(text("SELECT DATABASE()")).scalar() + out["system_configs"] = [ + {"id": r[0], "config_key": r[1], "config_value": r[2], "description": r[3], + "updated_at": str(r[4])} + for r in c.execute(text( + "SELECT id, config_key, config_value, description, updated_at " + "FROM system_configs ORDER BY config_key")) + ] + out["role_permissions"] = [ + {"id": r[0], "key": r[1], "value": r[2], "updated_at": str(r[3])} + for r in c.execute(text("SELECT id, `key`, `value`, updated_at FROM role_permissions ORDER BY `key`")) + ] + out["entities"] = [ + {"id": r[0], "name": r[1], "short_name": r[2], "status": r[3]} + for r in c.execute(text("SELECT id, name, short_name, status FROM entities ORDER BY id")) + ] + out["op_logs_actions"] = [ + {"action": r[0], "n": r[1], "last": str(r[2])} + for r in c.execute(text( + "SELECT action, COUNT(*) n, MAX(created_at) last FROM operation_logs " + "GROUP BY action ORDER BY n DESC")) + ] + out["op_logs_switch_rows"] = [ + {"id": r[0], "user_id": r[1], "action": r[2], "target_type": r[3], "target_id": r[4], + "detail": (r[5] or "")[:300], "created_at": str(r[6])} + for r in c.execute(text( + "SELECT id, user_id, action, target_type, target_id, detail, created_at " + "FROM operation_logs WHERE action LIKE '%switch%' OR action LIKE '%tenant%' " + "ORDER BY id DESC LIMIT 10")) + ] + out["op_logs_recent"] = [ + {"id": r[0], "user_id": r[1], "action": r[2], "target_type": r[3], "target_id": r[4], + "created_at": str(r[5])} + for r in c.execute(text( + "SELECT id, user_id, action, target_type, target_id, created_at " + "FROM operation_logs ORDER BY id DESC LIMIT 12")) + ] + out["system_configs_cols"] = [ + {"col": r[0], "type": r[1], "null": r[2], "key": r[3], "default": str(r[4])} + for r in c.execute(text( + "SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_KEY, COLUMN_DEFAULT " + "FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() " + "AND TABLE_NAME IN ('system_configs','role_permissions','operation_logs') " + "ORDER BY TABLE_NAME, ORDINAL_POSITION")) + ] + out["kpi_def_cols"] = [ + r[0] for r in c.execute(text( + "SELECT COLUMN_NAME FROM information_schema.COLUMNS " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='kpi_definitions' " + "ORDER BY ORDINAL_POSITION")) + ] + out["kpi_by_entity"] = [ + {"entity_id": r[0], "n": r[1]} for r in c.execute(text( + "SELECT entity_id, COUNT(*) n FROM kpi_definitions GROUP BY entity_id ORDER BY entity_id")) + ] + out["budgetplan_by_entity"] = [ + {"entity_id": r[0], "n": r[1]} for r in c.execute(text( + "SELECT kd.entity_id, COUNT(*) n FROM budget_plans bp " + "JOIN kpi_definitions kd ON kd.id=bp.kpi_id GROUP BY kd.entity_id ORDER BY kd.entity_id")) + ] + +print(json.dumps(out, ensure_ascii=False, indent=1, default=str)) +if "--json" in sys.argv: + p = sys.argv[sys.argv.index("--json") + 1] + Path(p).write_text(json.dumps(out, ensure_ascii=False, indent=1, default=str)) + print("WROTE", p) diff --git a/backend/scripts/migrate_higher_better_key_entity_20260923.py b/backend/scripts/migrate_higher_better_key_entity_20260923.py new file mode 100644 index 00000000..ab326c44 --- /dev/null +++ b/backend/scripts/migrate_higher_better_key_entity_20260923.py @@ -0,0 +1,244 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""全局键拆账套 — `kpi_alert_higher_better` 一次性搬运（数据迁移，非 DDL） + +任务：`dama-entity-isolation-sweep-20260923` §一①（DAMA 独立回查新增漏项，任总 2026-09-23 批） + +背景 +---- +`system_configs.config_key` 为 **unique 单行键** ⇒ `kpi_alert_higher_better` 全库只有一行， +被所有账套共用。消费点（拆键前，读的都是那唯一一行）： + + backend/app/utils/deviation_engine.py get_higher_better_codes() L32/L37-38 + build_deviation_alert() L305 + backend/app/api/budget.py GET /alert-direction L1087 + PUT /alert-direction L1114-1125 + +任一套账写一次，**方向判定（实际低于预算才告警 vs 实际超预算才告警）全平台翻转**。 + +治本（方案 A，零 DDL）：键名带账套后缀 → `kpi_alert_higher_better.` + · 读：只读本账套键；本账套未配置 → `DEFAULT_HIGHER_BETTER`（内置默认表，兜底）， + **不回退旧全局键**（回退即跨账套借用）。 + · 写：`PUT /alert-direction` 只写会话账套键，body 里的 entity_id 不信。 + +本脚本 = 旧值一次性搬运。 + +⚠️ 生产实况（2026-09-23 只读核对）：**库内无 `kpi_alert_higher_better` 行** + ⇒ `--check` 报 `legacy=None`，`--apply` 走 `skip`（无可搬运值），零写入。 + 即：拆键后各账套行为与拆键前**逐字一致**（都走内置默认表），不存在「其余账套留空 + 导致行为退化」的问题。若将来出现旧全局行（例如先于本单部署的 PUT 写入过）， + 本脚本可直接把它落到 `--target-entity`。 + +用法 +---- + # 只读预检（默认动作，零写入） + python3 scripts/migrate_higher_better_key_entity_20260923.py --check + + # 写库（**需审批 / 生产写**；--apply 才写） + python3 scripts/migrate_higher_better_key_entity_20260923.py --apply --target-entity 1 + + # 回退（删新键；旧全局键原样保留） + python3 scripts/migrate_higher_better_key_entity_20260923.py --down + +退出码 +------ + 0 成功 ｜ 1 执行失败 ｜ 2 用法/保护拒绝（库名不符、目标账套不存在） + +⚠️ 本脚本写的是**生产业务库**：不得在开发级会话直接 `--apply`。 + 依据《生产资产派单纪律》：碰生产资产一律走派单 + 审批；`--check` 同样是生产读，须留单痕。 +""" +import argparse +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sqlalchemy import text # noqa: E402 + +KEY = "kpi_alert_higher_better" # 旧全局键（保留不删） +PROD_DB_NAME = "cma" # 生产库名（保护口径） +DESC = "越高越好型KPI编码列表（实际低于预算才告警）— 账套 {eid}" + + +def _engine(): + from app.database import get_engine + return get_engine() + + +def _scalar(conn, sql, **params): + row = conn.execute(text(sql), params).fetchone() + return row[0] if row else None + + +def _detect_db_name(conn) -> str: + try: + return _scalar(conn, "SELECT DATABASE()") or "" + except Exception: + return "" + + +def _entity_evidence(conn): + """逐账套只读证据：KPI 定义数 / 预算行数（供「原值该落哪家」人工确认）""" + out = {} + try: + kpi_rows = conn.execute(text( + "SELECT entity_id, COUNT(*) FROM kpi_definitions GROUP BY entity_id ORDER BY entity_id" + )).fetchall() + bud_rows = conn.execute(text( + "SELECT kd.entity_id, COUNT(*) FROM budget_plans bp " + "JOIN kpi_definitions kd ON kd.id = bp.kpi_id GROUP BY kd.entity_id ORDER BY kd.entity_id" + )).fetchall() + kpi = {int(r[0]) for r in kpi_rows if r[0] is not None} + bud = {int(r[0]): int(r[1]) for r in bud_rows if r[0] is not None} + for eid in sorted(kpi | set(bud)): + out[eid] = {"budget_plans": bud.get(eid, 0)} + except Exception as e: # 表缺失/权限 → 只报告，不阻断 + out["_error"] = str(e)[:300] + return out + + +def _snapshot(conn): + snap = { + "db": _detect_db_name(conn), + "legacy_key": KEY, + "legacy": None, + "new_keys": {}, + "entities": [], + "entity_evidence": _entity_evidence(conn), + "suggested_target_entity": None, + } + row = conn.execute(text( + "SELECT config_value, description, updated_at FROM system_configs WHERE config_key = :k" + ), {"k": KEY}).fetchone() + if row: + snap["legacy"] = {"config_value": row[0], "description": row[1], "updated_at": str(row[2])} + + for r in conn.execute(text( + "SELECT id, name, short_name, status FROM entities ORDER BY id" + )).fetchall(): + snap["entities"].append({"id": int(r[0]), "name": r[1], "short_name": r[2], "status": r[3]}) + + for r in conn.execute(text( + "SELECT config_key, config_value FROM system_configs WHERE config_key LIKE :p ORDER BY config_key" + ), {"p": KEY + ".%"}).fetchall(): + snap["new_keys"][r[0]] = r[1] + + # 「原值落哪家」：旧全局值唯一时，若只有一家账套是 active 生产账套 → 建议它； + # 多 active 账套（e1 酣客 + e2 博海）⇒ 不给自动建议，需人工裁定（写进报告）。 + active = [e["id"] for e in snap["entities"] if e["status"] == "active"] + if snap["legacy"] and len(active) == 1: + snap["suggested_target_entity"] = active[0] + return snap + + +def cmd_check(args): + eng = _engine() + with eng.connect() as conn: + snap = _snapshot(conn) + if args.json: + print(json.dumps(snap, ensure_ascii=False, indent=2)) + return 0 + print(f"目标库: {snap['db']}") + print(f"旧全局键 {KEY} : {snap['legacy']}") + print(f"已存在新键: {snap['new_keys']}") + print("逐账套只读证据（预算行数）:") + for eid, info in snap["entity_evidence"].items(): + print(f" entity {eid}: {info}") + print(f"建议 --target-entity : {snap['suggested_target_entity']}" + f"（None = 无旧值 或 多 active 账套需人工裁定）") + print("【只读预检，零写入】") + return 0 + + +def _apply(conn, target: int, force: bool): + new_key = f"{KEY}.{target}" + row = conn.execute(text( + "SELECT config_value FROM system_configs WHERE config_key = :k" + ), {"k": KEY}).fetchone() + if not row: + return {"key": new_key, "action": "skip", "reason": "旧全局键不存在（无可搬运值）"} + + exist = conn.execute(text( + "SELECT config_value FROM system_configs WHERE config_key = :k" + ), {"k": new_key}).fetchone() + if exist and not force: + return {"key": new_key, "action": "noop", + "reason": "新键已存在且未 --force（保留既有值，幂等）", + "existing_value": exist[0]} + + if exist: + conn.execute(text( + "UPDATE system_configs SET config_value = :v, description = :d WHERE config_key = :k" + ), {"v": row[0], "d": DESC.format(eid=target), "k": new_key}) + return {"key": new_key, "action": "updated", "value": row[0]} + conn.execute(text( + "INSERT INTO system_configs (config_key, config_value, description) VALUES (:k, :v, :d)" + ), {"k": new_key, "v": row[0], "d": DESC.format(eid=target)}) + return {"key": new_key, "action": "inserted", "value": row[0]} + + +def _guard_db(args, conn): + db = _detect_db_name(conn) + if args.allow_db: + return db + if db != PROD_DB_NAME: + print(f"[拒绝] 库名 {db!r} ≠ 生产库 {PROD_DB_NAME!r}；" + f"如确为预期，请显式加 --allow-db {db}", file=sys.stderr) + sys.exit(2) + return db + + +def cmd_apply(args): + eng = _engine() + with eng.begin() as conn: + db = _guard_db(args, conn) + ent = conn.execute(text("SELECT id FROM entities WHERE id = :i"), + {"i": args.target_entity}).fetchone() + if not ent: + print(f"[拒绝] 目标账套 entity_id={args.target_entity} 不存在", file=sys.stderr) + sys.exit(2) + results = [_apply(conn, args.target_entity, args.force)] + out = {"mode": "apply", "db": db, "target_entity": args.target_entity, "results": results} + print(json.dumps(out, ensure_ascii=False, indent=2) if args.json else + "\n".join(f" {r}" for r in results)) + return 0 + + +def cmd_down(args): + """回退：删除新键（旧全局键原样保留，老代码即可继续读）""" + eng = _engine() + with eng.begin() as conn: + db = _guard_db(args, conn) + r = conn.execute(text("DELETE FROM system_configs WHERE config_key LIKE :p"), + {"p": KEY + ".%"}) + removed = [f"system_configs({KEY}.*) deleted={r.rowcount}"] + print(json.dumps({"mode": "down", "db": db, "removed": removed}, + ensure_ascii=False, indent=2) if args.json else "\n".join(removed)) + print("旧全局键未改动 → 代码回退到老版本即可继续读旧键。") + return 0 + + +def main(): + p = argparse.ArgumentParser( + description="全局键拆账套 · 一次性搬运（kpi_alert_higher_better）") + g = p.add_mutually_exclusive_group() + g.add_argument("--check", action="store_true", help="只读预检（默认）") + g.add_argument("--apply", action="store_true", help="写库（需审批）") + g.add_argument("--down", action="store_true", help="回退：删除新键") + p.add_argument("--target-entity", type=int, default=1, + help="旧值落到哪个账套（默认 1 = 陕西酣客，当前生产账套）") + p.add_argument("--force", action="store_true", help="允许覆盖已存在的新键") + p.add_argument("--json", action="store_true", help="机器可读输出") + p.add_argument("--allow-db", default=None, help="生产库名保护豁免（默认仅允许 cma）") + args = p.parse_args() + + if args.apply: + return cmd_apply(args) + if args.down: + return cmd_down(args) + return cmd_check(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_budget_tech_improve.py b/backend/tests/test_budget_tech_improve.py index 0a6d8d2d..150519d5 100644 --- a/backend/tests/test_budget_tech_improve.py +++ b/backend/tests/test_budget_tech_improve.py @@ -407,7 +407,8 @@ class TestP2SingleAlertPath: create_test_user(db) token = get_token_for_user(client) kpi = create_test_kpi(db, kpi_code="HB_KPI", kpi_name="营业收入") - assert "SALES_TOTAL" in get_higher_better_codes(db) + # 2026-09-23 拆账套：签名改为 entity_id 必传（漏传即 TypeError，防落默认账套） + assert "SALES_TOTAL" in get_higher_better_codes(db, 1) class TestP2CashClassify: diff --git a/backend/tests/test_entity_isolation_sweep_20260923.py b/backend/tests/test_entity_isolation_sweep_20260923.py new file mode 100644 index 00000000..7601f5b7 --- /dev/null +++ b/backend/tests/test_entity_isolation_sweep_20260923.py @@ -0,0 +1,380 @@ +"""隔离扫尾 · 同源全局键补漏 + 审计缺口收口 — 2026-09-23 + +任务：`dama-entity-isolation-sweep-20260923` +（DAMA 独立回查新增漏项 / 任总 2026-09-23「按照建议执行」；父单 +`dama-workbench-entity-isolation-fullstack-20260922`） + +三块（逐条对应一个测试类）： + + ① `kpi_alert_higher_better` 拆账套 → `kpi_alert_higher_better.`（零 DDL） + （消费点：`utils/deviation_engine.py::get_higher_better_codes`； + `api/budget.py` GET/PUT `/alert-direction`） + + ② `kpi_visibility`（`role_permissions`）**口径裁定**：角色级天然全局、**非账套维度**， + 不拆；口径写进 `auth_middleware` 注释 + `docs/global-config-keys-caliber-20260923.md` + + ③ `POST /tenant/switch` 审计收口：`detail` 显式 from→to/source/operator； + **审计写失败不静默**（同事务回滚 + `AUDIT_WRITE_FAILED` ERROR + HTTP 500） + +外加：搬运脚本 `scripts/migrate_higher_better_key_entity_20260923.py` 的 +check/apply/down 幂等、「原值只落目标账套」与「库内无旧值 ⇒ skip 零写入」。 + +全部用例只跑内存 SQLite / 临时文件 SQLite（tests/conftest.py 机制）， +**零生产库 / 零生产 Redis 连接**（conftest 已硬阻断）。 +""" +import importlib +import json +import os +import re +import sys + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine, inspect, text +from sqlalchemy.orm import Session + +from app.models import Entity, OperationLog, RolePermission, SystemConfig +from app.database import Base +from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user + +BUDGET = "/api/cma/budget" +TENANT = "/api/cma/tenant" + +LEGACY_KEY = "kpi_alert_higher_better" +SENTINEL_CODE = "ZZ_LEGACY_ONLY_CODE" # 只存在于「旧全局键」的哨兵，用于证伪回退 + + +# ────────────────────────────────────────────── +# 夹具 +# ────────────────────────────────────────────── + +def _ensure_entity(db: Session, eid: int, name: str, status: str = "active") -> Entity: + ent = db.query(Entity).filter(Entity.id == eid).first() + if not ent: + ent = Entity(id=eid, name=name, short_name=name, status=status) + db.add(ent) + db.commit() + db.refresh(ent) + return ent + + +def _seed_two_entities(db: Session): + _ensure_entity(db, 1, "陕西酣客文化传媒") + _ensure_entity(db, 2, "陕西博海网络科技") + + +def _login_as(client: TestClient, username: str, entity_id: int) -> str: + resp = client.post("/api/cma/auth/login", json={ + "username": username, + "password": TEST_USER_PASSWORD, + "entity_id": entity_id, + }) + assert resp.status_code == 200, f"登录 {username}@entity{entity_id} 失败: {resp.text[:300]}" + return resp.json()["token"] + + +def _set_legacy_global(db: Session, codes: list) -> None: + """模拟事故现场：旧全局唯一键有值（拆键前所有账套共用的那一行）""" + cfg = db.query(SystemConfig).filter(SystemConfig.config_key == LEGACY_KEY).first() + val = json.dumps(codes, ensure_ascii=False) + if cfg: + cfg.config_value = val + else: + db.add(SystemConfig(config_key=LEGACY_KEY, config_value=val, + description="历史全局键（事故值）")) + db.commit() + + +# ══════════════════════════════════════════════ +# ① kpi_alert_higher_better 拆账套 +# ══════════════════════════════════════════════ + +class TestHigherBetterPerEntity: + + def test_two_entities_do_not_overwrite_each_other(self, client: TestClient, db: Session): + """e1 / e2 各写方向配置后各自读回本账套值（拆键前共用一个唯一键）""" + _seed_two_entities(db) + create_test_user(db) + t1 = _login_as(client, "testadmin", 1) + t2 = _login_as(client, "testadmin", 2) + + r1 = client.put(f"{BUDGET}/alert-direction", headers=auth_header(t1), + json={"codes": ["SALES_TOTAL", "CUSTOMER_COUNT"]}) + assert r1.status_code == 200, r1.text + r2 = client.put(f"{BUDGET}/alert-direction", headers=auth_header(t2), + json={"codes": ["COST_TOTAL"]}) + assert r2.status_code == 200, r2.text + + v1 = client.get(f"{BUDGET}/alert-direction", headers=auth_header(t1)).json() + v2 = client.get(f"{BUDGET}/alert-direction", headers=auth_header(t2)).json() + assert v1["codes"] == ["SALES_TOTAL", "CUSTOMER_COUNT"], f"e1 配置被覆盖：{v1}" + assert v2["codes"] == ["COST_TOTAL"], f"e2 配置被覆盖：{v2}" + assert v1["entity_id"] == 1 and v2["entity_id"] == 2 + assert v1["config_key"] == f"{LEGACY_KEY}.1" + assert v2["config_key"] == f"{LEGACY_KEY}.2" + + db.expire_all() + keys = {r[0] for r in db.query(SystemConfig.config_key) + .filter(SystemConfig.config_key.like(f"{LEGACY_KEY}%")).all()} + assert keys == {f"{LEGACY_KEY}.1", f"{LEGACY_KEY}.2"}, keys + assert LEGACY_KEY not in keys, "不得再写全局唯一键" + + def test_no_cross_entity_fallback_to_legacy_key(self, client: TestClient, db: Session): + """旧全局键有值（事故值），本账套未配置 → 走内置默认，**不借用**旧全局值""" + _seed_two_entities(db) + create_test_user(db) + _set_legacy_global(db, [SENTINEL_CODE]) # 事故现场 + + t2 = _login_as(client, "testadmin", 2) + body = client.get(f"{BUDGET}/alert-direction", headers=auth_header(t2)).json() + assert SENTINEL_CODE not in body["codes"], \ + f"e2 未配置却读到旧全局键（跨账套借用）：{body}" + assert body["codes"], "未配置须回落内置默认表（兜底语义）" + assert body["is_configured"] is False + + def test_helper_requires_entity_id(self): + """读/写辅助函数的账套参数必传：漏传 → TypeError / ValueError（防落默认账套）""" + from app.utils.deviation_engine import get_higher_better_codes, higher_better_key + with pytest.raises(TypeError): + get_higher_better_codes(None) # 缺 entity_id（位置参数必传） + with pytest.raises(ValueError): + higher_better_key(None) + assert higher_better_key(2) == f"{LEGACY_KEY}.2" + + def test_body_entity_id_not_trusted(self, client: TestClient, db: Session): + """带 token 时 body 里的 entity_id 不可信（仍写会话账套）""" + _seed_two_entities(db) + create_test_user(db) + t1 = _login_as(client, "testadmin", 1) + r = client.put(f"{BUDGET}/alert-direction", headers=auth_header(t1), + json={"codes": ["SALES_TOTAL"], "entity_id": 2}) + assert r.status_code == 200, r.text + assert r.json()["entity_id"] == 1, r.json() + db.expire_all() + assert db.query(SystemConfig).filter( + SystemConfig.config_key == f"{LEGACY_KEY}.2").first() is None + + def test_patch_migrate_scope_unchanged(self): + """读侧不得再出现「回退旧全局键」分支（源级断言，防回归）""" + src = open(os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "app", "utils", "deviation_engine.py"), encoding="utf-8").read() + fn = src[src.index("def get_higher_better_codes"):src.index("# ====", src.index("def get_higher_better_codes"))] + assert "higher_better_key(entity_id)" in fn + assert re.search(r"config_key\s*==\s*CONFIG_KEY_HIGHER_BETTER\b", fn) is None, \ + "读路径不得回退旧全局键（回退即跨账套借用）" + + +# ══════════════════════════════════════════════ +# ② kpi_visibility 口径裁定：非账套维度 +# ══════════════════════════════════════════════ + +class TestKpiVisibilityNonEntityDimension: + """裁定：角色级天然全局（不拆账套）——用结构 + 行为 + 登记三面钉住""" + + def test_role_permissions_has_no_entity_column(self, db: Session): + """结构面：role_permissions 无 entity_id 列（key 唯一）⇒ 天然无账套维度""" + cols = {c["name"] for c in inspect(db.get_bind()).get_columns("role_permissions")} + assert "entity_id" not in cols, cols + + def test_visibility_is_role_keyed_not_entity_keyed(self, db: Session): + """行为面：读的是裸键 `kpi_visibility`（值 = {role: [规则]}），不受账套影响""" + from app import auth_middleware as am + db.add(RolePermission(key="kpi_visibility", + value={"finance": ["finance_*"], "ceo": ["*"]})) + db.commit() + am._load_kpi_visibility._cache = {"data": None, "ts": 0} # 清缓存 + vis = am._load_kpi_visibility(db) + assert vis["finance"] == ["finance_*"] and vis["ceo"] == ["*"] + am._load_kpi_visibility._cache = {"data": None, "ts": 0} + # 不带账套参数也照读（角色维度，无账套语义） + assert am._load_kpi_visibility(db)["finance"] == ["finance_*"] + + def test_caliber_documented_in_code(self): + """登记面：「非账套维度」口径必须写在代码里（否则下轮审计重复报缺口）""" + p = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "app", "auth_middleware.py") + src = open(p, encoding="utf-8").read() + assert "非账套维度" in src and "kpi_visibility" in src + + +# ══════════════════════════════════════════════ +# ③ tenant/switch 审计收口 +# ══════════════════════════════════════════════ + +class TestTenantSwitchAudit: + + def _seed_tenant_key(self, db: Session, entity_id: int, name: str): + db.add(SystemConfig( + config_key="current_tenant", + config_value=json.dumps({"tenant_id": "company_a", "entity_id": entity_id, + "name": name, "short_name": name, + "source": "seed", "switched_at": "2026-09-22T10:36:01"}, + ensure_ascii=False), + description="当前租户状态记录")) + db.commit() + + def test_audit_detail_carries_from_to_source_operator(self, client: TestClient, db: Session): + _seed_two_entities(db) + create_test_user(db) + self._seed_tenant_key(db, 1, "陕西酣客文化传媒") + t1 = _login_as(client, "testadmin", 1) + + r = client.post(f"{TENANT}/switch", headers=auth_header(t1), + json={"entity_id": 2, "source": "sweep-test"}) + assert r.status_code == 200, r.text + + db.expire_all() + log = (db.query(OperationLog) + .filter(OperationLog.action == "switch_tenant_state") + .order_by(OperationLog.id.desc()).first()) + assert log is not None, "写路径必须落审计（修复前无 operation_logs 行）" + assert log.user_id == 1 + assert log.target_type == "entity" and log.target_id == 2 + d = log.detail or {} + assert d.get("from_entity_id") == 1, f"detail 须含 from→to：{d}" + assert d.get("to_entity_id") == 2, d + assert d.get("entity_id") == 2, d # 兼容既有消费方 + assert d.get("source") == "sweep-test", d + op = d.get("operator") or {} + assert op.get("user_id") == 1 and op.get("username") == "testadmin", op + assert op.get("role") == "ceo", op + + def test_audit_write_failure_is_not_silent(self, client: TestClient, db: Session, + monkeypatch, caplog): + """审计写不进去 → 状态变更一并回滚 + ERROR 留痕 + HTTP 500（不静默）""" + import logging + from app.api import tenant as tenant_mod + + _seed_two_entities(db) + create_test_user(db) + self._seed_tenant_key(db, 1, "陕西酣客文化传媒") + t1 = _login_as(client, "testadmin", 1) + + class _Boom: + def __init__(self, *a, **kw): + raise RuntimeError("simulated audit table failure") + + monkeypatch.setattr(tenant_mod, "OperationLog", _Boom) + + with caplog.at_level(logging.ERROR): + r = client.post(f"{TENANT}/switch", headers=auth_header(t1), + json={"entity_id": 2, "source": "sweep-test"}) + + assert r.status_code == 500, f"审计失败必须显式失败，实际 {r.status_code}: {r.text[:200]}" + assert any("AUDIT_WRITE_FAILED" in rec.getMessage() for rec in caplog.records), \ + [rec.getMessage() for rec in caplog.records] + + db.expire_all() + cfg = db.query(SystemConfig).filter(SystemConfig.config_key == "current_tenant").first() + assert json.loads(cfg.config_value)["entity_id"] == 1, \ + "审计失败时状态变更必须回滚（不得留「改了但没审计」）" + assert db.query(OperationLog).filter( + OperationLog.action == "switch_tenant_state").count() == 0 + + def test_get_current_has_no_write_path(self, client: TestClient, db: Session): + """GET /current 纯只读：读一次不产生任何审计行（不给读操作写审计表）""" + _ensure_entity(db, 1, "陕西酣客文化传媒") + create_test_user(db) + t1 = _login_as(client, "testadmin", 1) + assert client.get(f"{TENANT}/current", headers=auth_header(t1)).status_code == 200 + db.expire_all() + assert db.query(OperationLog).count() == 0 + + +# ══════════════════════════════════════════════ +# 搬运脚本（临时文件 SQLite，绝不碰 app.database → 生产库） +# ══════════════════════════════════════════════ + +MIGRATION = "scripts.migrate_higher_better_key_entity_20260923" + + +class TestMigrationScript: + pytestmark = [] + + @pytest.fixture + def mig_env(self, tmp_path, monkeypatch): + sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + mod = importlib.import_module(MIGRATION) + eng = create_engine(f"sqlite:///{tmp_path/'mig.db'}") + Base.metadata.create_all(bind=eng) + with eng.begin() as conn: + conn.execute(text("INSERT INTO entities (id, name, short_name, status) " + "VALUES (1, '酣客', '酣客', 'active')")) + conn.execute(text("INSERT INTO entities (id, name, short_name, status) " + "VALUES (2, '博海', '博海', 'active')")) + conn.execute(text("INSERT INTO system_configs (config_key, config_value, description) " + "VALUES ('kpi_alert_higher_better', '[\"SALES_TOTAL\"]', '历史全局键')")) + monkeypatch.setattr(mod, "_engine", lambda: eng) + return mod, eng + + def _run(self, mod, argv): + old = sys.argv + sys.argv = ["migrate"] + argv + try: + return mod.main() + finally: + sys.argv = old + + def test_check_is_readonly(self, mig_env): + mod, eng = mig_env + assert self._run(mod, ["--check", "--allow-db", "sqlite"]) == 0 + with eng.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key LIKE 'kpi_alert_higher_better.%'")).scalar() == 0 + + def test_apply_moves_value_to_target_only_and_is_idempotent(self, mig_env): + mod, eng = mig_env + assert self._run(mod, ["--apply", "--allow-db", "sqlite", + "--target-entity", "1", "--json"]) == 0 + with eng.connect() as conn: + v = conn.execute(text("SELECT config_value FROM system_configs " + "WHERE config_key = 'kpi_alert_higher_better.1'")).scalar() + assert json.loads(v) == ["SALES_TOTAL"] + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key = 'kpi_alert_higher_better.2'")).scalar() == 0 + assert conn.execute(text("SELECT config_value FROM system_configs " + "WHERE config_key = 'kpi_alert_higher_better'")).scalar() == \ + '["SALES_TOTAL"]' # 旧键保留（回退杠杆） + assert self._run(mod, ["--apply", "--allow-db", "sqlite", + "--target-entity", "1", "--json"]) == 0 # 二跑幂等 + with eng.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key LIKE 'kpi_alert_higher_better.%'")).scalar() == 1 + + def test_apply_without_legacy_row_is_skip_zero_write(self, tmp_path, monkeypatch): + """生产实况：库内无旧值 ⇒ --apply 走 skip、零写入（拆键后行为与拆键前逐字一致）""" + sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + mod = importlib.import_module(MIGRATION) + eng = create_engine(f"sqlite:///{tmp_path/'empty.db'}") + Base.metadata.create_all(bind=eng) + with eng.begin() as conn: + conn.execute(text("INSERT INTO entities (id, name, short_name, status) " + "VALUES (1, '酣客', '酣客', 'active')")) + monkeypatch.setattr(mod, "_engine", lambda: eng) + assert self._run(mod, ["--apply", "--allow-db", "sqlite", "--json"]) == 0 + with eng.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key LIKE 'kpi_alert_higher_better%'")).scalar() == 0 + + def test_down_removes_new_keys_keeps_legacy(self, mig_env): + mod, eng = mig_env + self._run(mod, ["--apply", "--allow-db", "sqlite", "--json"]) + assert self._run(mod, ["--down", "--allow-db", "sqlite", "--json"]) == 0 + with eng.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key LIKE 'kpi_alert_higher_better.%'")).scalar() == 0 + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key = 'kpi_alert_higher_better'")).scalar() == 1 + + def test_prod_db_name_guard(self, mig_env): + mod, eng = mig_env + with pytest.raises(SystemExit) as e: + self._run(mod, ["--apply", "--json"]) # sqlite → DATABASE() 不可用 → db='' + assert e.value.code == 2 + + def test_target_entity_must_exist(self, mig_env): + mod, eng = mig_env + with pytest.raises(SystemExit) as e: + self._run(mod, ["--apply", "--allow-db", "sqlite", "--target-entity", "99", "--json"]) + assert e.value.code == 2 diff --git a/docs/audit-and-notification-caliber-20260922.md b/docs/audit-and-notification-caliber-20260922.md index 1c2fdf03..0e21d5f6 100644 --- a/docs/audit-and-notification-caliber-20260922.md +++ b/docs/audit-and-notification-caliber-20260922.md @@ -32,6 +32,22 @@ 并把 `target_type='entity'` / `target_id=entity_id` 与 `switch_entity` 对齐。 - `detail` JSON 仍保留 `entity_id / entity_name / source / previous`（人读明细）。 +**2026-09-23 收口（`dama-entity-isolation-sweep-20260923` §二）**： + +- `detail` 再补 **显式 from→to** 与操作者三段：`from_entity_id` / `to_entity_id`（原只有 `previous` 整包， + 要人肉解析才知道「从哪来」）、`source`、`operator{user_id,username,name,role}`。 + `entity_id` 字段保留（= `to_entity_id`，兼容既有消费方）。 +- **审计失败不静默**：状态键与审计行**同一事务**提交，审计写入异常 → `rollback()` + （状态变更一并回滚，绝不留「改了但没审计」的窗口）+ `logger.error("AUDIT_WRITE_FAILED …")` + HTTP 500。 + 运维可在 `logs` 里 grep `AUDIT_WRITE_FAILED` 订阅。 +- `GET /tenant/current` **无写路径**（纯只读、已限 it/ceo）⇒ 不落审计（不给读操作写审计表）。 +- 历史缺口（2026-09-22 10:36 止血那次 `POST /tenant/switch` 无审计行）**不补写**：审计不得事后伪造。 + 该次操作的留痕以 ops 止血单 + 本档 §一「现网实况」为准。 + +**在线实测（2026-09-23，`POST /api/cma/tenant/switch` 带会话 Bearer，`from=1 → to=1`）**： +切换一次 → `operation_logs` 出现 `switch_tenant_state` 行（`user_id` 非空、`target_type='entity'`） +⇒ 「全平台影响动作无审计」的缺口已在生产关闭（实测证据见回报件 §二）。 + **现网实况（只读核对，2026-09-22 11:15，`cma_reader` 16/16 PASS）**： `switch_entity` = 49 行、`switch_tenant_state` = 0 行（本单尚未部署 ⇒ 字面是新引入的， 上线前库里无该字面属正常）。ops 止血那次 `POST /tenant/switch`（10:36:01，部署码）**没有**审计行—— @@ -63,5 +79,8 @@ 1. `kpi_alerts` id 589/590 跨账套错挂（`entity_id=1` 挂 e2 的 `BH_ADM_EXPENSE`）：数据订正须 另立单 + 任总放行（生产写）。 2. `kpi_id=1` 兜底残留清点（9 行、其中 1 行 `pending`）与归属裁定。 -3. 同源全局键继续拆账套：`kpi_alert_higher_better`、`kpi_visibility`（DAMA §7 反向挑错新增漏项）、 - `budget_mode` / `budget_driver_mode`（batch5 §七 已记录未施工）。 +3. ~~同源全局键继续拆账套：`kpi_alert_higher_better`、`kpi_visibility`（DAMA §7 反向挑错新增漏项）、 + `budget_mode` / `budget_driver_mode`（batch5 §七 已记录未施工）。~~ + → **2026-09-23 已分流**：`kpi_alert_higher_better` 已拆账套、`kpi_visibility` 裁定「非账套维度」 + 关闭（见 `global-config-keys-caliber-20260923.md` §二/§三）；`budget_mode` / `budget_driver_mode` + 评估为「应拆」但**另立 P2 单**（原值落账需拍板，见同档 §四）。 diff --git a/docs/global-config-keys-caliber-20260923.md b/docs/global-config-keys-caliber-20260923.md new file mode 100644 index 00000000..9a3146a4 --- /dev/null +++ b/docs/global-config-keys-caliber-20260923.md @@ -0,0 +1,106 @@ +# 口径登记：全局配置键的「维度字典」（账套维度 / 非账套维度 / 集团共享） + +- **立档**: wecom-fullstack，2026-09-23 +- **来源**: `dama-entity-isolation-sweep-20260923` §一（DAMA 独立回查新增漏项，任总 2026-09-23 + 「按照建议执行」）；父单 `dama-workbench-entity-isolation-fullstack-20260922` +- **性质**: 口径登记 + 代码收口说明。**不含**任何数据订正（数据订正另立单 + 审批） +- **配套**: 审计字面/通知日志口径见 `audit-and-notification-caliber-20260922.md` + +--- + +## 一、判据（把「全局唯一键」分三档，别再一刀切） + +`system_configs.config_key` 与 `role_permissions.key` 都是 **unique 单行键** ⇒ 一次写入全平台生效。 +并非所有全局键都是缺陷，必须按键的**语义维度**分档： + +| 档 | 名称 | 判据（问：这个值描述的是谁的什么？） | 处置 | +|---|---|---|---| +| **A** | 账套维度 | 值描述**某个账套**的状态，或**按账套业务**取值（余额、方向、期间窗口…） | **必须拆**：键名带 `.`（零 DDL）；读**不回退**旧全局键；写侧 `entity_id` 必传；配 `--check/--apply/--down` 搬运脚本 | +| **B** | 非账套维度 | 值按**角色**（或其它平台级维度）组织，与账套无关 | **保持全局**，但必须在代码里写明「**有意不做账套后缀，这是集团共享口径，不是漏隔离**」并在本档登记 | +| **C** | 集团共享（配置/模板） | 对齐模式模板、通知渠道、OKR 模板、路由/动作权限 | 保持全局 + docstring 声明口径 | + +**不写声明的代价**：下一轮隔离审计会把 B/C 档重复报成缺口，来回返工（本仓已发生两轮）。 + +--- + +## 二、逐键裁定表（2026-09-23 现况） + +| 键 | 载体 | 消费点 | 档 | 处置 | 状态 | +|---|---|---|---|---|---| +| `kpi_alert_higher_better` | `system_configs` | `utils/deviation_engine.py::get_higher_better_codes`（L35-66）、`build_deviation_alert`（L305）；`api/budget.py` GET/PUT `/alert-direction` | **A** | 拆为 `kpi_alert_higher_better.`；本账套未配置 → `DEFAULT_HIGHER_BETTER`（内置兜底） | ✅ 2026-09-23 本单 | +| `cash.current_balance` | `system_configs` | `utils/cash_forecast_engine.py`、`api/cash.py` | A | 同上（B1） | ✅ 2026-09-22 | +| `alignment_config` | `role_permissions` | `api/alignment.py`（GET/POST `/config`、GET `/tree`） | A | 同上（B3a） | ✅ 2026-09-22 | +| `current_tenant` | `system_configs` | `api/tenant.py` | A（特殊） | **语义收敛**：非数据维度，仅项目Bot A2A 状态记录；GET 限 it/ceo；POST 落审计（含 from→to） | ✅ 2026-09-22 / 审计收口 2026-09-23 | +| `kpi_visibility` | `role_permissions` | `auth_middleware.py::_load_kpi_visibility`（L248-286） | **B** | **不拆**；代码注释登记 | ✅ 2026-09-23 关闭 | +| `route_permissions` / `action_permissions` | `role_permissions` | `auth_middleware.py`（路由/动作权限） | C | 集团共享 | ✅ | +| `notification_channels` / `okr_templates` | 各自表 | 通知渠道 / OKR 模板 | C | 集团共享（父单 docstring 已声明） | ✅ | +| `budget_mode` | `system_configs` | `api/budget.py` L509（GET `/config`）、L535（POST `/config`）、L568（`/roll-forward`）、L677（预测期间窗口） | **A（待拆）** | 见 §三 | ⏳ 排期 | +| `budget_driver_mode` | `system_configs` | `api/driver_budget.py` L114（GET `/mode`）、L135（POST `/mode`） | **A（待拆）** | 见 §三 | ⏳ 排期 | + +--- + +## 三、`kpi_visibility` 裁定（本单 §一②，**关闭本项**） + +**结论：属「非账套维度」的全局配置，不拆账套；按 B 档登记后关闭。** + +理由（三条，均以库内/代码实况为据）： + +1. **值本身是角色维度字典**：`{role: [维度规则]}`（默认 `DEFAULT_KPI_VISIBILITY`： + `ceo/it → ["*"]`、`finance → ["finance_*"]`、`business → ["customer_*","process_*","learning_*"]`）， + 条目的键是**角色**，没有任何账套语义。 +2. **载体天然无账套列**：`role_permissions` 只有 `id/key/value/updated_at`，`key` unique + ⇒ 表结构上不存在「按账套存一份」的位置（要按账套分必须 DDL，与「零 DDL 优先」相悖）。 +3. **同表同族一致性**：`route_permissions` / `action_permissions` 同为角色级权限面，都是全局； + 只拆 `kpi_visibility` 会造成同表三种维度口径分裂，反而更难维护。 + +**口径**：同表同族的键按同一维度处理。本键**不消费任何账套状态**，也不被账套写侧改写 +（现库该键 0 行 ⇒ 走内置默认表，与任何账套无关）。 + +**落地**：代码注释 + `_load_kpi_visibility` docstring 已写明「**有意不做账套后缀**」并指向本档； +用例锁「结构面（无 `entity_id` 列）+ 行为面（裸键读值）+ 登记面（口径写在代码里）」。 + +> 现网实况（2026-09-23 只读核对）：`role_permissions` 4 行 = `route_permissions` / +> `action_permissions` / `alignment_config` / `alignment_config.1`；**`kpi_visibility` 无行**。 + +--- + +## 四、`budget_mode` / `budget_driver_mode` 评估结论（本单「顺带」项，**不并入本单**） + +**现况（只读核对）**：库内各 1 行 —— +`budget_mode = {"mode":"rolling","rolling_months":12}`（2026-09-08 更新）、 +`budget_driver_mode = {"mode":"driver","label":"驱动因子模式"}`（2026-08-26 更新）。 + +**为什么属 A 档（不是集团共享）**： + +- `budget_mode` 决定**本账套**预算预测的**期间窗口**（`budget.py:677-702`：`rolling` = 当月起 12 个月、 + `fixed` = 当年 1-12 月），而其数据查询本身**按会话 entity 过滤**（同函数 L710-713） + ⇒ 全局键在决定「本账套的取数窗口」，属跨账套污染同源。 +- `budget_driver_mode` 决定**本账套**预算编制模式（`subject` 科目模式 / `driver` 驱动因子模式）。 +- 任一账套调 `POST /budget/config`（或 `POST /driver-budget/mode`）即**改全平台模式**。 + +**结论：应拆（B1 同款，零 DDL），但不并入本单**，理由三条： + +1. **库内确有旧值，且两个 active 账套都在用预算**（`budget_plans` 逐账套：e1 = 276 行、 + e2 = 25 行）⇒ 必须裁定「原值落哪家」，或「是否双落」。若按单账套搬运（只落 e1）， + e2 会由 `rolling` 退化为默认 `fixed`（期间窗口从「当月+12 月」变「全年 1-12 月」）—— + 这是**跨账套行为变更**，须任总拍板，不能由执行方顺手决定。 +2. **牵动既有用例**：`tests/test_budget.py`（8 处）、`tests/test_cross_entity_isolation_batch2_20260918.py`（2 处）、 + `tests/test_budget_tech_improve.py` 均直接读写该全局键字面 ⇒ 应作为独立单元出 RED→GREEN + 全量回归。 +3. **又是一次生产写**（`system_configs` 新键）⇒ 按「**同一生产写目标只开一张单**」的约定不并入本单。 + +**建议排期**：P2 单「`budget_mode` / `budget_driver_mode` 拆账套（含原值落账裁定）」。 +**建议口径（供拍板）**：**原集团值落全部 active 账套（e1 + e2）、demo 账套留空走默认** —— +即「原值 = 拆键前的事实生效面」（该键拆键前对两个 active 账套同时生效）， +这样拆键**零行为变更**，只是把「谁改谁生效」的写入面收窄到本账套； +比「只落 e1、e2 退化」更保守，也比「读侧回退旧键」更干净（回退仍是借用通路）。 + +--- + +## 五、消费者自查法（防下一次漏扫） + +1. **枚举**：`grep -rn 'SystemConfig' backend/app`＋ + `grep -rhoP 'config_key == "\K[a-zA-Z0-9_.]+' backend/app`（`role_permissions` 同法 grep `RolePermission`）； +2. **分档**：对每个键问「这个值描述的是**谁的**什么」→ A / B / C； +3. **A 档改完必须四件套**：读**不回退**旧键 + 写侧 `entity_id` 必传（不给默认值，漏传即 `TypeError`） + + 本档登记 + 搬运脚本 `--check/--apply/--down`（库名保护 + 幂等 + 目标账套存在性校验）； +4. **B/C 档必须留声明**（代码注释/docstring 指回本档），否则下一轮审计重复报缺口。