zhanghongyu_0626
3995cb44d8
- Added `auth.py` for mock login and JWT issuance. - Introduced `chat.py` for handling chat requests with role-based access control. - Enhanced `main.py` to include new routers and middleware for tracing. - Implemented input validation in `input_guard.py` to prevent SQL injection. - Created repositories for managing agent sessions and audit logs. - Added exception handling for authorization errors. - Updated settings to include JWT configuration. - Introduced tests for authentication and input validation.
36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
"""RBAC：Agent 入口与角色准入。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from app.model.schemas import AgentType, AuthContext
|
||
from app.utils.exceptions import ForbiddenError
|
||
|
||
AGENT_ACCESS: dict[AgentType, dict[str, set[str]]] = {
|
||
"customer": {"token_types": {"customer"}, "roles": {"customer"}},
|
||
"advisor": {"token_types": {"staff"}, "roles": {"advisor", "compliance", "ops"}},
|
||
"analyst": {"token_types": {"staff"}, "roles": {"analyst", "compliance"}},
|
||
"risk": {"token_types": {"staff", "service"}, "roles": {"risk_officer", "service_risk"}},
|
||
}
|
||
|
||
|
||
def assert_agent_access(ctx: AuthContext) -> None:
|
||
rule = AGENT_ACCESS[ctx.agent_type]
|
||
if ctx.token_type not in rule["token_types"]:
|
||
raise ForbiddenError(
|
||
"Token 类型与 Agent 不匹配",
|
||
error_code="AUTH_403_AGENT_MISMATCH",
|
||
)
|
||
if not rule["roles"].intersection(ctx.roles):
|
||
raise ForbiddenError(
|
||
"角色无权访问该 Agent",
|
||
error_code="AUTH_403_ROLE",
|
||
)
|
||
perm = f"agent:{ctx.agent_type}:chat"
|
||
if ctx.agent_type == "risk" and ctx.token_type == "service":
|
||
return
|
||
if perm not in ctx.permissions and not ctx.has_perm(perm):
|
||
raise ForbiddenError(
|
||
"缺少 Agent 对话权限",
|
||
error_code="AUTH_403_PERMISSION",
|
||
)
|