193 lines
9.4 KiB
Python
193 lines
9.4 KiB
Python
"""造 RBAC 测试数据，验证真实链路身份加载与越权拦截。
|
||
|
||
数据使用 9001/9002/9003 号段（角色、权限分别用 9001 起的独立号段），便于清理，
|
||
不影响业务数据。
|
||
|
||
为什么需要这份种子：权限码必须与**代码实际声明**一致——工具声明写在
|
||
`app/service/agent/bootstrap.py`（`check_suitability` 需 `suitability:read`、
|
||
`query_fund_quote` 需 `fund:quote:read`），接口校验写在各 Service 的
|
||
`AuthorizationService.require(...)`。种子缺权限的后果不是"少测一项"，而是
|
||
"功能明明实现了却一直 403"，让验收与集成测试得出错误结论。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from datetime import UTC, datetime, timedelta
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.service.identity_service import IdentityService
|
||
|
||
CUSTOMER_USER = "9001"
|
||
RISK_USER = "9002"
|
||
ADMIN_USER = "9003"
|
||
|
||
# (权限 id, 权限码, resource, action, data_scope)
|
||
PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = (
|
||
(9001, "agent:run", "agent", "run", "self"),
|
||
(9002, "suitability:read", "suitability", "read", "self"),
|
||
(9003, "fund:quote:read", "fund", "quote", "self"),
|
||
(9004, "conversation:create", "conversation", "create", "self"),
|
||
(9005, "conversation:close", "conversation", "close", "self"),
|
||
(9006, "conversation:feedback", "conversation", "feedback", "self"),
|
||
(9007, "agent:cancel", "agent", "cancel", "self"),
|
||
(9008, "handover:create", "handover", "create", "self"),
|
||
(9009, "memory:read:self", "memory", "read", "self"),
|
||
(9010, "memory:read:customer", "memory", "read", "own_customers"),
|
||
(9011, "knowledge:reference:read", "knowledge", "reference", "self"),
|
||
(9012, "audit:read", "audit", "read", "all"),
|
||
(9013, "config:read", "config", "read", "all"),
|
||
(9014, "config:write", "config", "write", "all"),
|
||
(9015, "config:review", "config", "review", "all"),
|
||
(9016, "config:activate", "config", "activate", "all"),
|
||
(9017, "model-endpoint:manage", "model-endpoint", "manage", "all"),
|
||
# `query_knowledge` 工具声明该权限（`app/service/knowledge_tool.py` 的
|
||
# `REQUIRED_PERMISSION`）；缺失会让客服 Agent 调工具时被拒（缺少工具权限）。
|
||
# 补入种子以保证"重跑种子 / 换环境"后权限仍存在。
|
||
(9018, "knowledge:query", "knowledge", "query", "self"),
|
||
# 知识库**管理**面（Task 11）：上传/列表/删除三个端点统一要这个权限码
|
||
# （`app/service/knowledge_management_service.py` 的 `REQUIRED_PERMISSION`）。
|
||
# 刻意不把 `knowledge:query`（客户角色也有）复用成管理权限：那会让客户看到整库文档清单。
|
||
# 归属：只给 admin（ADMIN_PERMISSIONS 是全量元组，自动包含本行）。
|
||
(9019, "knowledge:manage", "knowledge", "manage", "all"),
|
||
(9020, "investment-goal:write:self", "investment-goal", "write", "self"),
|
||
(9021, "investment-goal:confirm:self", "investment-goal", "confirm", "self"),
|
||
(9022, "investment-goal:read:self", "investment-goal", "read", "self"),
|
||
(9023, "product-recommendation:generate:self", "product-recommendation", "generate", "self"),
|
||
(9024, "product-recommendation:read:self", "product-recommendation", "read", "self"),
|
||
(9025, "portfolio-analysis:read:self", "portfolio-analysis", "read", "self"),
|
||
(9026, "asset-allocation:generate:self", "asset-allocation", "generate", "self"),
|
||
(9027, "profile-governance:read", "profile-governance", "read", "all"),
|
||
(9028, "profile-governance:review", "profile-governance", "review", "all"),
|
||
(9029, "investment-goal:review", "investment-goal", "review", "all"),
|
||
(9030, "investment-goal:publish", "investment-goal", "publish", "all"),
|
||
(9031, "product-recommendation:review", "product-recommendation", "review", "all"),
|
||
(9032, "product-recommendation:publish", "product-recommendation", "publish", "all"),
|
||
(9033, "asset-allocation:backtest", "asset-allocation", "backtest", "all"),
|
||
(9034, "product-comparison:read:self", "product-comparison", "read", "self"),
|
||
)
|
||
|
||
# 客户：业务侧自助能力（自己的会话、反馈、转人工、自己的记忆画像）。
|
||
CUSTOMER_PERMISSIONS = (
|
||
9001, 9002, 9003, 9004, 9005, 9006, 9007, 9008, 9009, 9011, 9018,
|
||
9020, 9021, 9022, 9023, 9024, 9025, 9026, 9034,
|
||
)
|
||
# 风控专员：业务侧只读 + 跨客户记忆 + 审计只读，不含配置写权限。
|
||
RISK_PERMISSIONS = (9001, 9002, 9003, 9010, 9011, 9012)
|
||
# 平台管理员：管理面全套（配置发布四态 + 模型端点 + 审计）。
|
||
ADMIN_PERMISSIONS = tuple(row[0] for row in PERMISSIONS)
|
||
|
||
ROLES: tuple[tuple[int, str, str], ...] = (
|
||
(9001, "customer", "客户"),
|
||
(9002, "risk_operator", "风控专员"),
|
||
(9003, "admin", "平台管理员"),
|
||
)
|
||
|
||
USERS: tuple[tuple[int, str, str, str], ...] = (
|
||
(9001, "T-CUST", "cust_t", "customer"),
|
||
(9002, "T-RISK", "risk_t", "employee"),
|
||
(9003, "T-ADMIN", "admin_t", "employee"),
|
||
)
|
||
|
||
GRANTS: tuple[tuple[int, tuple[int, ...]], ...] = (
|
||
(9001, CUSTOMER_PERMISSIONS),
|
||
(9002, RISK_PERMISSIONS),
|
||
(9003, ADMIN_PERMISSIONS),
|
||
)
|
||
|
||
|
||
async def seed() -> None:
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
# MySQL 的 DATETIME(0) 会把微秒四舍五入到秒：若用"当前时间"写 assigned_at，
|
||
# 进位后可能落在未来，使紧随其后的 `assigned_at <= now` 授权校验判定
|
||
# "尚未生效"，于是刚种好的账号一个权限都拿不到（表现为 roles=() 而非报错）。
|
||
# 往前留 5 秒，彻底避开这个舍入窗口。
|
||
effective_at = now - timedelta(seconds=5)
|
||
user_ids = tuple(user[0] for user in USERS)
|
||
role_ids = tuple(role[0] for role in ROLES)
|
||
async with SessionFactory() as session:
|
||
await session.execute(
|
||
text("DELETE FROM sys_role_permission WHERE role_id IN (9001,9002,9003)")
|
||
)
|
||
await session.execute(text("DELETE FROM sys_user_role WHERE user_id IN (9001,9002,9003)"))
|
||
await session.execute(text("DELETE FROM sys_permission WHERE id BETWEEN 9001 AND 9099"))
|
||
await session.execute(text("DELETE FROM sys_role WHERE id IN (9001,9002,9003)"))
|
||
await session.execute(text("DELETE FROM sys_user WHERE id IN (9001,9002,9003)"))
|
||
|
||
for user_id, user_no, user_name, user_type in USERS:
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO sys_user (id, user_no, username, password_hash, user_type,"
|
||
" professional_investor_status, fund_account_status, status,"
|
||
" created_at, updated_at)"
|
||
" VALUES (:id,:no,:name,'x',:type,'none','closed','正常',:now,:now)"
|
||
),
|
||
{"id": user_id, "no": user_no, "name": user_name, "type": user_type, "now": now},
|
||
)
|
||
for role_id, role_code, role_name in ROLES:
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO sys_role (id, role_code, role_name, status, created_at, updated_at)"
|
||
" VALUES (:id,:code,:name,'active',:now,:now)"
|
||
),
|
||
{"id": role_id, "code": role_code, "name": role_name, "now": now},
|
||
)
|
||
for permission_id, code, resource, action, scope in PERMISSIONS:
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO sys_permission"
|
||
" (id, permission_code, resource, action, data_scope, created_at, updated_at)"
|
||
" VALUES (:id,:code,:resource,:action,:scope,:now,:now)"
|
||
),
|
||
{"id": permission_id, "code": code, "resource": resource, "action": action,
|
||
"scope": scope, "now": now},
|
||
)
|
||
for user_id, role_id in zip(user_ids, role_ids, strict=True):
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO sys_user_role (user_id, role_id, assigned_at)"
|
||
" VALUES (:user,:role,:now)"
|
||
),
|
||
{"user": user_id, "role": role_id, "now": effective_at},
|
||
)
|
||
for role_id, permission_ids in GRANTS:
|
||
for permission_id in permission_ids:
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO sys_role_permission (role_id, permission_id, created_at)"
|
||
" VALUES (:role,:perm,:now)"
|
||
),
|
||
{"role": role_id, "perm": permission_id, "now": now},
|
||
)
|
||
await session.commit()
|
||
|
||
|
||
async def verify() -> None:
|
||
service = IdentityService()
|
||
for label, user_id in (("customer", CUSTOMER_USER), ("risk_operator", RISK_USER),
|
||
("admin", ADMIN_USER)):
|
||
context = await service.resolve(RequestContext(user_id=user_id, trace_id="verify-trace"))
|
||
print(
|
||
f"{label:14s} roles={context.roles} permissions={sorted(context.permissions)} "
|
||
f"portal={context.portal} data_scope={context.data_scope}"
|
||
)
|
||
customer = await service.resolve(
|
||
RequestContext(user_id=CUSTOMER_USER, trace_id="verify-trace")
|
||
)
|
||
missing = {"agent:run", "suitability:read", "fund:quote:read"} - set(customer.permissions)
|
||
if missing:
|
||
raise SystemExit(f"客户角色缺少公共工具所需权限：{sorted(missing)}")
|
||
print("verified: customer can pass public tool authorization checks")
|
||
|
||
|
||
async def main() -> None:
|
||
await seed()
|
||
await verify()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|