feat(roadmap): implementar Sprints 1, 2 y 3 (MVC, Stateless JWT API y Spring Security)

app/services/jwt_service.py

@@ -0,0 +1,116 @@
+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