lzf_0626
554fbbcaab
实测发现：9002(risk_operator) 与 9003(admin) 都持有 all 级权限 （permission_scopes 里 audit:read='all'、risk:alert:read='all' 等）， 但 context.data_scope **永远是 'self'**——identity_repository.py 第 47 行把它写死了， 而上面第 34-39 行刚算出每个权限的 scope 并取了最高（rank 表都写好了）。 后果：凡按 context.data_scope == "all" 判断能否看全量的路径全部走不通 （risk_query_service.py:198、risk_analysis_service.py:126、 risk_evidence_archive_service.py:218、risk_action_service.py:212）， 风控专员拿着全量权限却查不到任何预警——这是上一轮三个工具"都返回空"的真正原因。 改为取该身份所有权限里的最高范围。没有 all 权限的角色行为不变 （实测 9001 customer 的 data_scope 仍是 self），因此不放松任何既有边界。 另加 tools/seed_risk_alert_demo_data.py：造 3 条演示预警覆盖三个只读工具的读路径 （高危待处理 / 中危调查中 / 低危已闭环），字段取值照 risk_scan_service.py:312-333 的 _build_alert 抄、状态用 OPEN_STATUSES，时间按库内约定存 UTC。注意该表 id 非自增， 所以脚本手工生成 id。 实测（9002 身份）： - 查看当前风险概览 → 未闭环 2 条、高危 2 条、待处理 1 条，含高优先级清单与证据摘要 - 查询高风险预警 → 2 条明细，命中 RW-002/003/007/012/015，附只读复核草案 - 查询 ALDEMO0001 的证据 → 完整快照事实 + 客户维度，并主动指出证据缺口 客服回归：9001 身份行为不变。
62 lines
3.1 KiB
Python
62 lines
3.1 KiB
Python
from datetime import UTC, datetime
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.core.errors import UnauthorizedAgentError
|
||
|
||
|
||
class IdentityRepository:
|
||
"""Read existing RBAC tables; never authorize from client role claims."""
|
||
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self.session = session
|
||
|
||
async def load_context(self, identity: RequestContext) -> RequestContext:
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
params = {"user_id": int(identity.user_id), "now": now}
|
||
status = await self.session.scalar(
|
||
text("SELECT status FROM sys_user WHERE id=:user_id"), params
|
||
)
|
||
if status != "正常":
|
||
raise UnauthorizedAgentError("账号不存在或未启用")
|
||
rows = (await self.session.execute(text("""
|
||
SELECT r.role_code, p.permission_code, p.data_scope
|
||
FROM sys_user_role ur JOIN sys_role r ON r.id=ur.role_id
|
||
LEFT JOIN sys_role_permission rp ON rp.role_id=r.id
|
||
LEFT JOIN sys_permission p ON p.id=rp.permission_id
|
||
WHERE ur.user_id=:user_id AND r.status='active'
|
||
AND ur.assigned_at<=:now AND (ur.expires_at IS NULL OR ur.expires_at>:now)
|
||
"""), params)).mappings().all()
|
||
roles = tuple(sorted({str(row["role_code"]) for row in rows}))
|
||
scopes: dict[str, str] = {}
|
||
rank = {"self": 0, "own_customers": 1, "all": 2}
|
||
for row in rows:
|
||
if row["permission_code"] and row["data_scope"] in rank:
|
||
code, scope = str(row["permission_code"]), str(row["data_scope"])
|
||
if code not in scopes or rank[scope] > rank[scopes[code]]:
|
||
scopes[code] = scope
|
||
customers = (await self.session.scalars(text("""
|
||
SELECT customer_id FROM sys_customer_assignment
|
||
WHERE employee_id=:user_id AND assigned_at<=:now
|
||
AND (unassigned_at IS NULL OR unassigned_at>:now)
|
||
"""), params)).all()
|
||
# data_scope 取该身份所有权限里的**最高**范围。
|
||
#
|
||
# 原先这里写死 "self"，于是上面刚算出来的 scope 白算了：permission_scopes 里
|
||
# 明明有 all，data_scope 却永远是 self，凡是按 `context.data_scope == "all"`
|
||
# 判断能否看全量的路径全部走不通（风控的 risk_query_service.py:198、
|
||
# risk_analysis_service.py:126、risk_evidence_archive_service.py:218、
|
||
# risk_action_service.py:212 都是这样判断的）。实测 9002(risk_operator) 与
|
||
# 9003(admin) 拿着 all 级权限却什么都查不到。
|
||
#
|
||
# 没有 all 权限的角色行为不变（如 customer 仍是 self），所以这不放松任何既有边界。
|
||
data_scope = max(scopes.values(), key=lambda value: rank[value]) if scopes else "self"
|
||
return identity.model_copy(update={
|
||
"roles": roles, "permissions": tuple(sorted(scopes)),
|
||
"permission_scopes": scopes, "data_scope": data_scope,
|
||
"customer_ids": tuple(str(value) for value in customers),
|
||
"portal": "api",
|
||
})
|