File size: 1,730 Bytes
f55a6b5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | 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),
] |