Protection de base

Ce que vous saurez faire

Rendez l’API assez sûre pour la mettre sur Internet. Seuls les appelants ayant la clé peuvent classifier. Un corps énorme ne peut pas épuiser la mémoire du serveur. Un classificateur lent ne peut pas laisser une requête attendre indéfiniment. Le service reste réactif, et chaque refus produit une erreur claire.

Avant de commencerCe qu’il faut savoir et avoir sous la mainEnviron 50 minutes

Ce que vous devez savoir

• Les secrets dans les variables d’environnement, dans Configuration et secrets : secrets.
• L’intergiciel d’identifiant de requête, dans Configuration et journalisation : un identifiant pour chaque requête.
• Les tests avec leurs propres paramètres, dans Configuration et journalisation : des tests avec leurs propres paramètres.

Ce qu’il vous faut sur votre ordinateur

• config.py, create_app dans main.py, l’identifiant de requête et tests/conftest.py, dans Configuration et journalisation : exercice guidé : paramètres et identifiants de requête. python -m pytest réussit.
• La collection et l’environnement Local dans le dossier postman, dans Tests et exécutions reproductibles : enregistrer la collection dans le projet.
• L’interface en ligne de commande de Postman, dans Tests et exécutions reproductibles : exécuter depuis le terminal : Postman CLI.
• Dans l’application Postman, la variable apiKey dans Local et l’autorisation par clé d’API de la collection. Vous les avez créées dans Collections, variables et environnements : exemple détaillé : deux environnements Postman.
• Si votre projet manque ou ne fonctionne plus, copiez l’instantané end-of-m05, le projet à la fin du module 5, comme dans Utiliser un instantané. Refaites ensuite les exercices guidés de Tests automatisés des points de terminaison et de Configuration et journalisation. Pour utiliser la collection dans l’application Postman, importez les fichiers de postman/ (Import, importer).

Testé avec Python 3.14.6, FastAPI 0.142.2, pytest 9.1.1 et l’interface en ligne de commande de Postman 1.69.0 sur macOS.

Trois risques, trois protections

Une fois l’API sur Internet, n’importe qui peut lui envoyer n’importe quoi. Trois problèmes risquent de survenir en premier :

Risque	Exemple	Protection
Des inconnus utilisent l’API	Quelqu’un trouve l’URL et classifie un million de billets de soutien avec votre compte	Une clé d’API que les appelants doivent envoyer
Un corps trop volumineux	Une requête contenant 500 Mo de texte épuise la mémoire du serveur	Une limite de taille du corps : 413
Une dépendance qui ne répond pas	Un serveur de modèle bloque. Chaque requête attend, et l’API ne répond plus à personne	Un délai maximal : 503 après 2 secondes

La clé d’API de cette leçon sert aux appels entre services. Un autre programme de votre entreprise, comme le système du soutien technique, appelle l’API avec une clé que vous lui avez donnée. Ce n’est pas une connexion pour les personnes. Les utilisateurs, les mots de passe et les permissions sont un autre problème, abordé dans le cours sur l’authentification.

La clé d’API

Une clé d’API est un long secret aléatoire qu’un client envoie dans un en-tête, ici X-API-Key. Le serveur la compare à la clé dans ses paramètres. Une clé incorrecte ou absente donne 401.

La clé est un paramètre et un secret, comme dans le cours de Python. Elle vient de la variable d’environnement API_KEY, et field(repr=False) l’exclut des paramètres affichés. Le délai maximal est aussi un paramètre :

ticket_api/config.py

import os
from dataclasses import dataclass, field


@dataclass(frozen=True)
class Settings:
    api_key: str | None = field(repr=False)
    allowed_origins: list[str]
    classifier_mode: str
    classifier_timeout: float
    log_level: str
    show_docs: bool


def load_settings() -> Settings:
    """Read the settings from environment variables, with safe defaults."""
    origins = os.environ.get("ALLOWED_ORIGINS", "")
    return Settings(
        api_key=os.environ.get("API_KEY") or None,
        allowed_origins=[o.strip() for o in origins.split(",") if o.strip()],
        classifier_mode=os.environ.get("CLASSIFIER_MODE", "keywords"),
        classifier_timeout=float(os.environ.get("CLASSIFIER_TIMEOUT", "2.0")),
        log_level=os.environ.get("LOG_LEVEL", "INFO").upper(),
        show_docs=os.environ.get("SHOW_DOCS", "true").lower() == "true",
    )

La vérification est une dépendance, dans un nouveau module :

ticket_api/security.py

import secrets
from typing import Annotated

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

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


def require_api_key(
    request: Request, api_key: Annotated[str | None, Depends(api_key_header)]
) -> None:
    """Allow the request only with the API key from the server's settings."""
    expected = request.app.state.settings.api_key
    if expected is None:
        return
    if api_key is None or not secrets.compare_digest(api_key, expected):
        raise HTTPException(
            status_code=401, detail="A valid X-API-Key header is required."
        )

Élément	Raison
APIKeyHeader(name="X-API-Key")	Lit l’en-tête et indique à FastAPI de le documenter. Swagger UI affiche alors un bouton Authorize (autoriser).
auto_error=False	Sans l’en-tête, la valeur est None. Cette fonction choisit donc l’erreur, dans le format standard.
if expected is None: return	Sans clé dans les paramètres, il n’y a pas de vérification. C’est pratique sur votre ordinateur, mais dangereux sur un serveur.
secrets.compare_digest	La comparaison prend le même temps pour chaque clé incorrecte. Un == normal s’arrête au premier caractère différent. La différence de temps peut aider un attaquant à deviner la clé, un caractère à la fois.

Comme « sans clé » signifie « ouvert », l’application émet un avertissement au démarrage si API_KEY n’est pas définie. Dans create_app, après app.state.settings = settings :

ticket_api/main.py (in create_app)

    if settings.api_key is None:
        logger.warning(
            "API_KEY is not set: /v1/classify accepts requests without a key"
        )

La route reçoit la dépendance et deux réponses documentées supplémentaires :

ticket_api/routes.py (changes to the classify route)

from ticket_api.security import require_api_key


@router.post(
    "/classify",
    # ... summary, description and response_description as before
    responses={
        401: {
            "model": ErrorResponse,
            "description": "The API key is missing or wrong.",
        },
        413: {"model": ErrorResponse, "description": "The request body is too large."},
        # ... 422 and 503 as before
    },
    dependencies=[Depends(require_api_key)],
)

dependencies=[...] exécute la vérification avant la route, sans passer de valeur à la fonction. /v1/categories et /health restent ouverts : ils ne révèlent aucun secret. Les outils d’hébergement doivent aussi accéder à /health sans clé. Ajoutez 401: "unauthorized" à CODES dans errors.py, pour donner un code clair à l’erreur. Si des navigateurs appellent l’API, ajoutez "X-API-Key" à allow_headers dans les paramètres CORS.

La limite de taille du corps

Un billet est court. La limite est de 16 Ko, soit environ 16 000 caractères. C’est bien plus que tout billet réel, et bien moins que ce qui pourrait nuire au serveur. L’intergiciel effectue la vérification avant que FastAPI lise le corps, car lire un corps énorme est le problème :

ticket_api/main.py (the middleware, changed)

MAX_BODY_BYTES = 16_384

    # in create_app:
    @app.middleware("http")
    async def request_id_and_limits(request: Request, call_next):
        request.state.request_id = uuid.uuid4().hex[:12]
        started = time.perf_counter()
        length = request.headers.get("content-length")
        if length and int(length) > MAX_BODY_BYTES:
            response = error_response(
                request,
                413,
                "body_too_large",
                f"The body is larger than {MAX_BODY_BYTES} bytes.",
            )
        else:
            response = await call_next(request)
        response.headers["X-Request-ID"] = request.state.request_id
        # ... the log line, as before
        return response

L’en-tête Content-Length du client indique la taille avant l’arrivée du corps. L’API refuse donc sans le lire. Un client peut omettre cet en-tête et envoyer le corps en plusieurs parties. Pour ce cas, les serveurs de production et les plateformes d’hébergement ont aussi leur propre limite. Cette vérification est la première couche, pas la seule.

Le délai maximal

Le classificateur s’exécute pendant la requête. S’il bloque, la requête attend, et le client aussi. Avec assez de requêtes bloquées, l’API ne répond plus à personne. Un délai maximal transforme une « attente indéfinie » en un 503 rapide et clair.

Pour tester avec un élément lent, classifier.py reçoit un troisième classificateur simulé, qui attend 5 secondes :

ticket_api/classifier.py (changes)

import time


class SlowClassifier(KeywordClassifier):
    """Simulates a classifier that takes too long to answer."""

    version = "slow"

    def predict(self, text: str) -> Prediction:
        time.sleep(5)
        return super().predict(text)


def make_classifier(mode: str) -> KeywordClassifier | UnavailableClassifier:
    classifiers = {
        "keywords": KeywordClassifier,
        "unavailable": UnavailableClassifier,
        "slow": SlowClassifier,
    }
    if mode not in classifiers:
        raise ValueError(f"unknown CLASSIFIER_MODE {mode!r}")
    return classifiers[mode]()

La route attend le classificateur pendant au plus classifier_timeout secondes :

ticket_api/routes.py (the classify function)

import asyncio

from fastapi.concurrency import run_in_threadpool

from ticket_api.classifier import ClassifierUnavailable, KeywordClassifier


async def classify(
    ticket: TicketIn,
    request: Request,
    classifier: Annotated[KeywordClassifier, Depends(get_classifier)],
) -> Classification:
    text = f"{ticket.subject}\n{ticket.body}"
    timeout = request.app.state.settings.classifier_timeout
    try:
        prediction = await asyncio.wait_for(
            run_in_threadpool(classifier.predict, text), timeout
        )
    except TimeoutError:
        raise ClassifierUnavailable(f"no answer within {timeout} seconds") from None
    return Classification(
        category=prediction.category,
        priority=prediction.priority,
        confidence=prediction.confidence,
        model_version=classifier.version,
    )

Élément	Signification
async def	La route peut maintenant attendre quelque chose sans bloquer le serveur.
run_in_threadpool(classifier.predict, text)	Exécute le predict normal (pas async) dans un fil d’exécution distinct. Le serveur peut ainsi répondre à d’autres requêtes entre-temps.
asyncio.wait_for(..., timeout)	Attend au plus timeout secondes. Lève ensuite TimeoutError.
raise ClassifierUnavailable(...) from None	Transforme le dépassement du délai en 503, comme dans la première leçon. from None garde le journal court.

Le fil d’exécution lent continue jusqu’à sa fin. Le délai maximal protège le client et la capacité du serveur à répondre. Avec un vrai serveur de modèle, définissez aussi un délai maximal pour le client HTTP qui l’appelle. Vous l’avez fait avec httpx dans le cours de Python.

Toutes les couches ensemble

Chaque couche refuse tôt, avec son propre code d'état et code d'erreur. Seule une requête qui passe chaque couche atteint le classificateur.

Description textuelle du diagramme

Une requête à POST /v1/classify passe cinq vérifications dans l’ordre. Un corps de plus de 16 Ko reçoit 413 body_too_large. Un chemin ou une méthode inconnus reçoivent 404 not_found ou 405 method_not_allowed. Une clé d’API absente ou incorrecte reçoit 401 unauthorized. Un corps non conforme à TicketIn reçoit 422 invalid_request. Un classificateur qui dépasse le délai maximal ou qui est hors service donne 503 classifier_unavailable. Seule une requête qui passe les cinq vérifications reçoit 200 avec une Classification.

Il existe une exception à cet ordre : FastAPI lit le JSON avant de vérifier la clé. Un corps qui n’est pas du JSON valide reçoit donc 422, même sans clé.

Exemple détaillé : l’API protégée

Amira démarre l’API avec une clé. Sur son ordinateur, la clé peut être simple. Sur un serveur, elle doit être longue et aléatoire (module 7) :

macOS

API_KEY=local-dev-key fastapi dev ticket_api/main.py

Windows

$env:API_KEY = "local-dev-key"
fastapi dev ticket_api/main.py

Linux

API_KEY=local-dev-key fastapi dev ticket_api/main.py

Requête	Réponse
Classifier, sans clé	401 {"error":{"code":"unauthorized","message":"A valid X-API-Key header is required.",...}}
Classifier, X-API-Key: wrong	401, la même
Classifier, X-API-Key: local-dev-key	200 {"category":"billing","priority":1,...}
Classifier, avec la clé et un corps de 20 000 caractères	413 {"error":{"code":"body_too_large","message":"The body is larger than 16384 bytes.",...}}
GET /v1/categories et GET /health, sans clé	200

Avec CLASSIFIER_MODE=slow CLASSIFIER_TIMEOUT=1, une requête de classification reçoit 503 après environ 1 seconde, pas 5. Le journal indique pourquoi :

API terminal

WARNING ticket_api: classifier unavailable: no answer within 1.0 seconds id=0d227e283ea8
INFO ticket_api: POST /v1/classify 503 1006ms id=0d227e283ea8

Swagger UI montre aussi la clé. Un bouton Authorize apparaît, et le point de terminaison de classification a une icône de cadenas :

Le cadenas indique quel point de terminaison exige la clé. Les points de terminaison ouverts n'ont pas de cadenas. L’image est en anglais.

Sélectionnez Authorize, saisissez la clé, et Swagger UI l'envoie avec chaque requête de la page. La clé reste dans votre navigateur. L’image est en anglais.

Tests des protections

Ajoutez ces tests à tests/test_classify.py. Les fixtures de la dernière leçon rendent chacun court. Ajoutez aussi api_key=None et classifier_timeout=2.0 à TEST_SETTINGS dans conftest.py :

tests/test_classify.py (add)

def test_body_too_large(client):
    response = client.post("/v1/classify", json={"subject": "x", "body": "a" * 20_000})
    assert response.status_code == 413
    assert response.json()["error"]["code"] == "body_too_large"


def test_slow_classifier_times_out(make_client):
    client = make_client(classifier_mode="slow", classifier_timeout=0.1)
    response = client.post("/v1/classify", json={"subject": "Refund"})
    assert response.status_code == 503


def test_api_key_is_required_when_set(make_client):
    client = make_client(api_key="test-key")
    assert client.post("/v1/classify", json={"subject": "x"}).status_code == 401
    response = client.post(
        "/v1/classify", json={"subject": "x"}, headers={"X-API-Key": "test-key"}
    )
    assert response.status_code == 200

classifier_timeout=0.1 garde le test rapide : il attend 0,1 seconde, pas 5.

Exercice guidé : protéger et vérifier

1. Créez security.py. Modifiez config.py, classifier.py, routes.py, main.py et errors.py (ajoutez 401: "unauthorized" à CODES).

2. Ajoutez les tests et les deux paramètres à TEST_SETTINGS. Exécutez les tests. Le test du contrat échoue : lisez les nouvelles réponses 401 et 413 dans les différences, puis exportez le contrat. Tous les tests réussissent.

3. Démarrez l’API sans API_KEY. Trouvez l’avertissement dans le terminal.

4. Démarrez-la avec API_KEY=local-dev-key. Envoyez les requêtes du tableau de l’exemple détaillé avec curl ou Postman.

5. Dans Swagger UI, sélectionnez Authorize, saisissez local-dev-key et essayez POST /v1/classify.

6. Dans Postman, donnez à apiKey la valeur locale local-dev-key dans l’environnement Local. Envoyez Classify a support ticket (classifier un billet de soutien) : 200.

7. Dupliquez Classify a support ticket sous le nom Reject a missing key (rejeter une clé absente). Dans son onglet Authorization (autorisation), choisissez No Auth (sans authentification). Remplacez ses tests par : l’état est 401, et le code est unauthorized.

8. Exportez de nouveau la collection dans postman/. Exécutez-la depuis le terminal, avec la clé dans la commande, pas dans un fichier :

   postman collection run postman/ticket-api.postman_collection.json -e postman/local.postman_environment.json --env-var "apiKey=local-dev-key"

   Sortie (fin)

   |                requests |                 7 |                 0 |
   |              assertions |                28 |                 0 |

À vous de jouer : la frontière de la limite du corps

Testez la limite à sa frontière, comme dans le cours de Python. Un corps d’exactement 16 384 octets doit passer la vérification de taille. Un corps de 16 385 octets doit recevoir 413. Écrivez le test. Indice : json.dumps(...) donne le corps sous forme de texte, et len(text.encode()) donne sa taille en octets. Envoyez-le avec content= et l’en-tête Content-Type, pour contrôler la taille exacte.

Quel code d’état le corps de 16 384 octets reçoit-il, si ce n’est pas 413? Pourquoi?

Afficher une réponse

import json


def body_of_size(size: int) -> str:
    """A JSON ticket body of exactly `size` bytes."""
    empty = json.dumps({"subject": "x", "body": ""})
    return json.dumps({"subject": "x", "body": "a" * (size - len(empty))})


@pytest.mark.parametrize(("size", "status"), [(16_384, 422), (16_385, 413)])
def test_body_limit_boundary(client, size, status):
    body = body_of_size(size)
    assert len(body.encode()) == size
    response = client.post(
        "/v1/classify", content=body, headers={"Content-Type": "application/json"}
    )
    assert response.status_code == status

16 384 octets passent la vérification de taille. Mais le champ du corps dépasse alors sa limite de 5 000 caractères, donc la validation répond 422. C’est correct : les deux limites protègent des choses différentes. La limite de taille protège la mémoire du serveur, avant toute lecture. La limite du champ fait partie du contrat. Un test de la limite de taille doit seulement montrer que ce n’est pas 413.

Erreur courante

Tout le monde reçoit 401 après la définition de la clé

Symptôme : après le démarrage de l’API avec API_KEY, chaque requête de classification reçoit 401. Même votre propre collection Postman le reçoit, malgré la clé définie.

Cause : la clé envoyée par le client n’est pas exactement celle du serveur. Plusieurs raisons sont courantes. L’environnement Postman n’est pas actif (le sélecteur affiche No environment, aucun environnement). Ou la clé est seulement dans la valeur shared (partagée) d’un autre environnement. Ou la valeur se termine par une espace ou un saut de ligne. Ou le serveur a démarré dans un terminal sans la variable (l’avertissement au démarrage l’indique).

Solution :

1. Vérifiez le serveur : si son terminal affiche API_KEY is not set (API_KEY n’est pas définie), redémarrez-le avec la variable.
2. Dans Postman, vérifiez l’environnement actif. Placez le pointeur sur {{apiKey}} pour voir la valeur qui sera envoyée.
3. Vérifiez l’onglet Authorization de la requête : Inherit auth from parent (hériter de l’authentification du parent). Vérifiez celui de la collection : API Key (clé d’API), clé X-API-Key, dans Header (en-tête).
4. Ne corrigez jamais le problème en désactivant la vérification. Un 401 que vous comprenez montre que la protection fonctionne.

Vérifiez votre compréhension

Un collègue dit que l'API est maintenant sécurisée, car elle a une clé d'API et CORS. Que protège chacun, et qu'est-ce qui reste sans protection?

Afficher la réponse

La clé d’API décide quels programmes peuvent classifier : seulement ceux qui ont la clé. CORS décide quelles pages Web un navigateur autorise à lire les réponses. Aucun des deux n’identifie une personne. Tous les programmes qui ont la clé la partagent. Si elle fuit, n’importe qui peut l’utiliser jusqu’à son remplacement. Il n’y a pas non plus de permissions par utilisateur ni de limite au nombre de requêtes d’un appelant. Il faut pour cela l’authentification des utilisateurs et des limites de fréquence, abordées dans les cours suivants.

Vérifiez votre compréhension

Vérification du module : avec CLASSIFIER_MODE=slow, que reçoit le client, combien de temps attend-il et que ne reçoit-il PAS?

Afficher la réponse

Le client reçoit 503 avec "code": "classifier_unavailable" et un identifiant de requête. Il le reçoit après le délai maximal (2 secondes par défaut), pas après les 5 secondes que prend le classificateur. Il ne reçoit ni trace de pile, ni message d’erreur interne, ni connexion bloquée. Le serveur reste réactif pour les autres requêtes, car le classificateur s’exécute dans un fil d’exécution distinct. Le journal contient un WARNING (avertissement) avec le même identifiant de requête.

De bout en boutServir un modèle de prix des maisonsÉtude de cas

L’étude de cas sur le prix des maisons ajoute une vérification de somme de contrôle du modèle aux protections de cette leçon : la clé d’API et la limite du corps.

Récapitulatif

• Une clé d’API dans un en-tête (X-API-Key) limite qui peut appeler un point de terminaison. Comparez-la avec secrets.compare_digest. Émettez un avertissement si aucune clé n’est définie.
• APIKeyHeader documente la clé. Swagger UI affiche donc Authorize, et Postman l’envoie depuis l’onglet Authorization de la collection.
• Une limite de taille du corps dans l’intergiciel refuse les corps énormes avec 413 avant leur lecture.
• Un délai maximal autour du classificateur transforme une « attente indéfinie » en un 503 rapide.
• Placez les clés dans les variables d’environnement et les valeurs locales de Postman. Passez-les à l’interface en ligne de commande avec --env-var, jamais dans des fichiers.

L’API est maintenant testée, protégée et dotée de journaux. Dans le dernier module, vous la déployez.

Leçon suivante : Premiers pas avec l'hébergement