Comment implémenter RBAC (Contrôle d'accès basé sur les rôles) dans une application FastAPI

Dans le cadre de la gestion des accès et des rôles dans une application SaaS, l'implémentation du modèle RBAC (Role-Based Access Control) constitue une approche fondamentale pour organiser les permissions des utilisateurs de manière sécurisée et flexible. Nous allons voir ici comment définir et gérer les rôles dans une application FastAPI, tout en appliquant des contrôles d'accès précis.

Pour commencer, il est essentiel de structurer la base de données afin d'intégrer la notion de rôles dans le modèle utilisateur. À cette fin, nous allons définir une classe Role utilisant un Enum pour représenter les différents niveaux d’accès. Ces rôles peuvent, par exemple, être « basic » pour un utilisateur standard et « premium » pour un utilisateur avec des privilèges plus élevés.

Dans notre application, nous allons d'abord étendre le modèle de l'utilisateur pour inclure un champ role, comme suit :

python
Copy code
from enum import Enum

from sqlalchemy.orm import mapped_column, Mapped
from sqlalchemy import Column, String
class Role(str, Enum):
    basic = "basic"
    premium = "premium"
class User(Base):
    __tablename__ = "users"
    # autres champs existants
    role: Mapped[Role] = mapped_column(default=Role.basic)

Ici, la colonne role dans la table users est définie avec une valeur par défaut « basic », indiquant que, par défaut, les utilisateurs seront considérés comme des utilisateurs basiques, à moins qu'un autre rôle ne soit spécifié.

La fonction add_user dans le module operations.py permet maintenant d'ajouter des utilisateurs en précisant leur rôle. Par défaut, le rôle est celui d'un utilisateur de type « basic » :

python
Copy code
from models import Role
def add_user(
    session: Session,
    username: str,
    password: str,
    email: str,
    role: Role = Role.basic,
) -> User | None:
    hashed_password = pwd_context.hash(password)
    db_user = User(
        username=username,
        email=email,
        hashed_password=hashed_password,
        role=role,
    )
    # Reste de la fonction

Nous pouvons maintenant définir des points d'accès spécifiques pour des utilisateurs avec des rôles différents. Par exemple, pour les utilisateurs « premium », nous allons créer un module distinct qui gère l'enregistrement des utilisateurs premium :

python
Copy code
@router.post(

    "/register/premium-user",
    status_code=status.HTTP_201_CREATED,
    response_model=ResponseCreateUser,
)
def register_premium_user(
    user: UserCreateBody,
    session: Session = Depends(get_session),
):
    user = add_user(session=session, *user.model_dump(), role=Role.premium)
    if not user:
        raise HTTPException(
            status.HTTP_409_CONFLICT, "username or email already exists"
        )
    user_response = UserCreate(username=user.username, email=user.email)
    return {"message": "user created", "user": user_response}

La création des endpoints repose également sur la mise en place de fonctions d'assistance, comme get_current_user et get_premium_user, qui servent de dépendances pour vérifier que l'utilisateur actuel a le rôle approprié avant d’accéder à certains points d'accès. Ces fonctions sont définies dans un module séparé appelé rbac.py :

python
Copy code
from typing import Annotated

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: Session = Depends(get_session),
) -> UserCreateRequestWithRole:
    user = decode_access_token(token, session)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not authorized",
        )
    return UserCreateRequestWithRole(
        username=user.username, email=user.email, role=user.role
    )

Pour s’assurer qu’un utilisateur ait accès à un point d’accès réservé aux utilisateurs premium, nous utilisons une vérification supplémentaire dans la fonction get_premium_user :

python
Copy code
def get_premium_user(current_user: Annotated[get_current_user, Depends()]):
    if current_user.role != Role.premium:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not authorized",
        )
    return current_user

Ces vérifications permettent d'imposer un contrôle d'accès basé sur les rôles, garantissant que les utilisateurs ne puissent accéder qu'aux ressources auxquelles leur rôle leur donne droit.

Ensuite, la création de points d'accès devient triviale. Par exemple, un point d'accès pour les utilisateurs de tous les rôles serait défini comme suit :

python
Copy code
@router.get("/welcome/all-users")

def all_users_can_access(user: Annotated[get_current_user, Depends()]):
    return {f"Hello {user.username}, welcome to your space"}

Tandis qu'un point d'accès réservé uniquement aux utilisateurs premium ressemblerait à ceci :

python
Copy code
@router.get("/welcome/premium-user")

def only_premium_users_can_access(
    user: UserCreateResponseWithRole = Depends(get_premium_user),
):
    return {f"Hello {user.username}, Welcome to your premium space"}

Cette approche assure une gestion fine des permissions basée sur les rôles des utilisateurs, et elle est facilement extensible à d’autres rôles et ressources.

En complément de cette approche, une autre méthode pour appliquer RBAC est l'attribution de « scopes » à un jeton. Un « scope » représente une chaîne de caractères qui détermine les permissions d'un utilisateur, et il peut être défini directement dans le système de génération de jetons. FastAPI offre des outils puissants pour cela, permettant de définir des scopes à l'intérieur des jetons OAuth2. Cette méthode offre une flexibilité accrue et une gestion des permissions encore plus granulaire. Pour plus d'informations, il est recommandé de consulter la documentation officielle de FastAPI sur les scopes OAuth2.

Comment mettre en œuvre les opérations CRUD avec FastAPI et SQLAlchemy : gestion des migrations et des relations en base de données

Les opérations CRUD (Créer, Lire, Mettre à jour, Supprimer) sont essentielles dans toute application de gestion de données. Dans ce contexte, nous allons explorer la mise en œuvre de ces opérations avec FastAPI et SQLAlchemy en utilisant des bases de données SQL, et plus spécifiquement comment gérer les migrations de schéma et les relations entre les tables.

Pour commencer, nous avons défini une opération de suppression pour gérer la suppression d'un ticket de la base de données :

python
Copy code
async def delete_ticket(db_session: AsyncSession, ticket_id) -> bool:

    async with db_session as session:
        tickets_removed = await session.execute(
            delete(Ticket).where(Ticket.id == ticket_id)
        )
        await session.commit()
        if tickets_removed.rowcount == 0:
            return False
        return True

Cette fonction utilise SQLAlchemy pour interagir avec la base de données. Si aucun ticket ne correspond à l'ID donné, la fonction retourne False, sinon elle retourne True. Il s'agit d'un comportement similaire à l'opération de mise à jour, où une valeur booléenne est retournée pour indiquer le succès de l'opération.

Après avoir défini les opérations, il convient de les exposer via des points de terminaison (endpoints) dans l'application FastAPI. Pour l'opération de création de ticket, on peut définir un endpoint comme suit :

python
Copy code
@app.post("/ticket", response_model=dict[str, int])

async def create_ticket_route(ticket: TicketRequest, db_session: Annotated[AsyncSession, Depends(get_db_session)]):
    ticket_id = await create_ticket(db_session, ticket.show, ticket.user, ticket.price)
    return {"ticket_id": ticket_id}

Une fois les endpoints définis, il est crucial de tester leur bon fonctionnement. Pour ce faire, on peut utiliser l'interface de documentation interactive de FastAPI, accessible à l'adresse http://localhost:8000/docs. En accédant à cette page, l'utilisateur peut tester les endpoints et vérifier que l'interaction avec la base de données est correctement effectuée, avec des résultats visibles dans le fichier .database.db.

Ensuite, une étape clé dans la gestion des bases de données est la gestion des migrations. Les migrations permettent de versionner le schéma de la base de données et de garder une trace de l'évolution du schéma à travers différentes versions. Cela est particulièrement important lorsque des modifications sont apportées à la structure de la base de données, sans risque de perte de données.

Pour cela, nous utilisons Alembic, un outil bien connu pour la gestion des migrations dans les projets Python utilisant SQLAlchemy. Voici les étapes à suivre pour configurer Alembic dans votre projet :

1. Initialiser Alembic : Depuis le répertoire racine du projet, la commande alembic init alembic crée les fichiers nécessaires à la gestion des migrations, dont le fichier alembic.ini et un dossier alembic/ contenant des fichiers de configuration.

2. Configurer la connexion à la base de données : Dans le fichier alembic.ini, il faut configurer l'URL de la base de données en spécifiant, par exemple, pour une base SQLite :

   ini
   Copy code
   sqlalchemy.url = sqlite:///.database.db
   

3. Mettre à jour le fichier de migration : Après avoir configuré Alembic, il est nécessaire d'ajouter la configuration de votre modèle SQLAlchemy. Cela se fait en définissant la variable target_metadata dans le fichier env.py pour qu'elle pointe vers la métadonnée de votre application.

4. Créer une première migration : La commande alembic revision --autogenerate -m "Start database" génère automatiquement un script de migration. Cette migration peut ensuite être appliquée avec alembic upgrade head, créant ainsi la table des tickets dans la base de données.

5. Effectuer des mises à jour du schéma : Pour ajouter de nouveaux champs ou tables à la base de données, Alembic permet de gérer les modifications sans perte de données. Par exemple, pour ajouter un champ sold à la table des tickets :

   python
   Copy code
   class Ticket(Base):
   
       __tablename__ = "tickets"
       id: Mapped[int] = mapped_column(primary_key=True)
       price: Mapped[float] = mapped_column(nullable=True)
       show: Mapped[str | None]
       user: Mapped[str | None]
       sold: Mapped[bool] = mapped_column(default=False)
   

   Après avoir modifié le modèle, vous pouvez créer une nouvelle migration avec la commande alembic revision --autogenerate -m "Add sold field", et appliquer la migration avec alembic upgrade head. Cela mettra à jour le schéma de la base de données tout en préservant les données existantes.

Enfin, un aspect souvent négligé mais crucial dans les bases de données relationnelles est la gestion des relations entre les tables. SQLAlchemy offre une flexibilité importante pour définir des relations de type un-à-un, un-à-plusieurs et plusieurs-à-plusieurs, permettant de créer des structures de données complexes.

Pour l'exemple de notre système de gestion de tickets, nous pouvons envisager une relation un-à-un entre un ticket et une table supplémentaire contenant des détails supplémentaires sur le ticket, comme ceci :

python
Copy code
class TicketDetail(Base):

    __tablename__ = "ticket_details"
    ticket_id: Mapped[int] = mapped_column(ForeignKey("tickets.id"), primary_key=True)
    detail: Mapped[str] = mapped_column(nullable=False)

Cela permet de lier un ticket à un détail spécifique tout en respectant les contraintes d'intégrité de la base de données. SQLAlchemy facilite la création de telles relations grâce à des mécanismes comme les clés étrangères (ForeignKey) et la gestion des objets associés.

Le travail avec des migrations et des relations en base de données est une compétence essentielle pour construire des applications robustes et évolutives. En maîtrisant ces concepts avec FastAPI et SQLAlchemy, vous serez en mesure de gérer de manière efficace et sécurisée l'interaction avec vos données, tout en assurant la continuité de votre application à travers des mises à jour progressives et sans perte d'information.

Comment sécuriser une connexion WebSocket avec OAuth2 dans FastAPI ?

Pour commencer, nous devons définir l’objet oauth2_scheme_for_ws qui sert à valider les connexions WebSocket en vérifiant le jeton d'authentification. Cela se fait par la classe OAuth2WebSocketPasswordBearer, que nous importons depuis le fichier app.ws_password_bearer :

python
Copy code
from app.ws_password_bearer import OAuth2WebSocketPasswordBearer

oauth2_scheme_for_ws = OAuth2WebSocketPasswordBearer(
    tokenUrl="/token"
)

L'argument tokenUrl spécifie l'URL de l'endpoint pour récupérer le jeton. Cet endpoint doit être mis en place en fonction de la méthode de résolution du jeton que vous utilisez dans votre application. Ensuite, nous pouvons créer une fonction qui extrait le nom d'utilisateur du jeton reçu.

python
Copy code
def get_username_from_token(

    token: str = Depends(oauth2_scheme_for_ws),
) -> str:
    user = fake_token_resolver(token)
    if not user:
        raise WebSocketException(
            code=status.HTTP_401_UNAUTHORIZED,
            reason="Invalid authentication credentials"
        )
    return user.username

La fonction fake_token_resolver est utilisée pour simuler la résolution du jeton. Elle est définie dans le fichier security.py et utilise deux utilisateurs fictifs (johndoe et janedoe) pour des fins de test. Ce système n'apporte aucune sécurité réelle, il est utilisé ici à titre d'exemple. Dans un environnement de production, il est essentiel d'utiliser un jeton JWT (JSON Web Token) ou un fournisseur externe pour la résolution des jetons, comme détaillé dans le chapitre 4 sur l’authentification et l’autorisation.

L’étape suivante consiste à sécuriser notre endpoint WebSocket /secured-ws dans le fichier main.py :

python
Copy code
from fastapi import Depends
from app.security import get_username_from_token
@app.websocket("/secured-ws")
async def secured_websocket(
    websocket: WebSocket,
    username: Annotated[get_username_from_token, Depends()],
):
    # reste du code pour l'endpoint

Pour autoriser la connexion, il est nécessaire de récupérer un jeton via l'endpoint /token, puis de le transmettre dans l’en-tête de la requête WebSocket. Par exemple, si l’on utilise le générateur de jetons fictifs, pour johndoe, le jeton sera tokenizedjohndoe. Ce jeton sera ensuite inclus dans l’en-tête Authorization de la requête WebSocket, comme suit :

makefile
Copy code
Authorization: Bearer tokenizedjohndoe

Une fois cette configuration effectuée, la connexion sera établie et vous pourrez échanger des messages via le WebSocket sécurisé. Ainsi, vous aurez sécurisé votre endpoint WebSocket avec OAuth2 et renforcé la sécurité de votre application FastAPI, protégeant ainsi la communication WebSocket contre les menaces potentielles.

Conseils pour l’implémentation d’un chat sécurisé :
Si vous souhaitez développer une fonctionnalité de chat sécurisé, il est important que l’endpoint qui renvoie la page HTML vérifie d'abord la présence du jeton dans les cookies du navigateur. Si le jeton est absent ou invalide, une redirection vers une page de connexion doit avoir lieu pour fournir un jeton valide et l'ajouter dans les cookies du navigateur. Cela peut être réalisé en utilisant la classe RedirectResponse de FastAPI, qui permet de gérer les redirections de manière simple et efficace.

Intégration avec OAuth2 et WebSocket :

Comment créer et gérer un projet Python avec Hatch : Un guide pour les applications FastAPI

L'utilisation de Hatch permet de simplifier et d'automatiser la gestion de vos projets Python, notamment lorsque vous développez des applications basées sur FastAPI. Ce processus commence par la création d'un package Python que l'on peut ensuite importer dans un autre projet. Prenons l'exemple de la création d'un serveur FastAPI en utilisant un package externe.

Tout d'abord, pour créer le serveur de base avec FastAPI, vous allez définir une route simple comme ceci :

python
Copy code
app = APIRouter()

@app.get("/")
def read_root():
    return {
        "message": "Welcome to the FastAPI Cookbook Application!"
    }

Ensuite, pour rendre ce code réutilisable, vous devez importer ce routeur dans le fichier src/fca_server/__init__.py de votre package, en utilisant la ligne suivante :

python
Copy code
from fca_server.main import router

bash
Copy code
$ hatch build -t sdist ../dist

Cela générera un fichier .tar.gz (par exemple fca_server-0.0.1.tar.gz) dans un dossier dist situé à l'extérieur du projet principal. Ce fichier pourra ensuite être utilisé dans un autre projet.

Tester et installer le package

bash
Copy code
$ python -m venv .venv

Activez l'environnement virtuel. Sur Linux ou macOS, utilisez la commande suivante :

bash
Copy code
$ source .venv/Scripts/activate

Sur Windows, utilisez cette commande à la place :

bash
Copy code
$ .venv\Scripts\activate

Ensuite, installez votre package fca_server avec pip :

bash
Copy code
$ pip install ..\dist\fca_server-0.0.1.tar.gz

Assurez-vous de spécifier le chemin correct vers votre fichier .tar.gz. Vous pouvez maintenant créer un fichier main.py dans votre nouveau projet et y importer le routeur de votre package fca_server :

python
Copy code
from fastapi import FastAPI
from fca_server import router
app = FastAPI(
    title="Import FCA Server Application"
)
app.include_router(router)

Enfin, lancez le serveur FastAPI avec la commande suivante :

bash
Copy code
$ fastapi run

Visitez l'interface de documentation interactive à l'adresse http://localhost:8000/docs pour voir les points de terminaison fournis par le package externe.

Avantages de Hatch et gestion avancée des projets

Hatch ne se limite pas à la création de packages et à leur installation. Il permet également une gestion avancée des environnements et des configurations, ce qui est essentiel pour une application évolutive. Par exemple, vous pouvez personnaliser l'emplacement des fichiers de l'environnement virtuel, en choisissant s'ils doivent être centralisés ou placés dans le dossier du projet. Cette option peut être configurée dans le fichier config.toml, qui contient les paramètres de configuration pour Hatch. Pour trouver ce fichier, vous pouvez exécuter la commande suivante dans votre terminal :

bash
Copy code
$ hatch config find

Une autre caractéristique intéressante de Hatch est la possibilité de créer des builds de votre package au format .whl, un format binaire plus efficace et compatible que le format source traditionnel. Vous pouvez également publier votre package directement sur le Python Package Index (PyPI), facilitant ainsi le partage de votre code avec la communauté.

Gestion des dépendances et des environnements

L'un des avantages majeurs de Hatch réside dans sa gestion des dépendances et des environnements. En effet, il vous permet de configurer plusieurs environnements virtuels pour un même projet, selon vos besoins. La gestion des dépendances est également facilitée grâce au fichier pyproject.toml, qui centralise les informations sur les dépendances, la configuration du système de build, et les autres métadonnées du projet. Pour en savoir plus sur ce fichier et sur sa rédaction, vous pouvez consulter la documentation officielle de Python.

Conclusion

La création et la gestion de projets Python avec Hatch ne sont pas seulement des étapes techniques. Elles permettent de renforcer la modularité et la réutilisabilité du code, tout en facilitant son déploiement et sa maintenance. La capacité d'utiliser des environnements virtuels isolés et de gérer les dépendances avec flexibilité est un atout majeur pour les développeurs souhaitant travailler sur des projets de grande envergure. En outre, l'intégration de Hatch avec FastAPI permet de tirer parti de ces fonctionnalités avancées pour développer des applications robustes et performantes.