zhanghongyu_0626
70aa861983
- Added new modules for advisor compliance, KYC sessions, and script templates, enhancing the advisor agent's capabilities. - Implemented a comprehensive API structure under the `/api/advisor-agent` prefix, ensuring clear organization and access to new features. - Established database models and repositories for compliance rules and KYC sessions, facilitating robust data management. - Integrated exception handling and response models to improve error management and user feedback. - Updated settings to include new configurations for compliance and KYC features, ensuring flexibility and adaptability. This update significantly expands the advisor agent's functionality, providing essential tools for compliance and customer interaction while maintaining a structured API design.
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""deps.AuthContext → 投资顾问 Agent 服务视图 + 细粒度 permission（S4 接缝）。"""
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass, field
|
||
|
||
from fastapi import Depends, Request
|
||
|
||
from app.api.deps import AuthContext as DepsAuthContext
|
||
from app.api.deps import get_platform_auth_context
|
||
from app.service.audit_service import audit_service
|
||
from app.utils.exceptions import PermissionDenied
|
||
from app.utils.trace import current_trace
|
||
|
||
|
||
@dataclass
|
||
class AdvisorAuthContext:
|
||
"""顾问线服务层身份（对齐源分支 AuthContext 字段名）。"""
|
||
|
||
user_id: str
|
||
token_type: str
|
||
roles: list[str]
|
||
permissions: list[str] = field(default_factory=list)
|
||
trace_id: str = ""
|
||
customer_id: str | None = None
|
||
advisor_id: str | None = None
|
||
|
||
def has_role(self, role: str) -> bool:
|
||
return role in self.roles
|
||
|
||
|
||
def advisor_auth_from_deps(
|
||
ctx: DepsAuthContext,
|
||
*,
|
||
trace_id: str | None = None,
|
||
) -> AdvisorAuthContext:
|
||
tid = trace_id or current_trace() or ""
|
||
advisor_id = ctx.actor_id if ctx.token_type == "staff" and "advisor" in ctx.roles else None
|
||
return AdvisorAuthContext(
|
||
user_id=ctx.actor_id,
|
||
token_type=ctx.token_type,
|
||
roles=list(ctx.roles),
|
||
permissions=list(ctx.permissions),
|
||
trace_id=tid,
|
||
customer_id=ctx.customer_id,
|
||
advisor_id=advisor_id,
|
||
)
|
||
|
||
|
||
def _trace_from_request(request: Request) -> str:
|
||
return getattr(request.state, "trace_id", None) or current_trace() or "unknown"
|
||
|
||
|
||
def get_advisor_auth(
|
||
request: Request,
|
||
ctx: DepsAuthContext = Depends(get_platform_auth_context),
|
||
) -> AdvisorAuthContext:
|
||
return advisor_auth_from_deps(ctx, trace_id=_trace_from_request(request))
|
||
|
||
|
||
def require_advisor_permission(permission: str) -> Callable[..., AdvisorAuthContext]:
|
||
def dependency(
|
||
request: Request,
|
||
ctx: DepsAuthContext = Depends(get_platform_auth_context),
|
||
) -> AdvisorAuthContext:
|
||
auth = advisor_auth_from_deps(ctx, trace_id=_trace_from_request(request))
|
||
if "admin:all" in auth.permissions or permission in auth.permissions:
|
||
return auth
|
||
audit_service.record(
|
||
trace_id=auth.trace_id,
|
||
event_type="rbac_denied",
|
||
agent_type="advisor",
|
||
actor_id=auth.user_id,
|
||
decision=permission,
|
||
)
|
||
raise PermissionDenied("AUTH_403_PERMISSION", f"missing permission: {permission}")
|
||
|
||
return dependency
|