Initial commit
This commit is contained in:
271
backend/app/core/security.py
Normal file
271
backend/app/core/security.py
Normal file
@@ -0,0 +1,271 @@
|
||||
"""
|
||||
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
|
||||
|
||||
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)
|
||||
to_encode.update({"exp": expire, "type": "refresh"})
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user