|
|
from datetime import datetime, timedelta |
|
|
from typing import Optional |
|
|
from jose import JWTError, jwt |
|
|
from passlib.context import CryptContext |
|
|
from fastapi import HTTPException, status |
|
|
import os |
|
|
|
|
|
|
|
|
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key-change-in-production-123456789") |
|
|
ALGORITHM = "HS256" |
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 1440 |
|
|
|
|
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") |
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool: |
|
|
"""Verificar password contra hash""" |
|
|
return pwd_context.verify(plain_password, hashed_password) |
|
|
|
|
|
def get_password_hash(password: str) -> str: |
|
|
"""Hashear password con bcrypt""" |
|
|
return pwd_context.hash(password) |
|
|
|
|
|
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): |
|
|
"""Crear JWT token con expiración""" |
|
|
to_encode = data.copy() |
|
|
|
|
|
if expires_delta: |
|
|
expire = datetime.utcnow() + expires_delta |
|
|
else: |
|
|
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) |
|
|
|
|
|
to_encode.update({"exp": expire, "iat": datetime.utcnow()}) |
|
|
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) |
|
|
return encoded_jwt |
|
|
|
|
|
def verify_token(token: str) -> Optional[str]: |
|
|
"""Verificar JWT token y devolver username""" |
|
|
try: |
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) |
|
|
username: str = payload.get("sub") |
|
|
if username is None: |
|
|
return None |
|
|
return username |
|
|
except JWTError: |
|
|
return None |
|
|
|
|
|
def decode_token(token: str) -> dict: |
|
|
"""Decodificar token completo para debugging""" |
|
|
try: |
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) |
|
|
return payload |
|
|
except JWTError as e: |
|
|
raise HTTPException( |
|
|
status_code=status.HTTP_401_UNAUTHORIZED, |
|
|
detail=f"Token invalid: {str(e)}", |
|
|
headers={"WWW-Authenticate": "Bearer"}, |
|
|
) |
|
|
|
|
|
def get_token_from_header(authorization: str) -> Optional[str]: |
|
|
"""Extraer token del header Authorization""" |
|
|
if not authorization or not authorization.startswith("Bearer "): |
|
|
return None |
|
|
return authorization.split(" ")[1] |
|
|
|
|
|
def validate_password_strength(password: str) -> bool: |
|
|
"""Validar fortaleza de password (mínimo 6 caracteres para testing)""" |
|
|
if len(password) < 6: |
|
|
return False |
|
|
|
|
|
return True |
|
|
|
