Hermes CI Fix
758f820970
- 新增 app/risk_levels.py: RISK_LEVELS定义 + @risk_level装饰器 + API_RISK_MAP (L1只读21 / L2业务写5 / L3批量写3 / L4=0安全底线) - 新增 app/api/audit_log.py: Bot API审计中间件 → backend/logs/bot_audit.log (JSON行: timestamp/bot_name/endpoint/method/risk_level/entity_id/status, L3额外记rows行数, 不阻塞业务) - bot_bridge/bot_bridge_v2/bot_kpis/bot_iron_law 全部29路由标注级别 - 新增 GET /api/cma/bot/risk-levels (X-BOT-KEY鉴权): API→级别→处理方式清单 - main.py 注册审计中间件 - tests/test_risk_levels.py: 覆盖路由标注/risk-levels端点/L4不存在/审计日志
121 lines
4.6 KiB
Python
121 lines
4.6 KiB
Python
"""
|
||
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项开放 = 天然隔离）。
|
||
"""
|
||
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单条业务写） ──
|
||
"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创建（方案清单未列出，按单条业务写归类）
|
||
# ── L3 批量写/创建（3项） ──
|
||
"POST /api/cma/bot/import": "L3",
|
||
"POST /api/cma/bot/kpis/create-with-links": "L3",
|
||
"POST /api/cma/bot-bridge/mpm-result": "L3",
|
||
# ── 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
|