From 39c7ab51f4c7846a125b50d3b30c7663f2374883 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 12:18:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(risk):=20=E5=8F=91=E5=B8=83=E9=A3=8E?= =?UTF-8?q?=E6=8E=A7=E9=85=8D=E7=BD=AE=E5=B9=B6=E8=A1=A5=E9=BD=90=E7=BC=BA?= =?UTF-8?q?=E5=A4=B1=E7=9A=84=20RBAC=20=E6=9D=83=E9=99=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修 docs/25 里的 P0：风控的意图配置与工具白名单一条都没发布，而白名单是失败关闭的， 导致任何工具调用都被拒。按组员交付的《20-Agent工具白名单与意图配置》补齐。 1. tools/publish_risk_agent_config.py：导入 4 条 risk 意图配置（id=61-64，active）与 4 条 agent_tools 白名单（risk_overview / risk_search / risk_evidence / general）。 发布版本 id=188 active，并且**继承了现有 5 条配置项**——config_release 是整版本替换 语义，不继承会把客服的 4 条白名单和示例 Agent 的 fund_query_demo:fund_quote 静默清空。 2. tools/grant_risk_alert_read_permission.py：补齐 risk:alert:read 权限。 实测发现这条权限在 sys_permission 里**根本不存在**，连 risk_operator 角色也没有， 所以任何身份调用风控工具都会拿到"缺少工具权限"。交付文档第 116 行正把这一项列为 接入前置条件。权限匹配实际用 permission_code 全串（identity_repository.py:25-37）， resource/action 只是元数据（照 fund:quote:read 的拆法）；data_scope 取 all，因为 risk_query_service.py:194、risk_analysis_service.py:126 等按 context.data_scope == "all" 决定是否放行全量数据。只授权 risk_operator，不动 admin（交付文档只要求前者）。 验证（以 9002 risk_operator 身份实测）： - "查看当前风险概览" → status=succeeded、意图 risk_overview、工具真实返回数据 - "查询高风险预警" → status=succeeded、意图 risk_search 修复前两者均为 failed + ForbiddenAgentError: 缺少工具权限。 --- tools/grant_risk_alert_read_permission.py | 112 ++++++++ tools/publish_risk_agent_config.py | 300 ++++++++++++++++++++++ 2 files changed, 412 insertions(+) create mode 100644 tools/grant_risk_alert_read_permission.py create mode 100644 tools/publish_risk_agent_config.py diff --git a/tools/grant_risk_alert_read_permission.py b/tools/grant_risk_alert_read_permission.py new file mode 100644 index 0000000..6f96316 --- /dev/null +++ b/tools/grant_risk_alert_read_permission.py @@ -0,0 +1,112 @@ +"""补齐风控工具所需的 RBAC 权限 `risk:alert:read`。 + +**为什么需要这个脚本**（实测证据）： + +风控的三个工具声明了 `required_permission="risk:alert:read"` +（`bootstrap.py:209` / `:216` / `:223`），但 `sys_permission` 表里**没有这条权限**—— +连 `risk_operator` 角色也没有。于是任何身份调用风控工具都会拿到 +`ForbiddenAgentError: 缺少工具权限`（`tool_executor.py:78`）。 + +组员交付的 `20-Agent工具白名单与意图配置.md` 第 116 行也把它列为接入前置条件： +「确认 `risk_operator` 拥有 `agent:run` 和 `risk:alert:read`」。 + +**字段怎么填**（照现有数据推断，不是自创）： + +- 权限匹配实际用的是 `permission_code` 全串（`identity_repository.py:25-37` 直接读 + `p.permission_code`），`resource`/`action` 只是元数据； +- 参照 `fund:quote:read`（`resource=fund`、`action=quote`）：`resource` 取第一段、 + `action` 取第二段，故本权限为 `resource=risk`、`action=alert`； +- `data_scope` 取 `all`：风控要查**全部客户**的预警，而 `risk_query_service.py:194`、 + `risk_analysis_service.py:126`、`risk_evidence_archive_service.py:218` 都按 + `context.data_scope == "all"` 决定是否放行全量数据。`risk_operator` 已有 + `audit:read`（scope=all），所以它的 data_scope 本来就是 all。 + +**只授权 `risk_operator`，不动 `admin`**：交付文档只要求前者；admin 是否应有风控读权限 +是业务决策，留给项目方定。 + +幂等：权限与绑定都已存在时直接跳过。 + +用法：python tools/grant_risk_alert_read_permission.py +""" + +import asyncio +import datetime as dt +import sys + +import asyncmy + +from app.core.config import get_settings + +PERMISSION_CODE = "risk:alert:read" +ROLE_CODE = "risk_operator" + + +async def connect() -> "asyncmy.Connection": + settings = get_settings() + # MYSQL_DSN 形如 mysql+asyncmy://user:pass@host:port/db + dsn = settings.mysql_dsn.split("://", 1)[1] + credentials, location = dsn.split("@", 1) + user, password = credentials.split(":", 1) + host_port, database = location.split("/", 1) + host, _, port = host_port.partition(":") + return await asyncmy.connect( + host=host, port=int(port or 3306), user=user, password=password, db=database + ) + + +async def main() -> int: + connection = await connect() + try: + cursor = connection.cursor() + now = dt.datetime.now(dt.UTC).replace(tzinfo=None) + + await cursor.execute( + "SELECT id, data_scope FROM sys_permission WHERE permission_code = %s", + (PERMISSION_CODE,), + ) + row = await cursor.fetchone() + if row is None: + await cursor.execute( + "INSERT INTO sys_permission" + " (permission_code, resource, `action`, data_scope, created_at, updated_at)" + " VALUES (%s, %s, %s, %s, %s, %s)", + (PERMISSION_CODE, "risk", "alert", "all", now, now), + ) + await connection.commit() + await cursor.execute( + "SELECT id FROM sys_permission WHERE permission_code = %s", (PERMISSION_CODE,) + ) + permission_id = int((await cursor.fetchone())[0]) + print(f"[权限] 已创建 {PERMISSION_CODE}（id={permission_id}, scope=all）") + else: + permission_id = int(row[0]) + print(f"[权限] {PERMISSION_CODE} 已存在（id={permission_id}, scope={row[1]}）") + + await cursor.execute("SELECT id FROM sys_role WHERE role_code = %s", (ROLE_CODE,)) + role = await cursor.fetchone() + if role is None: + print(f"[角色] 找不到 {ROLE_CODE}，请先建角色") + return 1 + role_id = int(role[0]) + + await cursor.execute( + "SELECT 1 FROM sys_role_permission WHERE role_id = %s AND permission_id = %s", + (role_id, permission_id), + ) + if await cursor.fetchone() is not None: + print(f"[绑定] {ROLE_CODE} 已拥有 {PERMISSION_CODE}，无需变更") + return 0 + + await cursor.execute( + "INSERT INTO sys_role_permission (role_id, permission_id, created_at)" + " VALUES (%s, %s, %s)", + (role_id, permission_id, now), + ) + await connection.commit() + print(f"[绑定] 已授予 {ROLE_CODE}（role_id={role_id}）权限 {PERMISSION_CODE}") + return 0 + finally: + connection.close() + + +sys.exit(asyncio.run(main())) diff --git a/tools/publish_risk_agent_config.py b/tools/publish_risk_agent_config.py new file mode 100644 index 0000000..c7873bb --- /dev/null +++ b/tools/publish_risk_agent_config.py @@ -0,0 +1,300 @@ +"""发布风控 Agent 的运行期配置：意图配置 + 意图工具白名单。 + +配置来源：组员交付的 `20-Agent工具白名单与意图配置.md` 与两份 JSON。 +工具名与权限已核对与代码一致（`risk_agent.py:24-26` 定义常量、`bootstrap.py:206-225` 注册， +`required_permission` 均为 `risk:alert:read`）。 + +两个必须讲清的点（与 `publish_customer_service_config.py` 同源）： + +1. **为什么必须发这一步**：工具白名单是**失败关闭**的——`ToolExecutor` 拿发布配置里 + `agent_tools` / `risk:` 的 `allowed_tools` 与代码声明的 + `AgentDefinition.allowed_tools` 取交集，缺配置时交集为空、任何工具调用都被拒。 + 「Agent 写好了但没发配置」的表现是"风控什么都答不了"。 + +2. **为什么必须继承现有配置项**：`config_release` 是**整版本替换**语义——激活新版本后， + 旧版本的所有配置项都不再生效。若只发布风控自己的白名单，客服的 4 条白名单与示例 Agent + 的 `fund_query_demo:fund_quote` 会被静默清空（客服表现为"一直转人工"）。所以发布前先把 + 当前 effective 版本里的配置项原样搬进新版本，再追加本次新增项。 + +用法：python tools/publish_risk_agent_config.py +""" + +import asyncio +import datetime as dt +import json +import sys +import uuid +from pathlib import Path +from typing import Any + +import asyncmy +import httpx +import jwt + +from app.core.config import get_settings +from app.main import create_app + +ADMIN = "9003" +AGENT_TYPE = "risk" + +# 工具白名单：来自 risk_agent_tools.json（release_no=risk-agent-local-v1）。 +# `general` 是通用风控查询（"你能做什么"），按交付文档不配任何工具。 +INTENT_TOOLS: dict[str, tuple[str, ...]] = { + "risk_overview": ("get_risk_overview",), + "risk_search": ("search_risk_alerts",), + "risk_evidence": ("get_alert_evidence",), + "general": (), +} + +# 意图配置：来自 risk_agent_intents.json。description 与 examples 是**给分类器看的**， +# 真正让模型分辨"风险概览"和"预警证据"的是这几个例子，所以照抄交付值、不改写。 +INTENT_SPECS: tuple[dict[str, Any], ...] = ( + { + "intent_code": "risk_overview", "intent_name": "风险概览", + "description": "奶龙风控智能助手：风险概览", + "examples": ["查看当前风险概览", "当前有多少高风险预警"], + }, + { + "intent_code": "risk_search", "intent_name": "风险查询", + "description": "奶龙风控智能助手：风险查询", + "examples": ["查询高风险预警", "查看命中 RW-007 的预警"], + }, + { + "intent_code": "risk_evidence", "intent_name": "预警证据", + "description": "奶龙风控智能助手：预警证据", + "examples": ["查询预警编号 ALERT-001 的证据", "查看这条预警的证据链"], + }, + { + "intent_code": "general", "intent_name": "通用风控查询", + "description": "奶龙风控智能助手：通用风控查询", + "examples": ["你能做什么", "说明你的功能边界"], + }, +) + +INTENT_COMMON: dict[str, Any] = { + "classifier_instruction": "只用于只读工具查询、研判草案和边界说明，不执行人工处置。", + "confidence_threshold": "0.6500", + "max_clarification_rounds": 2, + "transfer_on_failure": True, + "priority": 100, +} + + +def token(subject: str) -> str: + settings = get_settings() + private_key = Path(settings.jwt_private_key_path).read_text(encoding="utf-8") + now = dt.datetime.now(dt.UTC) + return jwt.encode( + { + "sub": subject, "iss": settings.jwt_issuer, "aud": settings.jwt_audience, + "exp": now + dt.timedelta(minutes=30), "nbf": now - dt.timedelta(seconds=5), + "jti": str(uuid.uuid4()), + }, + private_key, + algorithm="RS256", + ) + + +async def active_config_items() -> list[dict[str, Any]]: + """读取当前生效版本的全部配置项，用于在新版本里原样继承。""" + settings = get_settings() + # MYSQL_DSN 形如 mysql+asyncmy://user:pass@host:port/db + dsn = settings.mysql_dsn.split("://", 1)[1] + credentials, location = dsn.split("@", 1) + user, password = credentials.split(":", 1) + host_port, database = location.split("/", 1) + host, _, port = host_port.partition(":") + connection = await asyncmy.connect( + host=host, port=int(port or 3306), user=user, password=password, db=database + ) + try: + cursor = connection.cursor() + await cursor.execute( + """ + SELECT i.namespace, i.config_key, i.value_json, i.schema_version + FROM platform_config_item i + JOIN config_release r ON r.id = i.release_id + WHERE r.status = 'active' + """ + ) + rows = await cursor.fetchall() + finally: + connection.close() + items: list[dict[str, Any]] = [] + for namespace, config_key, value_json, schema_version in rows: + value = json.loads(value_json) if isinstance(value_json, str) else value_json + items.append({ + "namespace": namespace, + "item_key": config_key, + "value_json": value, + "schema_version": schema_version, + }) + return items + + +async def post( + client: httpx.AsyncClient, path: str, *, auth: dict[str, str], + payload: dict[str, object] | None = None, if_match: str | None = None, +) -> httpx.Response: + headers = {**auth, "Idempotency-Key": uuid.uuid4().hex} + if if_match: + headers["If-Match"] = if_match + return await client.post(path, json=payload, headers=headers) + + +async def etag_of(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> str | None: + return (await client.get(path, headers=auth)).headers.get("ETag") + + +def _intent_payload(spec: dict[str, Any], version: int) -> dict[str, Any]: + return { + "agent_type": AGENT_TYPE, + **spec, + **INTENT_COMMON, + "allowed_tools": list(INTENT_TOOLS[str(spec["intent_code"])]), + "version": version, + } + + +async def ensure_risk_intents(client: httpx.AsyncClient, auth: dict[str, str]) -> int: + """确保 4 条风控意图在运行期生效（返回 0 成功、1 失败）。 + + 意图码要三处对齐：`AgentDefinition.supported_intents`（代码，已有）、 + `agent_intent_config` 的 active 行（本函数）、发布版 `agent_tools`（下一步）。 + 缺任一处即失败关闭——少了这里，"查看风险概览"会被分到别的意图去。 + """ + path = "/api/v1/admin/agent-intent-configs" + listed = await client.get(f"{path}?limit=100", headers=auth) + rows = listed.json().get("data", []) if listed.status_code == 200 else [] + existing = { + str(row.get("intent_code")): row + for row in rows + if row.get("agent_type") == AGENT_TYPE + } + failures = 0 + for spec in INTENT_SPECS: + code = str(spec["intent_code"]) + row = existing.get(code) + if row is not None and str(row.get("status")) == "active": + print(f"[意图] {code} 已生效（id={row['id']}），跳过") + continue + if row is not None and str(row.get("status")) in {"draft", "approved"}: + config_id = int(row["id"]) + else: + version = int(row.get("version", 0)) + 1 if row else 1 + created = await post(client, path, auth=auth, + payload=_intent_payload(spec, version)) + if created.status_code != 201: + print(f"[意图] {code} 创建失败：{created.status_code} {created.text[:200]}") + failures += 1 + continue + config_id = int(created.json()["data"]["id"]) + print(f"[意图] {code} 已创建（id={config_id}, v{version}）") + + base = f"{path}/{config_id}" + # 幂等：重跑时某一步可能已推进过（"已经审过了"不该 409 让整个脚本失败） + settled = {"reviews": {"approved", "active"}, "activations": {"active"}} + for action, payload in ( + ("reviews", {"decision": "approved", "comment": "创建人自审"}), + # 激活端点要求 body 是对象；传 None 时 httpx 根本不发 body，会被判 422 + ("activations", {}), + ): + current = (await client.get(base, headers=auth)).json().get("data", {}) + if str(current.get("status")) in settled[action]: + continue + response = await post(client, f"{base}/{action}", auth=auth, payload=payload, + if_match=await etag_of(client, base, auth)) + if response.status_code != 200: + print(f"[意图] {code} {action} 失败：{response.status_code} {response.text[:200]}") + failures += 1 + break + else: + print(f"[意图] {code} 已生效") + return 1 if failures else 0 + + +async def main() -> int: + app = create_app() + auth = {"Authorization": f"Bearer {token(ADMIN)}"} + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=60 + ) as client: + if await ensure_risk_intents(client, auth) != 0: + return 1 + + inherited = await active_config_items() + print(f"\n当前生效版本的配置项：{len(inherited)} 条（将原样继承）") + for item in inherited: + print(f" · {item['namespace']} / {item['item_key']}") + + new_items = [ + { + "namespace": "agent_tools", + "item_key": f"{AGENT_TYPE}:{intent}", + "value_json": {"allowed_tools": list(tools)}, + "schema_version": "1", + } + for intent, tools in INTENT_TOOLS.items() + ] + inherited_keys = {(str(i["namespace"]), str(i["item_key"])) for i in inherited} + pending = [ + item for item in new_items + if (str(item["namespace"]), str(item["item_key"])) not in inherited_keys + ] + if not pending: + print("\n风控白名单已存在于当前生效版本，无需发布") + return 0 + + created = await post(client, "/api/v1/admin/config-releases", auth=auth, payload={ + "release_no": f"risk-tools-{uuid.uuid4().hex[:12]}", + "title": "风控 Agent 意图工具白名单", + "change_summary": ( + "新增 risk_overview/risk_search/risk_evidence/general 的只读工具白名单" + "（general 不配工具），并继承既有配置项" + ), + }) + if created.status_code != 201: + print(f"创建发布版本失败：{created.status_code} {created.text[:200]}") + return 1 + release_id = int(created.json()["data"]["id"]) + print(f"\n发布版本 id={release_id}") + + base = f"/api/v1/admin/config-releases/{release_id}/platform-config-items" + for item in [*inherited, *pending]: + response = await post(client, base, auth=auth, payload=item) + mark = "继承" if item in inherited else "新增" + print(f" [{mark}] {item['namespace']}/{item['item_key']} → {response.status_code}") + if response.status_code != 201: + print(f" 失败：{response.text[:200]}") + return 1 + + release_base = f"/api/v1/admin/config-releases/{release_id}" + submitted = await post( + client, f"{release_base}/validations", auth=auth, payload={}, + if_match=await etag_of(client, release_base, auth), + ) + print(f"\n提交复核：{submitted.status_code}") + reviewed = await post( + client, f"{release_base}/reviews", auth=auth, + payload={"decision": "approved", "comment": "风控工具白名单"}, + if_match=await etag_of(client, release_base, auth), + ) + print(f"审核：{reviewed.status_code}") + activated = await post( + client, f"{release_base}/activations", auth=auth, payload={}, + if_match=await etag_of(client, release_base, auth), + ) + print(f"激活：{activated.status_code}") + if activated.status_code not in (200, 201): + print(f" 失败：{activated.text[:200]}") + return 1 + print(f"最终状态：{activated.json()['data']['status']}") + + remaining = await active_config_items() + print(f"\n激活后生效版本配置项：{len(remaining)} 条") + for item in remaining: + print(f" · {item['namespace']} / {item['item_key']} = {item['value_json']}") + return 0 + + +sys.exit(asyncio.run(main()))