office_translator/middleware/api_key_auth.py

"""
API Key Authentication Middleware

Provides reusable dependencies for API key authentication across all endpoints.
Story 3.4: Authentification API via X-API-Key
Story 6.4: Bind user_id to structlog context when user is resolved (logs include user_id).
"""

from typing import Optional, Any, Union
from fastapi import Header, Depends
from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

security = HTTPBearer(auto_error=False)


class APIKeyError(Exception):
    """Exception for API key authentication errors with structured error codes."""
    
    INVALID_API_KEY = "INVALID_API_KEY"
    API_KEY_REVOKED = "API_KEY_REVOKED"
    API_KEY_EXPIRED = "API_KEY_EXPIRED"
    MISSING_API_KEY = "MISSING_API_KEY"
    UNAUTHORIZED = "UNAUTHORIZED"
    
    ERROR_MESSAGES = {
        INVALID_API_KEY: "Clé API invalide ou non reconnue.",
        API_KEY_REVOKED: "Cette clé API a été révoquée.",
        API_KEY_EXPIRED: "Cette clé API a expiré.",
        MISSING_API_KEY: "Clé API requise pour cet endpoint.",
        UNAUTHORIZED: "Authentification requise. Utilisez X-API-Key ou Authorization: Bearer.",
    }
    
    def __init__(self, code: str, message: Optional[str] = None):
        self.code = code
        self.message = message or self.ERROR_MESSAGES.get(code, "Erreur d'authentification")
        super().__init__(self.message)
    
    def to_response(self, status_code: int = 401) -> JSONResponse:
        """Convert to JSONResponse for FastAPI."""
        return JSONResponse(
            status_code=status_code,
            content={
                "error": self.code,
                "message": self.message,
            },
        )


def _raise_api_key_error(code: str, message: Optional[str] = None) -> None:
    """Raise an APIKeyError and convert it to JSONResponse for FastAPI."""
    raise APIKeyError(code, message)


async def get_user_from_api_key(
    x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
) -> Optional[Any]:
    """
    Get user from X-API-Key header if provided.
    
    Returns:
        User object if valid API key provided
        None if no API key provided (caller should try other auth methods)
    
    Raises:
        APIKeyError: With structured error code if API key is invalid/revoked/expired
    """
    if not x_api_key:
        return None
    
    try:
        from services.auth_service import get_user_by_api_key
        from core.logging import bind_request_context

        user = get_user_by_api_key(x_api_key)
        if user is not None:
            bind_request_context(user_id=str(getattr(user, "id", user)))
        return user
        
    except ValueError as e:
        # Handle revoked/expired API keys with specific error codes
        error_code = str(e)
        
        if error_code == "API_KEY_REVOKED":
            raise APIKeyError("API_KEY_REVOKED", "Cette clé API a été révoquée.")
        elif error_code == "API_KEY_EXPIRED":
            raise APIKeyError("API_KEY_EXPIRED", "Cette clé API a expiré.")
        else:
            # Unknown error - treat as invalid
            raise APIKeyError("INVALID_API_KEY", "Clé API invalide ou non reconnue.")
    
    except Exception:
        # Unexpected error - treat as invalid
        raise APIKeyError("INVALID_API_KEY", "Clé API invalide ou non reconnue.")


async def get_current_user_optional(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
) -> Optional[Any]:
    """Get current user if authenticated via JWT, None otherwise."""
    if not credentials:
        return None
    try:
        from routes.auth_routes import get_current_user
        
        user = await get_current_user(credentials)
        return user
    except Exception:
        return None


async def get_authenticated_user_optional(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
    x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
) -> Optional[Any]:
    """
    Get authenticated user from API key or JWT (optional - returns None if not authenticated).
    
    Priority:
    1. X-API-Key header (automation users)
    2. JWT Bearer token (web users)
    3. None (unauthenticated)
    
    Returns:
        User object if authenticated, None otherwise (never raises for auth failures)
    """
    # Try API key first (priority for automation)
    if x_api_key:
        try:
            user = await get_user_from_api_key(x_api_key)
            if user:
                from core.logging import bind_request_context
                bind_request_context(user_id=str(getattr(user, "id", user)))
                return user
        except APIKeyError:
            # Invalid API key, fall through to JWT
            pass

    # Fall back to JWT
    if credentials:
        try:
            from routes.auth_routes import get_current_user

            user = await get_current_user(credentials)
            if user:
                from core.logging import bind_request_context
                bind_request_context(user_id=str(getattr(user, "id", user)))
            return user
        except Exception:
            pass

    return None


async def get_authenticated_user(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
    x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
) -> Optional[Any]:
    """
    Get authenticated user from API key or JWT.
    
    Priority:
    1. X-API-Key header (automation users)
    2. JWT Bearer token (web users)
    3. None (unauthenticated)
    
    Returns:
        User object if authenticated
        None if not authenticated
    
    Raises:
        APIKeyError: If API key is provided but invalid/revoked/expired
    """
    # Try API key first (priority for automation)
    if x_api_key:
        # get_user_from_api_key will raise APIKeyError for invalid keys
        user = await get_user_from_api_key(x_api_key)
        if user:
            from core.logging import bind_request_context
            bind_request_context(user_id=str(getattr(user, "id", user)))
            return user
        # Should not reach here - get_user_from_api_key returns None only if no key provided
        raise APIKeyError("INVALID_API_KEY", "Clé API invalide ou non reconnue.")

    # Fall back to JWT
    if credentials:
        try:
            from routes.auth_routes import get_current_user

            user = await get_current_user(credentials)
            if user:
                from core.logging import bind_request_context
                bind_request_context(user_id=str(getattr(user, "id", user)))
            return user
        except Exception:
            pass

    return None


async def require_authenticated_user(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
    x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
) -> Any:
    """
    Require authentication (API key or JWT).
    
    Raises:
        APIKeyError: 401 if not authenticated
    
    Returns:
        User object (guaranteed to be authenticated)
    """
    user = await get_authenticated_user(credentials, x_api_key)

    if not user:
        raise APIKeyError("MISSING_API_KEY", "Authentification requise. Utilisez X-API-Key ou Authorization: Bearer.")

    from core.logging import bind_request_context
    bind_request_context(user_id=str(getattr(user, "id", user)))
    return user


async def require_api_key(
    x_api_key: str = Header(..., alias="X-API-Key"),
) -> Any:
    """
    Require API key authentication (no JWT fallback).
    
    Use this for endpoints that MUST use API key (e.g., certain automation endpoints).
    
    Raises:
        APIKeyError: 401 if API key is missing, invalid, revoked, or expired
    
    Returns:
        User object (guaranteed to be authenticated via API key)
    """
    return await get_user_from_api_key(x_api_key)