app/core/security.py

"""Seguridad: hashing de contraseñas, emisión/validación de JWT y
dependencias de autenticación para FastAPI.
"""

from __future__ import annotations

from datetime import timedelta
from typing import Annotated

import bcrypt
import jwt
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer

from app.core.config import settings
from app.database import collections
from app.database.mongodb import get_collection
from app.database.objectid import to_object_id
from app.exceptions import AuthenticationError, AuthorizationError
from app.models.common import utcnow
from app.models.user import User

# bcrypt admite como máximo 72 bytes de contraseña.
_BCRYPT_MAX_BYTES = 72

oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_PREFIX}/auth/login")


# ---------------------------------------------------------------------------
# Contraseñas
# ---------------------------------------------------------------------------


def _encode_password(password: str) -> bytes:
    """Codifica la contraseña a bytes, truncando al límite de bcrypt."""
    return password.encode("utf-8")[:_BCRYPT_MAX_BYTES]


def hash_password(password: str) -> str:
    """Devuelve el hash bcrypt de una contraseña en texto plano."""
    return bcrypt.hashpw(_encode_password(password), bcrypt.gensalt()).decode("utf-8")


def verify_password(password: str, hashed_password: str) -> bool:
    """Comprueba si una contraseña coincide con su hash."""
    try:
        return bcrypt.checkpw(
            _encode_password(password), hashed_password.encode("utf-8")
        )
    except (ValueError, TypeError):
        return False


# ---------------------------------------------------------------------------
# JWT
# ---------------------------------------------------------------------------


def create_access_token(subject: str, expires_minutes: int | None = None) -> str:
    """Crea un token de acceso firmado para el ``subject`` indicado."""
    minutes = expires_minutes or settings.ACCESS_TOKEN_EXPIRE_MINUTES
    now = utcnow()
    payload = {
        "sub": subject,
        "iat": now,
        "exp": now + timedelta(minutes=minutes),
    }
    return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)


def decode_access_token(token: str) -> str:
    """Decodifica un token y devuelve su ``subject`` (id de usuario)."""
    try:
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
        )
    except jwt.PyJWTError as exc:
        raise AuthenticationError("Token inválido o expirado.") from exc

    subject = payload.get("sub")
    if not subject:
        raise AuthenticationError("Token inválido: falta el sujeto.")
    return subject


# ---------------------------------------------------------------------------
# Dependencias
# ---------------------------------------------------------------------------


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    """Resuelve el usuario autenticado a partir del token Bearer."""
    user_id = decode_access_token(token)
    collection = get_collection(collections.USERS)
    document = await collection.find_one({"_id": to_object_id(user_id)})
    if document is None:
        raise AuthenticationError("El usuario del token ya no existe.")
    return User(**document)


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    """Garantiza que el usuario autenticado está activo."""
    if not current_user.is_active:
        raise AuthorizationError("La cuenta está desactivada.")
    return current_user


async def get_current_superuser(
    current_user: Annotated[User, Depends(get_current_active_user)],
) -> User:
    """Garantiza que el usuario autenticado es superusuario."""
    if not current_user.is_superuser:
        raise AuthorizationError("Se requieren privilegios de administrador.")
    return current_user