mrleo1nid
c8d83b3fe7
Add immutable revisions, PGVN editor, asset replacement, merge and rebase workflows, authenticated media delivery and reachability GC. Harden the script interpreter, pin unrpyc and remove tracked build metadata. Include current compatibility checks, Naninovel import and regression tests.
55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
from collections.abc import Callable
|
|
from typing import Annotated
|
|
|
|
from fastapi import Cookie, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db import get_session
|
|
from app.models import User
|
|
from app.security import COOKIE_NAME, decode_token
|
|
|
|
DB = Annotated[AsyncSession, Depends(get_session)]
|
|
|
|
|
|
async def _load_user(db: AsyncSession, token: str | None) -> User:
|
|
if not token or not (decoded := decode_token(token)):
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Не авторизован")
|
|
user_id, token_version = decoded
|
|
user = await db.get(User, user_id)
|
|
if not user or not user.is_active or user.token_version != token_version:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Сессия недействительна")
|
|
return user
|
|
|
|
|
|
async def session_user(
|
|
db: DB, token: Annotated[str | None, Cookie(alias=COOKIE_NAME)] = None
|
|
) -> User:
|
|
"""Authenticated user, even if they still have to change credentials."""
|
|
return await _load_user(db, token)
|
|
|
|
|
|
async def current_user(user: Annotated[User, Depends(session_user)], db: DB) -> User:
|
|
"""Authenticated user who has finished the forced credential change."""
|
|
if user.must_change_password or user.must_change_username:
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN, "Необходимо сменить учётные данные")
|
|
db.info["user_id"] = user.id
|
|
return user
|
|
|
|
|
|
SessionUser = Annotated[User, Depends(session_user)]
|
|
CurrentUser = Annotated[User, Depends(current_user)]
|
|
|
|
|
|
def require(*perms: str) -> Callable:
|
|
async def checker(user: CurrentUser) -> User:
|
|
missing = [p for p in perms if p not in user.permissions]
|
|
if missing:
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN, f"Нет прав: {', '.join(missing)}")
|
|
return user
|
|
|
|
return checker
|
|
|
|
|
|
def has_perm(user: User, perm: str) -> bool:
|
|
return perm in user.permissions
|