160 lines
5.1 KiB
Python
160 lines
5.1 KiB
Python
from datetime import datetime, timedelta
|
|
from typing import Optional, List
|
|
import os
|
|
#========================
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
|
|
|
|
security = HTTPBearer()
|
|
|
|
SECRET_KEY = os.getenv("SECRET_KEY", "xma-secret-key-change-in-production")
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_HOURS = int(os.getenv("ACCESS_TOKEN_EXPIRE_HOURS", "1"))
|
|
|
|
|
|
class UserType:
|
|
""" User type with permisions"""
|
|
|
|
ROOT ="root"
|
|
ADMIN ="admin"
|
|
PRIVILEGED ="privileged"
|
|
ADMIN_LICENCIAS ="admin_licencias"
|
|
UPDATER = "updater"
|
|
USER ="user"
|
|
|
|
|
|
HIERARCHY = {
|
|
ROOT :6,
|
|
ADMIN :5,
|
|
ADMIN_LICENCIAS : 4,
|
|
PRIVILEGED : 3,
|
|
UPDATER : 2,
|
|
USER : 1,
|
|
}
|
|
|
|
@classmethod
|
|
def has_permission(cls, user_type: str, required_type: str) -> bool:
|
|
""" VERIFY if user_type has permition equal o plus required_type"""
|
|
return cls.HIERARCHY.get(user_type, 0) >= cls.HIERARCHY.get(required_type, 0)
|
|
|
|
class OperationalRole:
|
|
""" Role's"""
|
|
COMPRAS ="compras"
|
|
VENTAS ="ventas"
|
|
LOGISTICA ="logistica"
|
|
ADUANA_SOFT ="aduana_soft"
|
|
CLIENTE ="cliente"
|
|
OPERATIVO ="operativo"
|
|
|
|
class AuthService:
|
|
""" Service auth"""
|
|
|
|
def __init__(self):
|
|
self.secret_key = SECRET_KEY
|
|
self.algorithm = ALGORITHM
|
|
|
|
|
|
def create_token(self, user_id: int, email: str, user_type: str, rol_operativo: str) -> str:
|
|
|
|
expire = datetime.utcnow() + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"email": email,
|
|
"user_type": user_type,
|
|
"rol_operativo": rol_operativo,
|
|
"exp": expire,
|
|
"iat": datetime.utcnow()
|
|
}
|
|
return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
|
|
|
|
def verify_token(self, token: str) -> dict:
|
|
|
|
try:
|
|
payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
raise HTTPException(status_code=401, detail="Token expirado")
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(status_code=401, detail="Token inválido")
|
|
|
|
def decode_token(self, token: str) -> dict:
|
|
|
|
try:
|
|
return jwt.decode(token, self.secret_key, algorithms=[self.algorithm], options={"verify_exp": False})
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(status_code=401, detail="Token inválido")
|
|
|
|
class CurrentUser:
|
|
"""Modelo del usuario autenticado"""
|
|
|
|
def __init__(self, id: int, email: str, user_type: str, rol_operativo: str):
|
|
self.id = id
|
|
self.email = email
|
|
self.user_type = user_type
|
|
self.rol_operativo = rol_operativo
|
|
self.ip_address = None
|
|
|
|
def is_root(self) -> bool:
|
|
return self.user_type == UserType.ROOT
|
|
|
|
def is_admin(self) -> bool:
|
|
return self.user_type == UserType.ADMIN
|
|
|
|
def is_admin_licencias(self) -> bool:
|
|
return self.user_type == UserType.ADMIN_LICENCIAS
|
|
|
|
def has_permission(self, required_type: str) -> bool:
|
|
"""Verifica si el usuario tiene el tipo requerido o superior"""
|
|
return UserType.has_permission(self.user_type, required_type)
|
|
|
|
def can_access_enterprise(self, enterprise_id: int, user_enterprise_id: int = None) -> bool:
|
|
"""Verifica si puede acceder a una empresa"""
|
|
if self.is_root():
|
|
return True
|
|
if user_enterprise_id and self.user_type == UserType.ADMIN:
|
|
return enterprise_id == user_enterprise_id
|
|
return False
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
auth_service: AuthService = Depends(lambda: AuthService())
|
|
) -> CurrentUser:
|
|
"""Dependencia para obtener el usuario actual desde el token"""
|
|
token = credentials.credentials
|
|
payload = auth_service.verify_token(token)
|
|
|
|
user = CurrentUser(
|
|
id=int(payload.get("sub")),
|
|
email=payload.get("email"),
|
|
user_type=payload.get("user_type", UserType.USER),
|
|
rol_operativo=payload.get("rol_operativo", OperationalRole.OPERATIVO)
|
|
)
|
|
|
|
return user
|
|
|
|
|
|
def require_user_type(required_type: str):
|
|
"""Dependencia para requerir un tipo de usuario específico"""
|
|
async def dependency(current_user: CurrentUser = Depends(get_current_user)):
|
|
if not current_user.has_permission(required_type):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"Se requiere tipo de usuario: {required_type}"
|
|
)
|
|
return current_user
|
|
return dependency
|
|
|
|
|
|
def require_rol(required_rol: str):
|
|
"""Dependencia para requerir un rol operativo específico"""
|
|
async def dependency(current_user: CurrentUser = Depends(get_current_user)):
|
|
if current_user.rol_operativo != required_rol and not current_user.is_root():
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"Se requiere rol operativo: {required_rol}"
|
|
)
|
|
return current_user
|
|
return dependency |