pedidos-express/auth/security.py

from typing import Union
from venv import logger
from fastapi import Request, HTTPException, Header, Depends
from typing import Annotated
import secrets
from logging import getLogger

logger = getLogger(__name__)

async def get_session_token(request: Request) -> Union[str, None]:
    """Get the anti-abuse token from the session"""
    return request.session.get("antiAbuseToken")


async def protect_chat_api(
    request: Request,
    x_app_token: Annotated[Union[str, None], Header(alias="X-App-Token")] = None,
    session_token: Annotated[Union[str, None], Depends(get_session_token)] = None
):
    """Protect chat API endpoints with token validation"""
    # Equivalent to protectChatAPI middleware
    if not session_token:
        if request.client:
            logger.error(f"Session token is not initialized or invalid. IP: {request.client.host}")
        else:
            logger.error("Session token is not initialized or invalid.")
        logger.error("Session token is not initialized or invalid.")
        raise HTTPException(status_code=403, detail="Acceso denegado: Sesión inválida o token no inicializado.")

    if not x_app_token:
        if request.client:
            logger.error(f"X-App-Token is missing. IP: {request.client.host}")
        else:
            logger.error("X-App-Token is missing.")
        raise HTTPException(status_code=401, detail="Acceso denegado: Falta el token X-Chat-Token.")

    if x_app_token != session_token:
        # Log this attempt for security monitoring
        logger.warning(f"Invalid token attempt. Expected: {session_token}, Received: {x_app_token}")
        raise HTTPException(status_code=403, detail="Acceso denegado: Token inválido.")
    return True  # Protection passed


def generate_anti_abuse_token() -> str:
    """Generate a new anti-abuse token"""
    return secrets.token_hex(32)