smolkik-code
7c7c88621d
- FastAPI REST API with JWT auth - aiogram 3 Telegram bot with admin middleware - APScheduler daily tasks (expiry, reminders, revoke, sync) - SQLAlchemy 2 async ORM with Alembic migrations - Jinja2 admin panel (Dashboard, Users, Payments, Servers, Tariffs) - VPN provider abstraction with MockProvider - Stats service with revenue/subscription analytics - Docker Compose (PostgreSQL + Redis + app) - Healthcheck endpoint
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from jose import JWTError, jwt
|
|
from loguru import logger
|
|
|
|
from app.models.admin import Admin
|
|
from app.repositories.admin import AdminRepository
|
|
from app.settings import settings
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self, admin_repo: AdminRepository):
|
|
self.admin_repo = admin_repo
|
|
|
|
def create_access_token(self, telegram_id: int, role: str) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
payload = {
|
|
"sub": str(telegram_id),
|
|
"role": role,
|
|
"type": "access",
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(
|
|
(now + timedelta(minutes=settings.jwt_access_expire_minutes)).timestamp()
|
|
),
|
|
}
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
|
|
|
|
def create_refresh_token(self, telegram_id: int) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
payload = {
|
|
"sub": str(telegram_id),
|
|
"type": "refresh",
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(
|
|
(now + timedelta(days=settings.jwt_refresh_expire_days)).timestamp()
|
|
),
|
|
}
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
|
|
|
|
def decode_token(self, token: str) -> dict:
|
|
try:
|
|
payload = jwt.decode(
|
|
token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]
|
|
)
|
|
return payload
|
|
except JWTError as e:
|
|
logger.warning("JWT decode failed: {}", e)
|
|
raise ValueError("Invalid or expired token") from e
|
|
|
|
async def authenticate(self, telegram_id: int, secret_key: str) -> dict | None:
|
|
if secret_key != settings.jwt_secret:
|
|
logger.warning("Auth failed: invalid secret key for tg={}", telegram_id)
|
|
return None
|
|
|
|
admin = await self.admin_repo.get_by_telegram_id(telegram_id)
|
|
if admin is None:
|
|
logger.warning("Auth failed: admin not found tg={}", telegram_id)
|
|
return None
|
|
|
|
if not admin.is_active:
|
|
logger.warning("Auth failed: admin deactivated tg={}", telegram_id)
|
|
return None
|
|
|
|
access_token = self.create_access_token(telegram_id, admin.role.value)
|
|
refresh_token = self.create_refresh_token(telegram_id)
|
|
|
|
logger.info("Admin logged in: tg={} role={}", telegram_id, admin.role.value)
|
|
|
|
return {
|
|
"access_token": access_token,
|
|
"refresh_token": refresh_token,
|
|
"token_type": "bearer",
|
|
}
|
|
|
|
def refresh_access_token(self, refresh_token: str) -> dict:
|
|
payload = self.decode_token(refresh_token)
|
|
|
|
if payload.get("type") != "refresh":
|
|
raise ValueError("Invalid token type")
|
|
|
|
telegram_id = int(payload["sub"])
|
|
role = payload.get("role", "moderator")
|
|
|
|
new_access = self.create_access_token(telegram_id, role)
|
|
|
|
return {
|
|
"access_token": new_access,
|
|
"refresh_token": refresh_token,
|
|
"token_type": "bearer",
|
|
}
|