64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
"""投顾灰度发布闸门。
|
|
|
|
灰度配置使用环境变量，避免为发布控制改动冻结的数据库基线。白名单只允许客户
|
|
身份进入投顾业务；管理员保留审核、发布和故障处置权限。未命中时统一复用权限拒绝
|
|
错误，不向客户暴露灰度配置细节。
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.dependencies.auth import build_request_context
|
|
from app.api.dependencies.database import get_session
|
|
from app.core.config import get_settings
|
|
from app.core.contracts import RequestContext
|
|
from app.core.errors import ForbiddenAgentError
|
|
from app.model.audit import InteractionAudit
|
|
|
|
|
|
class AdvisorRolloutService:
|
|
ADMIN_ROLES = frozenset({"admin", "super_admin"})
|
|
|
|
def __init__(self, session: AsyncSession | None = None) -> None:
|
|
self.session = session
|
|
|
|
@staticmethod
|
|
def customer_ids(raw: str) -> frozenset[str]:
|
|
return frozenset(item.strip() for item in raw.split(",") if item.strip())
|
|
|
|
@classmethod
|
|
def is_allowed(cls, context: RequestContext) -> bool:
|
|
settings = get_settings()
|
|
if not settings.advisor_rollout_enabled:
|
|
return True
|
|
if cls.ADMIN_ROLES.intersection(context.roles):
|
|
return True
|
|
allowed = cls.customer_ids(settings.advisor_rollout_customer_ids)
|
|
identities = {context.user_id, *context.customer_ids}
|
|
return bool(allowed.intersection(identities))
|
|
|
|
async def ensure_allowed(self, context: RequestContext) -> None:
|
|
if self.is_allowed(context):
|
|
return
|
|
if self.session is not None:
|
|
self.session.add(InteractionAudit(
|
|
actor_type="user",
|
|
actor_id=int(context.user_id),
|
|
portal=context.portal,
|
|
action_type="advisor.rollout_denied",
|
|
detail={"trace_id": context.trace_id, "reason": "not_in_rollout"},
|
|
created_at=datetime.now(UTC).replace(tzinfo=None),
|
|
))
|
|
await self.session.commit()
|
|
raise ForbiddenAgentError("当前投顾功能尚未对该账号开放")
|
|
|
|
|
|
async def enforce_advisor_rollout(
|
|
context: RequestContext = Depends(build_request_context), # noqa: B008
|
|
session: AsyncSession = Depends(get_session), # noqa: B008
|
|
) -> None:
|
|
"""FastAPI 依赖：保护投顾业务路由，认证依赖先完成身份解析。"""
|
|
await AdvisorRolloutService(session).ensure_allowed(context)
|