app/api/dependencies/internal_auth.py

import secrets
from typing import Annotated

from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader

from app.core.config import settings


sync_token_header = APIKeyHeader(
    name="X-Sync-Token",
    scheme_name="OpportunitySyncToken",
    description=(
        "Token interne autorisant la synchronisation "
        "automatique des opportunités."
    ),
    auto_error=False,
)


async def require_sync_token(
    provided_token: Annotated[
        str | None,
        Depends(sync_token_header),
    ],
) -> None:
    """
    Protège les opérations internes de synchronisation.

    Le token reçu dans l’en-tête X-Sync-Token est comparé
    au secret SYNC_API_KEY configuré sur le serveur.
    """

    configured_secret = settings.sync_api_key

    if configured_secret is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Opportunity synchronization is not configured.",
        )

    expected_token = configured_secret.get_secret_value()

    if not expected_token:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Opportunity synchronization is not configured.",
        )

    if (
        provided_token is None
        or not secrets.compare_digest(
            provided_token,
            expected_token,
        )
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing synchronization token.",
            headers={
                "WWW-Authenticate": "ApiKey",
            },
        )


SyncTokenDependency = Annotated[
    None,
    Depends(require_sync_token),
]