mirror of
https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git
synced 2026-02-27 14:51:19 +00:00
9a2aea038a
- Add pyproject.toml with uv and ruff configuration - Pin Python version to 3.13 via .python-version - Add Makefile commands: lint, format, fix - Apply ruff formatting to entire codebase - Remove unused imports (base64 in yookassa/simple_subscription) - Update .gitignore for new config files
112 lines
2.7 KiB
Python
112 lines
2.7 KiB
Python
"""JWT token handling for cabinet authentication."""
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
import jwt
|
|
|
|
from app.config import settings
|
|
|
|
|
|
JWT_ALGORITHM = 'HS256'
|
|
|
|
|
|
def create_access_token(user_id: int, telegram_id: int | None = None) -> str:
|
|
"""
|
|
Create a short-lived access token.
|
|
|
|
Args:
|
|
user_id: Database user ID
|
|
telegram_id: Telegram user ID (optional for email-only users)
|
|
|
|
Returns:
|
|
Encoded JWT access token
|
|
"""
|
|
expire_minutes = settings.get_cabinet_access_token_expire_minutes()
|
|
expires = datetime.utcnow() + timedelta(minutes=expire_minutes)
|
|
|
|
payload = {
|
|
'sub': str(user_id),
|
|
'type': 'access',
|
|
'exp': expires,
|
|
'iat': datetime.utcnow(),
|
|
}
|
|
|
|
# Добавляем telegram_id только если он есть
|
|
if telegram_id is not None:
|
|
payload['telegram_id'] = telegram_id
|
|
|
|
secret = settings.get_cabinet_jwt_secret()
|
|
return jwt.encode(payload, secret, algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def create_refresh_token(user_id: int) -> str:
|
|
"""
|
|
Create a long-lived refresh token.
|
|
|
|
Args:
|
|
user_id: Database user ID
|
|
|
|
Returns:
|
|
Encoded JWT refresh token
|
|
"""
|
|
expire_days = settings.get_cabinet_refresh_token_expire_days()
|
|
expires = datetime.utcnow() + timedelta(days=expire_days)
|
|
|
|
payload = {
|
|
'sub': str(user_id),
|
|
'type': 'refresh',
|
|
'exp': expires,
|
|
'iat': datetime.utcnow(),
|
|
}
|
|
|
|
secret = settings.get_cabinet_jwt_secret()
|
|
return jwt.encode(payload, secret, algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> dict[str, Any] | None:
|
|
"""
|
|
Decode and validate a JWT token.
|
|
|
|
Args:
|
|
token: JWT token string
|
|
|
|
Returns:
|
|
Decoded payload dict or None if invalid/expired
|
|
"""
|
|
try:
|
|
secret = settings.get_cabinet_jwt_secret()
|
|
return jwt.decode(token, secret, algorithms=[JWT_ALGORITHM])
|
|
except jwt.ExpiredSignatureError:
|
|
return None
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
|
|
def get_token_payload(token: str, expected_type: str = 'access') -> dict[str, Any] | None:
|
|
"""
|
|
Decode token and verify its type.
|
|
|
|
Args:
|
|
token: JWT token string
|
|
expected_type: Expected token type ("access" or "refresh")
|
|
|
|
Returns:
|
|
Decoded payload dict or None if invalid/expired/wrong type
|
|
"""
|
|
payload = decode_token(token)
|
|
|
|
if not payload:
|
|
return None
|
|
|
|
if payload.get('type') != expected_type:
|
|
return None
|
|
|
|
return payload
|
|
|
|
|
|
def get_refresh_token_expires_at() -> datetime:
|
|
"""Get the expiration datetime for a new refresh token."""
|
|
expire_days = settings.get_cabinet_refresh_token_expire_days()
|
|
return datetime.utcnow() + timedelta(days=expire_days)
|