Comment versionner une API et sécuriser ses endpoints avec OAuth2 dans FastAPI

Le versionnement d'une API est une pratique essentielle pour maintenir et faire évoluer les services web sans perturber les utilisateurs existants. Il permet aux développeurs d'introduire des modifications, des améliorations, voire des changements incompatibles, tout en garantissant la compatibilité avec les versions précédentes. Dans cette section, nous allons explorer le processus de versionnement d'une API avec FastAPI, en prenant l'exemple d'un gestionnaire de tâches, et comment sécuriser les endpoints de l'API à l'aide du protocole OAuth2.

Implémentation du versionnement d'API

Prenons un exemple où nous voulons ajouter un nouveau champ priority à nos tâches. Par défaut, la valeur de ce champ sera "lower", mais elle pourra être modifiée pour répondre à des besoins spécifiques. Pour cela, nous allons suivre les étapes suivantes :

1. Création de la classe TaskV2 dans le module models.py :

python
Copy code
from typing import Optional
class TaskV2(BaseModel):
    title: str
    description: str
    status: str
    priority: str | None = "lower"
class TaskV2WithID(TaskV2):
    id: int

2. Création de la fonction read_all_tasks_v2 dans le module operations.py pour lire toutes les tâches, en incluant le champ priority :

python
Copy code
from models import TaskV2WithID

def read_all_tasks_v2() -> list[TaskV2WithID]:
    with open(DATABASE_FILENAME) as csvfile:
        reader = csv.DictReader(csvfile)
        return [TaskV2WithID(**row) for row in reader]

3. Dans le module main.py, nous définissons le nouvel endpoint pour la version 2 de notre API :

python
Copy code
from models import TaskV2WithID
@app.get("/v2/tasks", response_model=list[TaskV2WithID])
def get_tasks_v2():
    tasks = read_all_tasks_v2()
    return tasks

Nous avons ainsi créé une nouvelle version de l'endpoint, qui inclut le champ priority. Cette approche permet de faire évoluer l'API en introduisant de nouvelles versions sans perturber les utilisateurs de la version précédente.

Tests et validation

Pour tester la nouvelle version de l'API, il est nécessaire de modifier le fichier tasks.csv afin d'y ajouter manuellement la nouvelle colonne priority. Par exemple :

arduino
Copy code
id,title,description,status,priority

1,Task One,Description One,Incomplete
2,Task Two,Description Two,Ongoing,higher

Lancez ensuite le serveur avec la commande suivante :

bash
Copy code
$ uvicorn main:app

Autres stratégies de versionnement

Le versionnement par URL que nous avons utilisé est la méthode la plus courante, mais il existe d'autres stratégies de versionnement, notamment :

• Versionnement par paramètre de requête : l'information de version est passée en paramètre dans l'URL, comme dans l'exemple suivant : https://api.example.com/resource?version=1. Cette méthode permet de garder l'URL de base uniforme à travers toutes les versions.

• Versionnement par en-tête : la version est spécifiée dans un en-tête personnalisé de la requête HTTP, tel que :
  GET /resource HTTP/1.1
Host: api.example.com
X-API-Version: 1

• Versionnement basé sur le consommateur : cette stratégie permet aux utilisateurs de choisir la version qu'ils souhaitent utiliser, et cette version est enregistrée lors de leur première interaction. À moins qu'ils ne la modifient, cette version sera utilisée lors de toutes leurs interactions futures.

Il peut également être pertinent d'utiliser un versionnement sémantique, où les numéros de version suivent un format du type MAJOR.MINOR.PATCH. Les changements majeurs (MAJOR) indiquent des modifications incompatibles, tandis que les changements mineurs (MINOR) et les correctifs (PATCH) sont rétrocompatibles.

Sécurisation de l'API avec OAuth2

1. Création d'une base de données fictive d'utilisateurs : dans un premier temps, créons un dictionnaire contenant des utilisateurs et leurs mots de passe hachés.

python
Copy code
fake_users_db = {

    "johndoe": {"username": "johndoe", "hashed_password": "hashedsecret"},

    "janedoe": {"username": "janedoe", "hashed_password": "hashedsecret2"},
}

2. Mécanisme de hachage : les mots de passe ne doivent pas être stockés en clair. Pour l'exemple, nous utilisons une fonction de hachage simplifiée, mais dans un environnement de production, un véritable mécanisme de hachage tel que bcrypt devrait être utilisé.

python
Copy code
def fakely_hash_password(password: str):
    return f"hashed{password}"

3. Gestion des utilisateurs : nous définissons une classe pour représenter un utilisateur et une fonction pour récupérer un utilisateur à partir de la base de données.

python
Copy code
class User(BaseModel):

    username: str
class UserInDB(User):
    hashed_password: str
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

4. Création de fonctions pour générer et résoudre des tokens : enfin, nous définissons des fonctions pour générer et résoudre des tokens d'authentification.

python
Copy code
def fake_token_generator(user: UserInDB) -> str:
    return f"tokenized{user.username}"

def fake_token_resolver(token: str) -> UserInDB | None:

    if token.startswith("tokenized"):
        user_id = token.removeprefix("tokenized")
        user = get_user(fake_users_db, user_id)
        return user

5. Sécurisation des endpoints avec OAuth2 : nous utilisons maintenant le mécanisme d'OAuth2 pour protéger nos endpoints.

python
Copy code
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

Ainsi, en utilisant OAuth2, nous avons sécurisé l'accès aux endpoints de l'API, permettant aux utilisateurs autorisés d'accéder uniquement aux données qu'ils sont censés voir.

Comment implémenter l'authentification sécurisée avec OAuth2 et JWT dans une application SaaS

L'authentification et l'autorisation sont des aspects cruciaux de toute application, en particulier dans un environnement SaaS où la sécurité des données des utilisateurs doit être primordiale. Utiliser OAuth2 avec des tokens JWT (JSON Web Tokens) est une solution moderne et robuste pour garantir que seules les personnes autorisées accèdent à des ressources protégées. Voici comment implémenter cette méthode dans votre application.

Lorsque vous créez un utilisateur dans un système, il est essentiel de vérifier si le nom d'utilisateur ou l'adresse e-mail existe déjà. Dans le cas où l’un de ces éléments est déjà présent dans la base de données, une réponse 409 est renvoyée pour indiquer qu'il y a un conflit et que la création de l'utilisateur est interdite. Afin de tester ce mécanisme, il suffit de démarrer le serveur à la racine du projet en exécutant la commande suivante :

bash
Copy code
$ uvicorn main:app

Ensuite, vous pouvez vous rendre à l'adresse localhost:8000/docs pour vérifier les endpoints nouvellement créés dans la documentation Swagger et effectuer des tests. Un exemple de test consiste à vérifier la fonction add_user et l'endpoint /register/user. Voici un exemple de test à ajouter pour ces fonctionnalités :

python
Copy code
def test_add_user_into_the_database(session): 

    user = add_user(...)  # compléter le test
def test_endpoint_add_basic_user(client): 
    response = client.post( 
        "/register/user", 
        json= ...  # continuer le test
    )

Maintenant, voyons comment sécuriser cette application en utilisant OAuth2 et JWT.

Mise en place de l'authentification avec OAuth2 et JWT

1. Préparation de l'environnement
   Avant de commencer, assurez-vous d’avoir installé les dépendances nécessaires à l'implémentation de l'authentification avec JWT. Si vous n'avez pas encore installé les packages du fichier requirements.txt, exécutez cette commande :

bash
Copy code
$ pip install python-jose[cryptography]

Ensuite, pour intégrer OAuth2 avec JWT, vous allez définir des fonctions d’authentification, de création et de décodage des tokens. Ces fonctions seront regroupées dans un module appelé security.py.

2. Définir la fonction d'authentification
   

   La première étape consiste à créer une fonction permettant de valider les utilisateurs, que ce soit par leur nom d'utilisateur ou par leur adresse e-mail. Cette fonction va interroger la base de données pour vérifier si un utilisateur existe et si le mot de passe est correct :

python
Copy code
from sqlalchemy.orm import Session
from models import User
from email_validator import validate_email, EmailNotValidError
from operations import pwd_context
def authenticate_user(session: Session, username_or_email: str, password: str) -> User | None:
    try:
        validate_email(username_or_email)
        query_filter = User.email
    except EmailNotValidError:
        query_filter = User.username
    user = session.query(User).filter(query_filter == username_or_email).first()

    if not user or not pwd_context.verify(password, user.hashed_password):

        return None
    return user

3. Création et décodage du token d'accès
   Ensuite, il faut créer un token d'accès et définir la clé secrète, l'algorithme utilisé pour générer le token et sa durée de validité. Voici comment cela se fait :

python
Copy code
from jose import jwt
from datetime import datetime, timedelta
SECRET_KEY = "a_very_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict) -> str:

    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

De même, pour décoder le token et en extraire les informations, on utilise la fonction suivante :

python
Copy code
from jose import JWTError

def decode_access_token(token: str, session: Session) -> User | None:

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
    except JWTError:
        return None
    if not username:
        return None
    user = get_user(session, username)
    return user

4. Création des endpoints d'authentification
   Une fois que nous avons les fonctions nécessaires, il est temps de les intégrer dans l'API. Nous utilisons l'APIRouter de FastAPI pour définir les endpoints qui gèrent la création du token d'accès et la récupération des informations de l'utilisateur. Le code suivant définit un endpoint qui retourne un token lorsque les informations d'identification sont valides :

python
Copy code
from fastapi import APIRouter, Depends, HTTPException, status

from fastapi.security import OAuth2PasswordRequestForm
router = APIRouter()
class Token(BaseModel):
    access_token: str
    token_type: str
@router.post("/token", response_model=Token)
def get_user_access_token(form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)):
    user = authenticate_user(session, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password")
    access_token = create_access_token(data={"sub": user.username})

    return {"access_token": access_token, "token_type": "bearer"}

5. Protéger les endpoints avec OAuth2
   Une fois le token généré, vous pouvez créer des endpoints protégés qui nécessitent l’envoi du token dans l'en-tête de la requête. Par exemple, un endpoint qui renvoie les informations de l'utilisateur connecté :

python
Copy code
from fastapi import Depends
@router.get("/users/me")
def read_user_me(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
    user = decode_access_token(token, session)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not authorized")
    return {"description": f"{user.username} authorized"}

6. Intégration dans l'application principale
   Enfin, vous devez inclure ce module dans le serveur FastAPI en important le routeur dans le fichier principal main.py :

python
Copy code
import security

app.include_router(security.router)

Conclusion

Une fois ces étapes terminées, votre application SaaS sera protégée par OAuth2 avec des tokens JWT. Ces tokens offrent une méthode sûre et fiable pour authentifier les utilisateurs, en remplaçant les systèmes traditionnels basés sur des sessions et des mots de passe. Les utilisateurs devront s'authentifier via l'endpoint /token et ensuite utiliser le token pour accéder aux ressources protégées.

Comment intégrer FastAPI avec gRPC pour créer une passerelle REST-gRPC

L’intégration de FastAPI avec gRPC permet de construire une passerelle performante et simple d’utilisation entre des clients HTTP et des services gRPC. Ce processus repose sur l’utilisation de fichiers .proto pour définir la structure des messages échangés entre le client et le serveur gRPC, ainsi que sur la mise en place d’un serveur gRPC basique avant de l’étendre pour inclure une interface HTTP avec FastAPI.

La première étape pour cette intégration consiste à définir le service gRPC dans un fichier .proto en utilisant la version proto3. Ce fichier contiendra les définitions des messages ainsi que des services gRPC, qui pourront ensuite être utilisées pour générer le code Python nécessaire. Par exemple, un fichier grpcserver.proto pourrait être structuré de la manière suivante :

proto
Copy code
syntax = "proto3";

service GrpcServer {
  rpc GetServerResponse(Message) returns (MessageResponse) {}
}
message Message {
  string message = 1;
}
message MessageResponse {
  string message = 1;
  bool received = 2;
}

Ce fichier définit un service GrpcServer avec une méthode RPC nommée GetServerResponse. Cette méthode prend un message de type Message en entrée et renvoie une réponse de type MessageResponse. La structure de ces messages est ensuite générée automatiquement à l’aide de l’outil grpc_tools.protoc, qui produit deux fichiers Python : grpcserver_pb2.py et grpcserver_pb2_grpc.py. Le premier contient les classes pour définir les messages, tandis que le second inclut les classes pour le service et la méthode RPC.

Une fois cette étape terminée, un serveur gRPC basique peut être mis en place. Cela nécessite la définition d’une classe serveur héritée de GrpcServerServicer, ainsi qu’une méthode GetServerResponse pour traiter les messages reçus et renvoyer une réponse adéquate :

python
Copy code
from grpcserver_pb2 import MessageResponse

from grpcserver_pb2_grpc import GrpcServerServicer
class Service(GrpcServerServicer):
    async def GetServerResponse(self, request, context):
        message = request.message
        logging.info(f"Received message: {message}")
        result = {
            "message": f"Hello I am up and running, received: {message}",
            "received": True
        }
        return MessageResponse(**result)

Ce serveur écoute les requêtes et renvoie une réponse structurée selon les spécifications définies dans le fichier .proto.

Ensuite, pour lancer le serveur sur le port 50051, le script Python suivant peut être utilisé :

python
Copy code
import grpc

from grpcserver_pb2_grpc import add_GrpcServerServicer_to_server
async def serve():
    server = grpc.aio.server()
    add_GrpcServerServicer_to_server(Service(), server)
    server.add_insecure_port("[::]:50051")
    logging.info("Starting server on port 50051")
    await server.start()
    await server.wait_for_termination()

Une fois ce serveur gRPC en place, il est possible de créer une passerelle REST en utilisant FastAPI. Pour cela, nous définirons un fichier main.py où nous utiliserons FastAPI pour interagir avec le serveur gRPC. Le principal avantage d’utiliser FastAPI réside dans la simplicité de la création d’un serveur HTTP tout en tirant parti de la puissance de gRPC en arrière-plan.

Tout d’abord, il faut définir un schéma de réponse avec Pydantic pour assurer la validation des données envoyées depuis FastAPI vers le client :

python
Copy code
from pydantic import BaseModel

class GRPCResponse(BaseModel):
    message: str
    received: bool

Ensuite, un canal gRPC sécurisé sera établi pour communiquer avec le serveur gRPC en utilisant grpc.aio.insecure_channel. Une fois le canal établi, un point de terminaison HTTP peut être configuré pour interagir avec gRPC :

python
Copy code
from fastapi import FastAPI
import grpc
from grpcserver_pb2_grpc import GrpcServerStub
from grpcserver_pb2 import Message
app = FastAPI()
grpc_channel = grpc.aio.insecure_channel("localhost:50051")
@app.get("/grpc")

async def call_grpc(message: str) -> GRPCResponse:

    async with grpc_channel as channel:
        grpc_stub = GrpcServerStub(channel)
        response = await grpc_stub.GetServerResponse(Message(message=message))
        return response

Ici, l’endpoint HTTP /grpc prend un paramètre message et envoie ce message au serveur gRPC via l’API. La réponse du serveur gRPC est ensuite renvoyée sous forme de réponse JSON au client HTTP.

Une fois le code en place, le serveur FastAPI peut être lancé avec la commande suivante :

bash
Copy code
$ uvicorn app.main:app

Cela permettra d’accéder à la documentation interactive de l’API via l’interface Swagger à l’adresse http://localhost:8000/docs, où vous pourrez tester l’endpoint /grpc.

L’intégration de gRPC avec FastAPI permet ainsi de bénéficier des avantages des deux technologies : FastAPI pour sa simplicité d’utilisation et son excellent support pour les APIs REST, et gRPC pour sa rapidité et son efficacité pour les appels entre services.

Il est également essentiel de noter que, bien que nous ayons implémenté ici un RPC simple de type "unary", d’autres types de communication existent en gRPC, tels que les flux de messages (streaming) dans les deux sens, permettant une communication bidirectionnelle entre client et serveur. Pour étendre cette intégration, il est possible d’explorer l’implémentation de ces différents types d’appels, ce qui peut être utile pour des cas d’utilisation nécessitant des échanges continus de données.