""" CMA Bot API 风险分级(L1-L4)与路由标注装饰器 ============================================= 分级定义(方案文档 nexa-permission-autonomy-plan-20260828 第二节): L1 只读查询 — X-BOT-KEY验证后直接放行 L2 业务写(单条) — 放行 + 写前校验(entity归属/字段校验) L3 批量写/创建 — 放行 + 限制批量 + source_batch审计 L4 危险 — 不向Bot API开放(DROP/TRUNCATE/批量DELETE/生产结构修改) 安全底线: Bot API面不存在L4端点(0项开放 = 天然隔离)。 写操作语义分级(业务语义 low/high,2026-09-03 分层授权落地 P0-②) ============================================================ 依据: CMA交互架构规范 v1 第5.1节 分层授权模型(任总拍板)—— 考核基准类(KPI目标值/预算调整)→ Bot只能"建议",人拍板生效; 监测参数类(红黄绿阈值)→ Bot建议+人批量确认(P1待办,未开放Bot直写); 预算 = 三阶段分权(编制人定框架/执行Bot自动/调整人审批)。 low/high 与 L1-L4 是两套正交分级,勿混: L1-L4 = 技术风险分级(操作面/写前校验/批量约束,本文件API_RISK_MAP) low/high = 业务写语义分级(值是不是"考核基准/资源承诺") 关系: low 类 = 现有 L2/L3 直写端点行为不变(规则型数据写); high 类 = Bot API面不提供直写端点,只提供建议通道端点(L2)。 端点归类清单(low=Bot直写+留痕 / high=建议通道人审批): [low · 直写 KPI实际值/流水/验证记录](写库即生效,靠 audit_log/source_batch 留痕) POST /api/cma/bot/kpi-value-with-check — 写KPI实际值+联动预警 (L2) POST /api/cma/bot-bridge/kpi-result — 写KPI实际值 (source_batch) (L2) POST /api/cma/bot-kpis/{kpi_id}/value — 写KPI当前值 (L2) POST /api/cma/bot/import — Excel批量导入KPI实际值 (L3) POST /api/cma/bot/dict/zxt-account — 智享通科目字典 UPSERT(zxt_account_dict) (L3, 2026-09-12 新增) POST /api/cma/bot/subjects — 会计科目字典 UPSERT(subjects,幂等/绝不DELETE) (L3, 2026-09-12 新增) POST /api/cma/bot-bridge/mpm-result — MPM分析结果→KPI实际值 (L3) POST /api/cma/bot-bridge/verify/{action_plan_id} — 验证ActionPlan执行 (L2, 轻量verify_log) POST /api/cma/bot/okr/create — 创建OKR Objective (L2, 管理类业务写,保留评审: 2026-09-03 P1待办) POST /api/cma/bot/kpis/create-with-links — 创建新KPI定义+因果链 (L3; 2026-09-03收口: 拒绝直写非零 target_value) [high · 建议通道(端点只建 ai_suggestions, status=unapplied, 人 apply 后生效+apply_detail留痕)] POST /api/cma/bot/ai/suggestions — Bot创建高危建议 (L2, 本任务新增) 说明: KPI目标值修改/预算调整 无任何Bot直写端点(grep实证)—— kpi_target/budget_adjust 的生效只经 /api/cma/ai/suggestions/{id}/apply(人token)。 """ import functools RISK_LEVELS = { "L1": "只读", "L2": "业务写", "L3": "批量写", "L4": "危险", } RISK_HANDLING = { "L1": "X-BOT-KEY验证后直接放行", "L2": "放行 + 写前校验(entity归属/字段校验)", "L3": "放行 + 限制批量 + source_batch审计", "L4": "不向Bot API开放(终端层拦截 + 人工审批)", } # ──────────────────────────────────────────────── # API → 风险级别 清单(Bot API面全量路由) # 键格式: "METHOD path"(path 与 FastAPI route.path 一致,含 {param} 占位符) # ──────────────────────────────────────────────── API_RISK_MAP = { # ── L1 只读(20项方案清单 + risk-levels查询端点) ── "GET /api/cma/bot/ping": "L1", "GET /api/cma/bot/overview": "L1", "GET /api/cma/bot/kpis": "L1", "GET /api/cma/bot/kpis/{kpi_id}/history": "L1", "GET /api/cma/bot/strategic-maps": "L1", "GET /api/cma/bot/alerts": "L1", "GET /api/cma/bot/budget/plans": "L1", "GET /api/cma/bot/cost/standard": "L1", "GET /api/cma/bot/cost/actual": "L1", "GET /api/cma/bot/actions": "L1", "GET /api/cma/bot/organization": "L1", "GET /api/cma/bot/data-sources": "L1", "GET /api/cma/bot/users": "L1", "GET /api/cma/bot/query": "L1", "GET /api/cma/bot/okr/list": "L1", "GET /api/cma/bot/nlp": "L1", "GET /api/cma/bot/iron-law": "L1", "GET /api/cma/bot/iron-law/bots": "L1", "GET /api/cma/bot-bridge/verify/{action_plan_id}/history": "L1", "GET /api/cma/bot-kpis": "L1", # 本任务新增的只读端点 "GET /api/cma/bot/risk-levels": "L1", # ── L2 业务写(4项方案清单 + okr/create单条业务写 + bot建议通道) ── "POST /api/cma/bot/kpi-value-with-check": "L2", "POST /api/cma/bot-bridge/kpi-result": "L2", "POST /api/cma/bot-kpis/{kpi_id}/value": "L2", "POST /api/cma/bot-bridge/verify/{action_plan_id}": "L2", "POST /api/cma/bot/okr/create": "L2", # 单条OKR创建(方案清单未列出,按单条业务写归类) "POST /api/cma/bot/ai/suggestions": "L2", # Bot高危建议创建通道(写操作语义 high,见文件头注释;本身只建建议不直写考核基准) # ── L3 批量写/创建(5项) ── "POST /api/cma/bot/import": "L3", "POST /api/cma/bot/kpis/create-with-links": "L3", "POST /api/cma/bot-bridge/mpm-result": "L3", "POST /api/cma/bot/dict/zxt-account": "L3", # 智享通科目字典 UPSERT(2026-09-12 ops-cma-dict-write-endpoint) "POST /api/cma/bot/subjects": "L3", # 会计科目字典 UPSERT(2026-09-12 dama-cma-subjects-write-endpoint) # ── L4 危险:Bot API面不存在(安全底线,不添加) ── } def risk_level(level: str): """路由标注装饰器: @risk_level('L1') 挂在路由函数上(router.get/post 之下)。 同时把级别属性写到原函数与包装函数上,保证 route.endpoint 无论取到哪个 都能通过 getattr(endpoint, 'risk_level') 解析。 """ def decorator(func): func.risk_level = level @functools.wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) wrapper.risk_level = level return wrapper return decorator def get_risk_level(method: str, path: str): """按 METHOD + path(FastAPI模板路径)查级别,未标注返回 None""" return API_RISK_MAP.get(f"{method.upper()} {path}") def get_handling(level: str) -> str: """级别 → 处理方式说明""" return RISK_HANDLING.get(level, "") def list_api_risk_map() -> list: """返回 API→级别→处理方式 清单(供 GET /api/cma/bot/risk-levels 使用)""" items = [] for key, level in API_RISK_MAP.items(): method, path = key.split(" ", 1) items.append({ "method": method, "path": path, "risk_level": level, "handling": get_handling(level), }) items.sort(key=lambda x: (x["risk_level"], x["method"], x["path"])) return items def risk_summary() -> dict: """各级别端点数量统计""" summary = {lv: 0 for lv in RISK_LEVELS} for level in API_RISK_MAP.values(): summary[level] = summary.get(level, 0) + 1 return summary