src/utils/security.py

import bcrypt
from datetime import datetime, timedelta
from typing import Optional, Union
import uuid
import hashlib
from jose import JWTError, jwt
from ..config import settings


def _prepare_password(password: str) -> str:
    """
    Prepare password for bcrypt by hashing if longer than 72 bytes.
    This avoids truncation and loss of information.
    """
    password_bytes = password.encode('utf-8')
    if len(password_bytes) > 72:
        # Hash long passwords with SHA256 first to bring them within bcrypt's limit
        password_hash = hashlib.sha256(password_bytes).hexdigest()
        return password_hash
    return password


def hash_password(password: str) -> str:
    """Hash a password using bcrypt, handling long passwords via SHA256 pre-hashing."""
    prepared = _prepare_password(password)
    password_bytes = prepared.encode('utf-8')
    salt = bcrypt.gensalt(rounds=12)
    return bcrypt.hashpw(password_bytes, salt).decode('utf-8')


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a plain password against its hash, handling long passwords via SHA256 pre-hashing."""
    prepared = _prepare_password(plain_password)
    password_bytes = prepared.encode('utf-8')
    hashed_bytes = hashed_password.encode('utf-8')
    try:
        return bcrypt.checkpw(password_bytes, hashed_bytes)
    except (ValueError, TypeError):
        # Hash format is invalid
        return False


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """Create a JWT access token."""
    to_encode = data.copy()

    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        # Default to 7 days if no expiration is provided
        expire = datetime.utcnow() + timedelta(days=settings.ACCESS_TOKEN_EXPIRE_DAYS)

    to_encode.update({"exp": expire, "iat": datetime.utcnow()})

    encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
    return encoded_jwt


def verify_token(token: str) -> Optional[dict]:
    """Verify a JWT token and return the payload if valid."""
    try:
        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
        return payload
    except JWTError:
        return None


def verify_user_id_from_token(token: str) -> Optional[uuid.UUID]:
    """Extract user_id from JWT token."""
    payload = verify_token(token)
    if payload:
        user_id_str = payload.get("sub")
        if user_id_str:
            try:
                return uuid.UUID(user_id_str)
            except ValueError:
                return None
    return None