diff --git a/backend/app/api/audit_log.py b/backend/app/api/audit_log.py new file mode 100644 index 00000000..9754abb9 --- /dev/null +++ b/backend/app/api/audit_log.py @@ -0,0 +1,226 @@ +""" +Bot API 操作审计中间件 +====================== + +每次 Bot API 调用（/api/cma/bot*）记录一行 JSON 审计日志到 +backend/logs/bot_audit.log（可用环境变量 CMA_BOT_AUDIT_LOG 覆盖路径）。 + +JSON 行字段: timestamp / bot_name / endpoint / method / risk_level / + entity_id / status +L3 批量写额外记录 rows（导入行数 / 批量条数）。 + +设计约束: + - 不阻塞业务: 所有日志写入失败仅静默跳过，不影响请求结果 + - 只读 body（starlette 会缓存并回放给下游），仅在 application/json 时解析 + - 不改动任何现有 API 行为与多租户隔离 +""" +import json +import logging +import os +from datetime import datetime + +from fastapi import Request + +_LOG_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), + "logs", +) +_DEFAULT_LOG_PATH = os.path.join(_LOG_DIR, "bot_audit.log") + +_audit_logger = None + + +def _get_logger(): + """构造/复用审计 logger（路径取 CMA_BOT_AUDIT_LOG 覆盖值，便于测试隔离）""" + global _audit_logger + if _audit_logger is None: + log_path = os.getenv("CMA_BOT_AUDIT_LOG") or _DEFAULT_LOG_PATH + os.makedirs(os.path.dirname(log_path), exist_ok=True) + logger = logging.getLogger("cma.bot_audit") + logger.setLevel(logging.INFO) + logger.propagate = False + # 清掉旧 handler，避免日志名单例导致路径切换失效 + for h in list(logger.handlers): + logger.removeHandler(h) + try: + h.close() + except Exception: + pass + handler = logging.FileHandler(log_path, encoding="utf-8") + handler.setFormatter(logging.Formatter("%(message)s")) + logger.addHandler(handler) + _audit_logger = logger + return _audit_logger + + +def write_audit_line(record: dict): + """写一行 JSON 审计日志；任何异常都静默（不阻塞业务）""" + try: + _get_logger().info(json.dumps(record, ensure_ascii=False)) + except Exception: + pass + + +def _resolve_bot_name(request: Request) -> str: + """解析调用方标识: X-BOT-KEY → 配置名/原始key；bridge → bot-bridge；其余 → web-user/unknown""" + key = request.headers.get("X-BOT-KEY") + if key: + try: + from app.api.bot_bridge import _BOT_API_KEYS + info = _BOT_API_KEYS.get(key) + if isinstance(info, dict) and info.get("name"): + return info["name"] + except Exception: + pass + return key + if request.headers.get("X-BRIDGE-TOKEN"): + return "bot-bridge" + if (request.headers.get("Authorization") or "").startswith("Bearer "): + return "web-user" + return "unknown" + + +def _resolve_risk_level(request: Request, route=None, endpoint=None): + """级别解析顺序: 路由函数装饰器标注 → API_RISK_MAP(route.path) → API_RISK_MAP(请求URL)""" + if endpoint is not None: + level = getattr(endpoint, "risk_level", None) + if level: + return level + if route is not None: + from app.risk_levels import get_risk_level + path = getattr(route, "path", None) + if path: + for m in getattr(route, "methods", set()) or set(): + if m in ("GET", "POST", "PUT", "DELETE", "PATCH"): + level = get_risk_level(m, path) + if level: + return level + try: + from app.risk_levels import get_risk_level + return get_risk_level(request.method, request.url.path) + except Exception: + return None + + +async def _extract_entity_id(request: Request): + """多租户 entity_id 提取（尽力而为）: 查询参数 → JSON body（仅 application/json）""" + try: + q = request.query_params.get("entity_id") + if q is not None and str(q) != "": + return int(q) if str(q).isdigit() else q + except Exception: + pass + ctype = (request.headers.get("content-type") or "").lower() + if "application/json" in ctype: + try: + raw = await request.body() + if raw: + data = json.loads(raw) + eid = data.get("entity_id") + if eid is not None: + return int(eid) if str(eid).isdigit() else eid + except Exception: + pass + return None + + +def _extract_l3_rows(level, resp_body: bytes): + """L3 批量写: 从响应体提取行数/条数（imported / kpi_updated / causality_links / total）""" + if level != "L3" or not resp_body: + return None + try: + data = json.loads(resp_body) + except Exception: + return None + if not isinstance(data, dict): + return None + for key in ("imported", "kpi_updated"): + v = data.get(key) + if isinstance(v, (int, float)): + return int(v) + links = data.get("causality_links") + if isinstance(links, list): + return len(links) + total = data.get("total") + if isinstance(total, (int, float)): + return int(total) + return None + + +async def bot_audit_middleware(request: Request, call_next): + """HTTP 中间件: 仅审计 /api/cma/bot* 路径；任何异常不影响业务""" + path = request.url.path + if not path.startswith("/api/cma/bot"): + return await call_next(request) + + entity_id = None + try: + entity_id = await _extract_entity_id(request) + except Exception: + entity_id = None + + try: + response = await call_next(request) + except Exception: + # 兜底记录（全局异常handler会返回500，此处保证审计不丢） + try: + write_audit_line({ + "timestamp": datetime.now().isoformat(timespec="seconds"), + "bot_name": _resolve_bot_name(request), + "endpoint": path, + "method": request.method, + "risk_level": _resolve_risk_level(request) or "NA", + "entity_id": entity_id, + "status": 500, + }) + except Exception: + pass + raise + + # 级别解析（route 由路由器在 call_next 内写入 scope） + level = None + try: + route = request.scope.get("route") + endpoint = getattr(route, "endpoint", None) if route else None + level = _resolve_risk_level(request, route=route, endpoint=endpoint) + except Exception: + level = None + + # 捕获响应体（L3 需要行数），并重放 + resp_body = b"" + try: + body_iterator = getattr(response, "body_iterator", None) + if body_iterator is not None: + chunks = [chunk async for chunk in body_iterator] + resp_body = b"".join(chunks) + rows = _extract_l3_rows(level, resp_body) + except Exception: + rows = None + + record = { + "timestamp": datetime.now().isoformat(timespec="seconds"), + "bot_name": _resolve_bot_name(request), + "endpoint": path, + "method": request.method, + "risk_level": level or "NA", + "entity_id": entity_id, + "status": response.status_code, + } + if rows is not None: + record["rows"] = rows + + try: + write_audit_line(record) + except Exception: + pass + + # 响应体已被消费 → 重建响应（Bot接口均为小JSON，非流式） + if resp_body: + from fastapi.responses import Response + return Response( + content=resp_body, + status_code=response.status_code, + headers=dict(response.headers), + media_type=response.media_type, + ) + return response diff --git a/backend/app/api/bot_bridge.py b/backend/app/api/bot_bridge.py index e0078a17..d810ce02 100644 --- a/backend/app/api/bot_bridge.py +++ b/backend/app/api/bot_bridge.py @@ -18,6 +18,7 @@ from app.models import ( from app.models.budget_plan import BudgetPlan from app.models.cost_model import StandardCost, ActualCost, AbcActivity, AbcAllocation from app.models import KPICausality +from app.risk_levels import risk_level import json logger = logging.getLogger("cma.bot_bridge") @@ -79,6 +80,7 @@ def _model_dict(obj, fields: dict): # ═══════════════ 端点 ═══════════════ @router.get("/ping") +@risk_level("L1") def ping(): return {"status": "ok", "version": "1.0", "timestamp": datetime.now().isoformat()} @@ -86,6 +88,7 @@ def ping(): # ── 总览 ── @router.get("/overview") +@risk_level("L1") def bot_overview( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), @@ -110,6 +113,7 @@ def bot_overview( # ── KPI ── @router.get("/kpis") +@risk_level("L1") def bot_kpis( dimension: Optional[str] = Query(None), status: str = Query("active"), @@ -145,6 +149,7 @@ def bot_kpis( @router.get("/kpis/{kpi_id}/history") +@risk_level("L1") def bot_kpi_history( kpi_id: int, limit: int = Query(12, le=60), bot: dict = Depends(verify_bot_key), @@ -171,6 +176,7 @@ def bot_kpi_history( # ── 战略地图 ── @router.get("/strategic-maps") +@risk_level("L1") def bot_maps( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), @@ -198,6 +204,7 @@ def bot_maps( # ── 预警 ── @router.get("/alerts") +@risk_level("L1") def bot_alerts( status: str = Query("pending"), level: Optional[str] = Query(None), @@ -228,6 +235,7 @@ def bot_alerts( # ── 预算 ── @router.get("/budget/plans") +@risk_level("L1") def bot_budget_plans( year: Optional[int] = Query(None), bot: dict = Depends(verify_bot_key), @@ -255,6 +263,7 @@ def bot_budget_plans( # ── 成本 ── @router.get("/cost/standard") +@risk_level("L1") def bot_standard_costs( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), @@ -278,6 +287,7 @@ def bot_standard_costs( @router.get("/cost/actual") +@risk_level("L1") def bot_actual_costs( period: Optional[str] = Query(None), bot: dict = Depends(verify_bot_key), @@ -306,6 +316,7 @@ def bot_actual_costs( # ── 行动方案 ── @router.get("/actions") +@risk_level("L1") def bot_actions( status: Optional[str] = Query(None), bot: dict = Depends(verify_bot_key), @@ -333,6 +344,7 @@ def bot_actions( # ── 组织 ── @router.get("/organization") +@risk_level("L1") def bot_org( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), @@ -354,6 +366,7 @@ def bot_org( # ── 数据源 ── @router.get("/data-sources") +@risk_level("L1") def bot_data_sources( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), @@ -377,6 +390,7 @@ def bot_data_sources( # ── 用户 ── @router.get("/users") +@risk_level("L1") def bot_users( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), @@ -395,6 +409,7 @@ def bot_users( # ── 统一查询（BOT首选） ── @router.get("/query") +@risk_level("L1") def bot_query( q: str = Query("overview", description="overview/kpis/alerts/maps/budget/cost/actions/all"), bot: dict = Depends(verify_bot_key), @@ -482,6 +497,7 @@ def bot_query( @router.post("/import") +@risk_level("L3") def bot_import_excel( file: UploadFile = File(...), bot: dict = Depends(verify_bot_key), @@ -546,6 +562,7 @@ def bot_import_excel( # ── 自然语言查询 ── @router.post("/okr/create") +@risk_level("L2") def bot_okr_create( title: str = Query(...), quarter: str = Query(...), @@ -563,6 +580,7 @@ def bot_okr_create( @router.get("/okr/list") +@risk_level("L1") def bot_okr_list( quarter: Optional[str] = Query(None), bot: dict = Depends(verify_bot_key), @@ -584,6 +602,7 @@ def bot_okr_list( @router.get("/nlp") +@risk_level("L1") def bot_nlp( intent: str = Query("overview"), bot: dict = Depends(verify_bot_key), @@ -613,6 +632,7 @@ def bot_nlp( # ════════════════════════════════════════════════════════════ @router.post("/kpi-value-with-check") +@risk_level("L2") def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)): """聚合A: 写KPI值 + 自动跑该KPI预警检查（Agent一次调用，免自拼check-all） body: {kpi_id, actual_value, period?, entity_id?, run_check?}""" @@ -681,6 +701,7 @@ def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dic @router.post("/kpis/create-with-links") +@risk_level("L3") def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)): """聚合B: 创建KPI + 关联战略地图 + 批量因果链（Agent建KPI标准动作） body: {kpi_code, kpi_name, dimension, entity_id?, target_value?, unit?, link_map_id?, link_causality?}""" @@ -736,3 +757,18 @@ def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: di db.commit() return {"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "map_id": link_map_id, "causality_links": links, "status": "ok"} + + +@router.get("/risk-levels") +@risk_level("L1") +def bot_risk_levels(bot: dict = Depends(verify_bot_key)): + """Bot API风险分级清单（API→级别→处理方式）— 验收/巡检/授权决策用""" + from app.risk_levels import RISK_LEVELS, list_api_risk_map, risk_summary + return { + "bot": bot["name"], + "risk_levels": RISK_LEVELS, + "apis": list_api_risk_map(), + "summary": risk_summary(), + "note": "L4（危险）不向Bot API开放：Bot只能通过白名单API读写，" + "DROP/TRUNCATE/批量DELETE/生产结构修改物理不可能，由终端层+approval-gate拦截", + } diff --git a/backend/app/api/bot_bridge_v2.py b/backend/app/api/bot_bridge_v2.py index 49c764b0..93de44ae 100644 --- a/backend/app/api/bot_bridge_v2.py +++ b/backend/app/api/bot_bridge_v2.py @@ -23,6 +23,7 @@ from app.models import ( KPIDefinition, KPIValue, KPIAlert, ActionPlan, Entity, ) +from app.risk_levels import risk_level logger = logging.getLogger("cma.bot_bridge_v2") @@ -279,6 +280,7 @@ ALERT_THRESHOLDS = { # ═══════════════════════════════════════════════ @router.post("/mpm-result") +@risk_level("L3") def receive_mpm_result( data: dict, bridge_bot: str = Depends(verify_bridge_token), @@ -404,6 +406,7 @@ def receive_mpm_result( @router.post("/kpi-result") +@risk_level("L2") def push_kpi_result( data: dict, bridge_bot: str = Depends(verify_bridge_token), @@ -521,6 +524,7 @@ def push_kpi_result( @router.post("/verify/{action_plan_id}") +@risk_level("L2") def verify_action_plan( action_plan_id: int, bridge_bot: str = Depends(verify_bridge_token), @@ -599,6 +603,7 @@ def verify_action_plan( @router.get("/verify/{action_plan_id}/history") +@risk_level("L1") def verify_history( action_plan_id: int, bridge_bot: str = Depends(verify_bridge_token), diff --git a/backend/app/api/bot_iron_law.py b/backend/app/api/bot_iron_law.py index 0f2ae891..3652f940 100644 --- a/backend/app/api/bot_iron_law.py +++ b/backend/app/api/bot_iron_law.py @@ -9,6 +9,7 @@ from sqlalchemy.orm import Session from sqlalchemy import func from app.database import get_db from app.models import ActionPlan +from app.risk_levels import risk_level logger = logging.getLogger("cma.iron_law") @@ -150,6 +151,7 @@ def _query_action_plan_verify(db: Session): # ═══════════════ 端点 ═══════════════ @router.get("/iron-law") +@risk_level("L1") def get_iron_law_kpis( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), @@ -224,6 +226,7 @@ def get_iron_law_kpis( @router.get("/iron-law/bots") +@risk_level("L1") def get_bot_iron_law_ranking( bot: dict = Depends(verify_bot_key), db: Session = Depends(get_db), diff --git a/backend/app/api/bot_kpis.py b/backend/app/api/bot_kpis.py index de14d365..f95446fa 100644 --- a/backend/app/api/bot_kpis.py +++ b/backend/app/api/bot_kpis.py @@ -7,6 +7,7 @@ from datetime import datetime from app.database import get_db from app.auth_middleware import require_auth from app.models import KPIDefinition, KPIValue +from app.risk_levels import risk_level router = APIRouter(prefix="/api/cma/bot-kpis", tags=["Bot KPI管理"], dependencies=[Depends(require_auth)], @@ -46,6 +47,7 @@ def _calc_bot_kpi_score(current_value, target_value, is_reverse=False): @router.get("") +@risk_level("L1") def list_bot_kpis( source: str = Query("finance-bot", description="Bot标识"), period: Optional[str] = None, @@ -132,6 +134,7 @@ def list_bot_kpis( @router.post("/{kpi_id}/value") +@risk_level("L2") def update_bot_kpi_value( kpi_id: int, data: dict, diff --git a/backend/app/main.py b/backend/app/main.py index 72ca0da8..aaea8906 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,6 +9,7 @@ from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync from app.auth_middleware import require_auth +from app.api.audit_log import bot_audit_middleware load_dotenv() @@ -30,6 +31,9 @@ app.add_middleware( allow_headers=["*"], ) +# Bot API 操作审计（L1-L4分级标注 + JSON行审计日志，不阻塞业务） +app.middleware("http")(bot_audit_middleware) + app.include_router(auth.router) app.include_router(kpis.router) app.include_router(kpi_governance.router) diff --git a/backend/app/risk_levels.py b/backend/app/risk_levels.py new file mode 100644 index 00000000..b454feab --- /dev/null +++ b/backend/app/risk_levels.py @@ -0,0 +1,120 @@ +""" +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 diff --git a/backend/tests/test_risk_levels.py b/backend/tests/test_risk_levels.py new file mode 100644 index 00000000..97340ba0 --- /dev/null +++ b/backend/tests/test_risk_levels.py @@ -0,0 +1,189 @@ +"""Bot API 风险分级（L1-L4）标注 + 操作审计日志 测试 + +覆盖: + 1. API_RISK_MAP 覆盖所有 /api/cma/bot* 路由（app.routes 遍历核对） + 2. GET /api/cma/bot/risk-levels 返回200且含L1-L4定义 + 3. Bot API面不存在L4端点（无 drop/truncate/delete 批量端点，安全底线） + 4. 审计日志在调用Bot API后写入（monkeypatch + 真实日志文件双验证） +""" +import io +import json + +import pytest +from openpyxl import Workbook + +from app.main import app +from app.models import KPIDefinition +from app.risk_levels import API_RISK_MAP, RISK_LEVELS +import app.api.audit_log as audit_log_module + +BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"} +_HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "PATCH") + + +def _bot_routes(): + """遍历 app.routes，返回所有 /api/cma/bot* 路由 (route, methods列表)""" + routes = [] + for r in app.routes: + path = getattr(r, "path", "") + if path.startswith("/api/cma/bot"): + methods = sorted(m for m in (getattr(r, "methods", set()) or set()) + if m in _HTTP_METHODS) + routes.append((r, methods)) + return routes + + +class TestRiskMapCoverage: + def test_api_risk_map_covers_all_bot_routes(self): + """每个 /api/cma/bot* 路由都在 API_RISK_MAP 有级别标注，且函数有@risk_level装饰器""" + missing = [] + unlabeled = [] + for r, methods in _bot_routes(): + for m in methods: + key = f"{m} {r.path}" + if key not in API_RISK_MAP: + missing.append(key) + if not getattr(r.endpoint, "risk_level", None): + unlabeled.append(f"{sorted(methods)} {r.path}") + assert not missing, f"API_RISK_MAP 缺少以下路由标注: {missing}" + assert not unlabeled, f"以下路由函数缺少 @risk_level 装饰器: {unlabeled}" + + def test_risk_level_counts(self): + """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=5(4项清单+okr/create)、L3=3、L4=0""" + from collections import Counter + counts = Counter(API_RISK_MAP.values()) + assert counts["L1"] == 21, counts + assert counts["L2"] == 5, counts + assert counts["L3"] == 3, counts + assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）" + + def test_no_l4_bot_endpoints(self): + """Bot API面不存在L4端点：无DELETE方法、无drop/truncate/delete危险路径""" + danger_keywords = ("drop", "truncate", "delete") + for r, methods in _bot_routes(): + assert getattr(r.endpoint, "risk_level", None) != "L4", \ + f"{r.path} 不应被标注为L4" + assert "DELETE" not in methods, f"Bot路由不应有DELETE方法: {r.path}" + low = r.path.lower() + for kw in danger_keywords: + assert kw not in low, f"Bot路由不应含危险路径片段: {r.path}" + + +class TestRiskLevelsEndpoint: + def test_risk_levels_requires_bot_key(self, client): + """无X-BOT-KEY → 401""" + resp = client.get("/api/cma/bot/risk-levels") + assert resp.status_code == 401 + + def test_risk_levels_returns_200_with_definitions(self, client): + """X-BOT-KEY → 200，含L1-L4定义与API→级别→处理方式清单""" + resp = client.get("/api/cma/bot/risk-levels", headers=BOT_KEY) + assert resp.status_code == 200 + data = resp.json() + + # L1-L4 定义齐全 + for level, label in RISK_LEVELS.items(): + assert data["risk_levels"][level] == label, f"缺少 {level} 定义" + + # API→级别→处理方式 清单 + apis = {f"{a['method']} {a['path']}": a for a in data["apis"]} + assert apis["GET /api/cma/bot/ping"]["risk_level"] == "L1" + assert apis["POST /api/cma/bot/kpi-value-with-check"]["risk_level"] == "L2" + assert apis["POST /api/cma/bot/import"]["risk_level"] == "L3" + assert apis["POST /api/cma/bot/import"]["handling"] # 处理方式非空 + + # 分级统计 + assert data["summary"]["L1"] == 21 + assert data["summary"]["L2"] == 5 + assert data["summary"]["L3"] == 3 + assert data["summary"]["L4"] == 0 + + +class TestAuditLog: + def test_audit_record_after_bot_call(self, client, monkeypatch): + """调用Bot API后写出审计记录（monkeypatch捕获）""" + records = [] + monkeypatch.setattr(audit_log_module, "write_audit_line", + lambda rec: records.append(rec)) + resp = client.get("/api/cma/bot/ping") + assert resp.status_code == 200 + assert len(records) >= 1 + rec = records[-1] + assert rec["method"] == "GET" + assert rec["endpoint"] == "/api/cma/bot/ping" + assert rec["risk_level"] == "L1" + assert rec["status"] == 200 + # 字段齐全 + for field in ("timestamp", "bot_name", "endpoint", "method", + "risk_level", "entity_id", "status"): + assert field in rec, f"审计记录缺少字段: {field}" + + def test_audit_file_written_json_lines(self, client, tmp_path, monkeypatch): + """真实日志文件：调用Bot API后 bot_audit.log 追加JSON行""" + log_file = tmp_path / "bot_audit.log" + monkeypatch.setenv("CMA_BOT_AUDIT_LOG", str(log_file)) + monkeypatch.setattr(audit_log_module, "_audit_logger", None) + + resp = client.get("/api/cma/bot/ping") + assert resp.status_code == 200 + assert log_file.exists() + + lines = log_file.read_text(encoding="utf-8").strip().splitlines() + assert lines, "审计日志文件为空" + rec = json.loads(lines[-1]) + assert rec["endpoint"] == "/api/cma/bot/ping" + assert rec["risk_level"] == "L1" + assert rec["status"] == 200 + + def test_audit_entity_id_and_bot_name(self, client, db, monkeypatch): + """审计记录含 bot_name（X-BOT-KEY映射）与 entity_id""" + records = [] + monkeypatch.setattr(audit_log_module, "write_audit_line", + lambda rec: records.append(rec)) + kpi = KPIDefinition(kpi_code="AUDIT_EID", kpi_name="审计实体", dimension="finance", + status="active", target_value=1.0, entity_id=1) + db.add(kpi) + db.commit() + + resp = client.post("/api/cma/bot/kpi-value-with-check", + json={"kpi_id": kpi.id, "actual_value": 66.0, + "period": "2026-08", "entity_id": 1}, + headers=BOT_KEY) + assert resp.status_code == 200 + assert records, "应有审计记录" + rec = records[-1] + assert rec["bot_name"] == "财务BOT" + assert rec["entity_id"] == 1 + assert rec["risk_level"] == "L2" + assert rec["status"] == 200 + # JSON body 读取未破坏业务 + assert resp.json()["status"] == "ok" + + def test_l3_batch_write_records_rows(self, client, db, monkeypatch): + """L3批量写（/import）：审计记录额外含 rows 行数""" + kpi = KPIDefinition(kpi_code="AUDIT_ROWS", kpi_name="审计行数", dimension="finance", + status="active", target_value=1.0) + db.add(kpi) + db.commit() + + wb = Workbook() + ws = wb.active + ws.append(["kpi_code", "period", "actual_value"]) + ws.append(["AUDIT_ROWS", "2026-08", 88.0]) + ws.append(["AUDIT_ROWS", "2026-07", 77.0]) + buf = io.BytesIO() + wb.save(buf) + + records = [] + monkeypatch.setattr(audit_log_module, "write_audit_line", + lambda rec: records.append(rec)) + files = {"file": ("kpi.xlsx", buf.getvalue(), + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")} + resp = client.post("/api/cma/bot/import", headers=BOT_KEY, files=files) + assert resp.status_code == 200 + assert resp.json()["imported"] == 2 + + assert records, "应有审计记录" + rec = records[-1] + assert rec["risk_level"] == "L3" + assert rec["rows"] == 2