117 lines
4.6 KiB
Python
117 lines
4.6 KiB
Python
import jwt
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Dict, Any, Optional
|
|
from flask import current_app
|
|
from app.models.user import User
|
|
|
|
# Lista negra de tokens revocados en memoria (jti -> timestamp de expiración)
|
|
_token_blacklist: Dict[str, datetime] = {}
|
|
|
|
class JWTService:
|
|
"""Servicio de emisión, verificación y revocación de tokens JWT (Dual Token Pattern)."""
|
|
|
|
ACCESS_TOKEN_EXPIRES = timedelta(minutes=15)
|
|
REFRESH_TOKEN_EXPIRES = timedelta(days=7)
|
|
ALGORITHM = "HS256"
|
|
|
|
@classmethod
|
|
def _clean_expired_blacklist(cls):
|
|
"""Limpia periódicamente identificadores de tokens que ya han expirado naturalmente."""
|
|
now = datetime.now(timezone.utc)
|
|
expired_jtis = [jti for jti, exp in _token_blacklist.items() if exp < now]
|
|
for jti in expired_jtis:
|
|
_token_blacklist.pop(jti, None)
|
|
|
|
@classmethod
|
|
def _get_secret(cls) -> str:
|
|
"""Obtiene la clave secreta desde la configuración de Flask."""
|
|
return current_app.config.get('SECRET_KEY', 'default-edu-space-secret-key')
|
|
|
|
@classmethod
|
|
def create_access_token(cls, user: User, expires_delta: Optional[timedelta] = None) -> str:
|
|
"""Emite un token de acceso de corta duración (15 min) con claims de rol y permisos."""
|
|
now = datetime.now(timezone.utc)
|
|
delta = expires_delta or cls.ACCESS_TOKEN_EXPIRES
|
|
role_name = user.role_obj.name if user.role_obj else user.role
|
|
|
|
# Mapeo de permisos para claims del token
|
|
permissions = {}
|
|
if user.role_obj and hasattr(user.role_obj, 'permissions'):
|
|
for p in user.role_obj.permissions:
|
|
permissions[p.module] = p.access_level
|
|
|
|
payload = {
|
|
'sub': str(user.id),
|
|
'email': user.email,
|
|
'name': user.name,
|
|
'role': role_name,
|
|
'is_admin': user.is_admin(),
|
|
'permissions': permissions,
|
|
'type': 'access',
|
|
'jti': str(uuid.uuid4()),
|
|
'iat': now,
|
|
'exp': now + delta
|
|
}
|
|
return jwt.encode(payload, cls._get_secret(), algorithm=cls.ALGORITHM)
|
|
|
|
@classmethod
|
|
def create_refresh_token(cls, user: User, expires_delta: Optional[timedelta] = None) -> str:
|
|
"""Emite un token de refresco de larga duración (7 días) para renovación silenciosa."""
|
|
now = datetime.now(timezone.utc)
|
|
delta = expires_delta or cls.REFRESH_TOKEN_EXPIRES
|
|
payload = {
|
|
'sub': str(user.id),
|
|
'type': 'refresh',
|
|
'jti': str(uuid.uuid4()),
|
|
'iat': now,
|
|
'exp': now + delta
|
|
}
|
|
return jwt.encode(payload, cls._get_secret(), algorithm=cls.ALGORITHM)
|
|
|
|
@classmethod
|
|
def generate_tokens(cls, user: User) -> Dict[str, Any]:
|
|
"""Genera el par de tokens (Access + Refresh) junto con los metadatos del usuario."""
|
|
access_token = cls.create_access_token(user)
|
|
refresh_token = cls.create_refresh_token(user)
|
|
return {
|
|
'access_token': access_token,
|
|
'refresh_token': refresh_token,
|
|
'token_type': 'Bearer',
|
|
'expires_in': int(cls.ACCESS_TOKEN_EXPIRES.total_seconds())
|
|
}
|
|
|
|
@classmethod
|
|
def decode_token(cls, token: str, expected_type: str = 'access') -> Dict[str, Any]:
|
|
"""
|
|
Decodifica y valida la firma, expiración y tipo del token.
|
|
Lanza jwt.PyJWTError si el token es inválido, expirado o fue revocado.
|
|
"""
|
|
cls._clean_expired_blacklist()
|
|
payload = jwt.decode(token, cls._get_secret(), algorithms=[cls.ALGORITHM])
|
|
|
|
jti = payload.get('jti')
|
|
if jti and jti in _token_blacklist:
|
|
raise jwt.InvalidTokenError("El token ha sido revocado.")
|
|
|
|
token_type = payload.get('type')
|
|
if token_type != expected_type:
|
|
raise jwt.InvalidTokenError(f"Tipo de token inválido. Se esperaba '{expected_type}', recibido '{token_type}'.")
|
|
|
|
return payload
|
|
|
|
@classmethod
|
|
def revoke_token(cls, token: str) -> bool:
|
|
"""Agrega el identificador del token a la lista negra hasta su fecha natural de expiración."""
|
|
try:
|
|
payload = jwt.decode(token, cls._get_secret(), algorithms=[cls.ALGORITHM], options={"verify_exp": False})
|
|
jti = payload.get('jti')
|
|
exp_ts = payload.get('exp')
|
|
if jti and exp_ts:
|
|
exp_dt = datetime.fromtimestamp(exp_ts, tz=timezone.utc)
|
|
_token_blacklist[jti] = exp_dt
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|