27 lines
1.0 KiB
Python
27 lines
1.0 KiB
Python
|
from datetime import UTC, datetime
|
||
|
|
|
||
|
|
from app.core.contracts import RequestContext
|
||
|
|
from app.core.errors import ForbiddenAgentError
|
||
|
|
from app.infrastructure.db import SessionFactory
|
||
|
|
from app.model.audit import InteractionAudit
|
||
|
|
|
||
|
|
|
||
|
|
class AuthorizationService:
|
||
|
|
@staticmethod
|
||
|
|
async def require(
|
||
|
|
context: RequestContext, permission: str, *, admin: bool = False
|
||
|
|
) -> None:
|
||
|
|
allowed = permission in context.permissions
|
||
|
|
if admin:
|
||
|
|
allowed = allowed and bool({"admin", "super_admin"}.intersection(context.roles))
|
||
|
|
if allowed:
|
||
|
|
return
|
||
|
|
async with SessionFactory() as session, session.begin():
|
||
|
|
session.add(InteractionAudit(
|
||
|
|
actor_type="user", actor_id=int(context.user_id), portal=context.portal,
|
||
|
|
action_type="permission.denied", detail={
|
||
|
|
"permission": permission, "trace_id": context.trace_id,
|
||
|
|
}, created_at=datetime.now(UTC).replace(tzinfo=None),
|
||
|
|
))
|
||
|
|
raise ForbiddenAgentError("缺少操作权限")
|