80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, Request
|
|
|
|
from backend.features.accounts.models import Principal
|
|
from backend.http.errors import AppError
|
|
|
|
SESSION_COOKIE = "xiaobai_session"
|
|
CSRF_COOKIE = "xiaobai_csrf"
|
|
|
|
|
|
def require_principal(request: Request) -> Principal:
|
|
service = request.app.state.container.accounts
|
|
principal = service.authenticate(request.cookies.get(SESSION_COOKIE))
|
|
if principal is None:
|
|
raise AppError("authentication_required", "请先登录。", 401)
|
|
return principal
|
|
|
|
|
|
AuthenticatedPrincipal = Annotated[Principal, Depends(require_principal)]
|
|
|
|
|
|
def require_csrf(
|
|
request: Request,
|
|
principal: AuthenticatedPrincipal,
|
|
) -> Principal:
|
|
valid = request.app.state.container.accounts.verify_csrf(
|
|
principal,
|
|
request.headers.get("X-CSRF-Token"),
|
|
request.cookies.get(CSRF_COOKIE),
|
|
)
|
|
if not valid:
|
|
raise AppError("csrf_failed", "页面状态已过期，请刷新后重试。", 403)
|
|
return principal
|
|
|
|
|
|
CsrfPrincipal = Annotated[Principal, Depends(require_csrf)]
|
|
|
|
|
|
def require_admin(principal: AuthenticatedPrincipal) -> Principal:
|
|
if not principal.user.is_admin:
|
|
raise AppError("access_denied", "当前账号无权执行此操作。", 403)
|
|
return principal
|
|
|
|
|
|
def require_admin_write(principal: CsrfPrincipal) -> Principal:
|
|
if not principal.user.is_admin:
|
|
raise AppError("access_denied", "当前账号无权执行此操作。", 403)
|
|
return principal
|
|
|
|
|
|
AdminPrincipal = Annotated[Principal, Depends(require_admin)]
|
|
AdminWritePrincipal = Annotated[Principal, Depends(require_admin_write)]
|
|
|
|
|
|
def require_smart_access(
|
|
request: Request,
|
|
principal: AuthenticatedPrincipal,
|
|
) -> Principal:
|
|
if not request.app.state.container.memberships.can_use_smart_features(principal):
|
|
raise AppError("membership_required", "该功能仅对会员开放。", 403)
|
|
return principal
|
|
|
|
|
|
SmartAccessPrincipal = Annotated[Principal, Depends(require_smart_access)]
|
|
|
|
|
|
def require_smart_write(
|
|
request: Request,
|
|
principal: CsrfPrincipal,
|
|
) -> Principal:
|
|
if not request.app.state.container.memberships.can_use_smart_features(principal):
|
|
raise AppError("membership_required", "该功能仅对会员开放。", 403)
|
|
return principal
|
|
|
|
|
|
SmartWritePrincipal = Annotated[Principal, Depends(require_smart_write)]
|