services/api/app/modules/admin/auth.py - master - Mihaham/Table-Time

services/api/app/modules/admin/auth.py
124 строки4 KB
MihahamYT
feat: dedicated web panels for all 46 games
14 июн 2026, 21:36
14 июн 2026, 21:364e11c3f
import random
import secrets
from datetime import UTC, datetime, timedelta

from fastapi import Header, HTTPException
from jose import JWTError, jwt

from app.config import get_settings
from app.events.redis_pub import get_redis
from app.modules.admin.schemas import AdminContext

ALGORITHM = "HS256"


class AdminAuthService:
    def __init__(self) -> None:
        self.settings = get_settings()

    def _otp_key(self, telegram_id: int) -> str:
        return f"admin:otp:{telegram_id}"

    def _attempts_key(self, telegram_id: int) -> str:
        return f"admin:otp:attempts:{telegram_id}"

    def _rate_key(self, telegram_id: int) -> str:
        return f"admin:otp:rate:{telegram_id}"

    def _verify_ip_key(self, ip: str) -> str:
        return f"admin:otp:verify_ip:{ip}"

    async def check_verify_rate_limit(self, ip: str | None) -> None:
        if not ip:
            return
        r = await get_redis()
        key = self._verify_ip_key(ip)
        count = await r.incr(key)
        if count == 1:
            await r.expire(key, 900)
        if count > 10:
            raise ValueError("RATE_LIMITED")

    async def request_otp(self, telegram_id: int) -> tuple[str, int]:
        if not self.settings.admin_telegram_id:
            raise ValueError("ADMIN_NOT_CONFIGURED")
        if telegram_id != self.settings.admin_telegram_id:
            raise ValueError("NOT_ADMIN")

        r = await get_redis()
        rate_key = self._rate_key(telegram_id)
        rate_count = await r.incr(rate_key)
        if rate_count == 1:
            await r.expire(rate_key, 900)
        if rate_count > 3:
            raise ValueError("RATE_LIMITED")

        code = f"{random.randint(0, 999999):06d}"
        otp_key = self._otp_key(telegram_id)
        ttl = self.settings.admin_otp_ttl_seconds
        await r.setex(otp_key, ttl, code)
        await r.delete(self._attempts_key(telegram_id))
        return code, ttl

    async def verify_otp(self, username: str, code: str) -> dict:
        if not self.settings.admin_telegram_id:
            raise ValueError("ADMIN_NOT_CONFIGURED")
        if username.strip().lower() != self.settings.admin_username.lower():
            raise ValueError("INVALID_USERNAME")

        telegram_id = self.settings.admin_telegram_id
        r = await get_redis()
        otp_key = self._otp_key(telegram_id)
        attempts_key = self._attempts_key(telegram_id)

        stored = await r.get(otp_key)
        if not stored:
            raise ValueError("OTP_EXPIRED")

        attempts = await r.incr(attempts_key)
        if attempts == 1:
            await r.expire(attempts_key, self.settings.admin_otp_ttl_seconds)
        if attempts > 3:
            await r.delete(otp_key)
            raise ValueError("OTP_LOCKED")

        if not secrets.compare_digest(stored, code.strip()):
            raise ValueError("INVALID_CODE")

        await r.delete(otp_key)
        await r.delete(attempts_key)

        expires_hours = self.settings.admin_jwt_ttl_hours
        expire = datetime.now(UTC) + timedelta(hours=expires_hours)
        payload = {
            "sub": "admin",
            "type": "admin",
            "username": self.settings.admin_username,
            "exp": expire,
        }
        token = jwt.encode(payload, self.settings.jwt_secret, algorithm=ALGORITHM)
        return {
            "access_token": token,
            "token_type": "admin",
            "expires_in": expires_hours * 3600,
        }

    def parse_admin_token(self, authorization: str | None) -> AdminContext | None:
        if not authorization or not authorization.startswith("Bearer "):
            return None
        token = authorization[7:]
        try:
            payload = jwt.decode(token, self.settings.jwt_secret, algorithms=[ALGORITHM])
        except JWTError:
            return None
        if payload.get("type") != "admin":
            return None
        return AdminContext(username=payload.get("username", self.settings.admin_username))


async def get_current_admin(authorization: str | None = Header(None, alias="Authorization")) -> AdminContext:
    service = AdminAuthService()
    admin = service.parse_admin_token(authorization)
    if not admin:
        raise HTTPException(status_code=403, detail="Admin access required")
    return admin