273 lines
7.4 KiB
Python
273 lines
7.4 KiB
Python
"""
|
|
Security Utilities - ServiceManagerWeb
|
|
|
|
Funciones de seguridad para autenticación y autorización
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Union, Dict, Any
|
|
from passlib.context import CryptContext
|
|
from passlib.handlers.argon2 import argon2
|
|
from jose import JWTError, jwt
|
|
import pyotp
|
|
import secrets
|
|
import base64
|
|
import struct
|
|
import uuid
|
|
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
# Password hashing context
|
|
pwd_context = CryptContext(
|
|
schemes=["argon2"],
|
|
deprecated="auto",
|
|
argon2__time_cost=settings.ARGON2_TIME_COST,
|
|
argon2__memory_cost=settings.ARGON2_MEMORY_COST,
|
|
argon2__parallelism=settings.ARGON2_PARALLELISM,
|
|
)
|
|
|
|
|
|
class SecurityUtils:
|
|
"""Utilidades de seguridad centralizadas."""
|
|
|
|
@staticmethod
|
|
def hash_password(password: str) -> str:
|
|
"""
|
|
Hash a password using Argon2.
|
|
|
|
Args:
|
|
password: Plain text password
|
|
|
|
Returns:
|
|
Hashed password
|
|
"""
|
|
return pwd_context.hash(password)
|
|
|
|
@staticmethod
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""
|
|
Verify a password against its hash.
|
|
|
|
Args:
|
|
plain_password: Plain text password
|
|
hashed_password: Hashed password
|
|
|
|
Returns:
|
|
True if password matches
|
|
"""
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
@staticmethod
|
|
def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
|
|
"""
|
|
Create a JWT access token.
|
|
|
|
Args:
|
|
data: Token payload
|
|
expires_delta: Token expiration time
|
|
|
|
Returns:
|
|
JWT token string
|
|
"""
|
|
to_encode = data.copy()
|
|
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
|
|
to_encode.update({"exp": expire})
|
|
|
|
encoded_jwt = jwt.encode(
|
|
to_encode,
|
|
settings.JWT_SECRET_KEY,
|
|
algorithm=settings.JWT_ALGORITHM
|
|
)
|
|
|
|
return encoded_jwt
|
|
|
|
@staticmethod
|
|
def create_refresh_token(data: Dict[str, Any]) -> str:
|
|
"""
|
|
Create a JWT refresh token.
|
|
|
|
Args:
|
|
data: Token payload
|
|
|
|
Returns:
|
|
JWT refresh token string
|
|
"""
|
|
to_encode = data.copy()
|
|
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
|
|
# Add a unique identifier so refresh tokens are never deterministic.
|
|
to_encode.update({"exp": expire, "type": "refresh", "jti": str(uuid.uuid4())})
|
|
|
|
encoded_jwt = jwt.encode(
|
|
to_encode,
|
|
settings.JWT_SECRET_KEY,
|
|
algorithm=settings.JWT_ALGORITHM
|
|
)
|
|
|
|
return encoded_jwt
|
|
|
|
@staticmethod
|
|
def verify_token(token: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Verify and decode a JWT token.
|
|
|
|
Args:
|
|
token: JWT token string
|
|
|
|
Returns:
|
|
Token payload if valid, None otherwise
|
|
"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
settings.JWT_SECRET_KEY,
|
|
algorithms=[settings.JWT_ALGORITHM]
|
|
)
|
|
return payload
|
|
except JWTError:
|
|
return None
|
|
|
|
@staticmethod
|
|
def generate_totp_secret() -> str:
|
|
"""
|
|
Generate a base32-encoded secret for TOTP.
|
|
|
|
Returns:
|
|
Base32 encoded secret
|
|
"""
|
|
return pyotp.random_base32()
|
|
|
|
@staticmethod
|
|
def generate_totp_uri(secret: str, email: str, issuer_name: str = "ServiceManager") -> str:
|
|
"""
|
|
Generate TOTP URI for QR code.
|
|
|
|
Args:
|
|
secret: Base32 encoded secret
|
|
email: User email
|
|
issuer_name: Application name
|
|
|
|
Returns:
|
|
TOTP URI
|
|
"""
|
|
totp = pyotp.TOTP(secret)
|
|
return totp.provisioning_uri(
|
|
name=email,
|
|
issuer_name=issuer_name
|
|
)
|
|
|
|
@staticmethod
|
|
def verify_totp(secret: str, token: str, window: int = 1) -> bool:
|
|
"""
|
|
Verify a TOTP token.
|
|
|
|
Args:
|
|
secret: Base32 encoded secret
|
|
token: TOTP token
|
|
window: Time window tolerance
|
|
|
|
Returns:
|
|
True if token is valid
|
|
"""
|
|
totp = pyotp.TOTP(secret)
|
|
return totp.verify(token, valid_window=window)
|
|
|
|
@staticmethod
|
|
def generate_backup_codes(count: int = 8) -> list[str]:
|
|
"""
|
|
Generate backup codes for 2FA.
|
|
|
|
Args:
|
|
count: Number of codes to generate
|
|
|
|
Returns:
|
|
List of backup codes
|
|
"""
|
|
codes = []
|
|
for _ in range(count):
|
|
code = secrets.token_hex(4).upper()
|
|
# Format as XXXX-XXXX
|
|
formatted_code = f"{code[:4]}-{code[4:]}"
|
|
codes.append(formatted_code)
|
|
return codes
|
|
|
|
@staticmethod
|
|
def hash_token(token: str) -> str:
|
|
"""
|
|
Hash a token for secure storage.
|
|
|
|
Args:
|
|
token: Token to hash
|
|
|
|
Returns:
|
|
Hashed token
|
|
"""
|
|
return pwd_context.hash(token)
|
|
|
|
@staticmethod
|
|
def verify_hashed_token(token: str, hashed_token: str) -> bool:
|
|
"""
|
|
Verify a token against its hash.
|
|
|
|
Args:
|
|
token: Plain token
|
|
hashed_token: Hashed token
|
|
|
|
Returns:
|
|
True if token matches
|
|
"""
|
|
return pwd_context.verify(token, hashed_token)
|
|
|
|
@staticmethod
|
|
def generate_secure_token(length: int = 32) -> str:
|
|
"""
|
|
Generate a cryptographically secure random token.
|
|
|
|
Args:
|
|
length: Token length in bytes
|
|
|
|
Returns:
|
|
URL-safe base64 encoded token
|
|
"""
|
|
token = secrets.token_bytes(length)
|
|
return base64.urlsafe_b64encode(token).decode('utf-8').rstrip('=')
|
|
|
|
@staticmethod
|
|
def is_strong_password(password: str) -> tuple[bool, list[str]]:
|
|
"""
|
|
Check if password meets security requirements.
|
|
|
|
Args:
|
|
password: Password to check
|
|
|
|
Returns:
|
|
Tuple of (is_valid, list_of_issues)
|
|
"""
|
|
issues = []
|
|
|
|
if len(password) < settings.PASSWORD_MIN_LENGTH:
|
|
issues.append(f"Password must be at least {settings.PASSWORD_MIN_LENGTH} characters long")
|
|
|
|
if not any(c.islower() for c in password):
|
|
issues.append("Password must contain at least one lowercase letter")
|
|
|
|
if not any(c.isupper() for c in password):
|
|
issues.append("Password must contain at least one uppercase letter")
|
|
|
|
if not any(c.isdigit() for c in password):
|
|
issues.append("Password must contain at least one digit")
|
|
|
|
if not any(c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in password):
|
|
issues.append("Password must contain at least one special character")
|
|
|
|
return len(issues) == 0, issues
|
|
|
|
|
|
# Create singleton instance
|
|
security = SecurityUtils() |