mrleo1nidandClaude Opus 5.5
364b312064
- FastAPI backend: admin bootstrap with forced credential change, admin-created users with one-time generated passwords, roles as permission sets - Games and versions with content-addressed blob storage; forks copy only the manifest, so mods share assets with the original - Worker: archive/RPA extraction, unrpyc for .rpyc, own .rpy parser, Python subset translated via ast, media converted to WebP/Opus/WebM - React panel (library search, game page, upload, users, roles) and a framework-free web player with saves, history, skip/auto, transitions - docker compose: postgres, backend, worker, nginx on a single HTTP port Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
54 lines
1.9 KiB
Python
54 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)]) -> 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, "Необходимо сменить учётные данные")
|
|
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
|