113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
from functools import wraps
|
|
from flask import request, jsonify, g
|
|
import jwt
|
|
from app.services.jwt_service import JWTService
|
|
from app.models.user import User
|
|
|
|
def jwt_required(f):
|
|
"""
|
|
Decorador para proteger endpoints REST mediante Bearer Token JWT.
|
|
Extrae y valida el token, inyectando el usuario en 'g.jwt_user' y el payload en 'g.jwt_payload'.
|
|
"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
auth_header = request.headers.get('Authorization', '')
|
|
token = None
|
|
|
|
if auth_header.startswith('Bearer '):
|
|
token = auth_header.split(' ', 1)[1].strip()
|
|
elif 'access_token' in request.cookies:
|
|
token = request.cookies.get('access_token')
|
|
|
|
if not token:
|
|
try:
|
|
from flask_login import current_user
|
|
if current_user.is_authenticated and current_user.is_active:
|
|
g.jwt_user = current_user
|
|
g.jwt_payload = {'sub': str(current_user.id), 'role': getattr(current_user, 'role', '')}
|
|
g.jwt_token = None
|
|
g.real_admin_user = current_user
|
|
g.is_impersonating = False
|
|
return f(*args, **kwargs)
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'error': 'Unauthorized',
|
|
'message': 'Encabezado Authorization con Bearer token no proporcionado.'
|
|
}), 401
|
|
|
|
try:
|
|
payload = JWTService.decode_token(token, expected_type='access')
|
|
user_id = int(payload.get('sub'))
|
|
from app import db
|
|
user = db.session.get(User, user_id)
|
|
|
|
if not user or not user.is_active:
|
|
return jsonify({
|
|
'error': 'Unauthorized',
|
|
'message': 'Usuario no encontrado o cuenta desactivada.'
|
|
}), 401
|
|
|
|
g.jwt_user = user
|
|
g.jwt_payload = payload
|
|
g.jwt_token = token
|
|
g.real_admin_user = user
|
|
g.is_impersonating = False
|
|
|
|
impersonate_header = request.headers.get('X-Impersonate-User')
|
|
if impersonate_header and user.is_admin():
|
|
target_user = None
|
|
if impersonate_header.strip().isdigit():
|
|
target_user = db.session.get(User, int(impersonate_header.strip()))
|
|
else:
|
|
target_user = User.query.filter_by(email=impersonate_header.strip()).first()
|
|
|
|
if target_user and target_user.is_active:
|
|
g.jwt_user = target_user
|
|
g.is_impersonating = True
|
|
|
|
except jwt.ExpiredSignatureError:
|
|
return jsonify({
|
|
'error': 'TokenExpired',
|
|
'message': 'El token de acceso ha expirado. Por favor renueve su sesión.'
|
|
}), 401
|
|
except jwt.InvalidTokenError as e:
|
|
return jsonify({
|
|
'error': 'InvalidToken',
|
|
'message': str(e)
|
|
}), 401
|
|
except Exception as e:
|
|
return jsonify({
|
|
'error': 'AuthenticationError',
|
|
'message': f'Fallo en la verificación del token: {str(e)}'
|
|
}), 401
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
def jwt_role_required(*allowed_roles):
|
|
"""
|
|
Decorador que verifica que el usuario autenticado vía JWT posea alguno de los roles permitidos.
|
|
"""
|
|
def decorator(f):
|
|
@wraps(f)
|
|
@jwt_required
|
|
def decorated_function(*args, **kwargs):
|
|
user: User = g.jwt_user
|
|
if user.is_admin():
|
|
return f(*args, **kwargs)
|
|
|
|
user_role = (user.role_obj.name if user.role_obj else user.role).upper()
|
|
allowed = [r.upper() for r in allowed_roles]
|
|
|
|
if user_role not in allowed:
|
|
return jsonify({
|
|
'error': 'Forbidden',
|
|
'message': f'Acceso denegado. Se requiere uno de los siguientes roles: {", ".join(allowed_roles)}'
|
|
}), 403
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
return decorator
|