From d9d31456d76a698b18febacb710b5ec9afe86157 Mon Sep 17 00:00:00 2001 From: Johan Date: Thu, 18 Dec 2025 14:54:37 +0100 Subject: [PATCH] First commit --- src/app/graphql/context.py | 13 +++--- src/app/graphql/resolvers/note_resolver.py | 48 ++++++++++++++-------- src/app/graphql/resolvers/user_resolver.py | 45 +++++++++++++++----- src/app/main.py | 26 ++++++++++-- 4 files changed, 97 insertions(+), 35 deletions(-) diff --git a/src/app/graphql/context.py b/src/app/graphql/context.py index aebda1f..fb740a3 100644 --- a/src/app/graphql/context.py +++ b/src/app/graphql/context.py @@ -13,12 +13,15 @@ async def get_context(request: Request) -> Dict[str, Any]: et injectons 'current_user' dans le contexte. """ current_user: Optional[UserModel] = None + auth_header = request.headers.get("Authorization") - # TODO 1.1: Implémenter l'authentification (AuthN) - # 1. Lire l'en-tête `Authorization` (request.headers.get(...)) - # 2. Simuler la validation du token (admin, alice, bob) - # 3. Récupérer l'utilisateur depuis `fake_db_repo` - # 4. Injecter l'utilisateur sous la clé "current_user" + # Simulation de validation de token + if auth_header == "Bearer admin-token": + current_user = await fake_db_repo.get_user_by_id(3) # Admin + elif auth_header == "Bearer alice-token": + current_user = await fake_db_repo.get_user_by_id(1) # Alice + elif auth_header == "Bearer bob-token": + current_user = await fake_db_repo.get_user_by_id(2) # Bob return { "request": request, diff --git a/src/app/graphql/resolvers/note_resolver.py b/src/app/graphql/resolvers/note_resolver.py index 2d66081..534ad72 100644 --- a/src/app/graphql/resolvers/note_resolver.py +++ b/src/app/graphql/resolvers/note_resolver.py @@ -10,16 +10,20 @@ from app.models.notemodel import NoteModel as NoteModel async def get_note_by_id(info: Info, id: strawberry.ID) -> NoteModel: """ Resolver pour 'note(id: ID!)' """ - # TODO 2.2: Récupérer le current_user (get_current_user_from_info) - + current_user = get_current_user_from_info(info) note = await fake_db_repo.get_note_by_id(int(id)) if not note: raise NotFoundBLLException(resource_name="Note", resource_id=id) - # TODO 2.2: Implémenter le contrôle d'accès (A01) - # Règle : admin OU propriétaire (is_owner) - # Lever AuthZException si l'accès est refusé + # --- FIX A01 (Broken Access Control) --- + is_admin = "admin" in current_user.roles + is_owner = current_user.id == note.owner_id + + if not (is_admin or is_owner): + raise AuthZException( + "Access denied: You do not have permission to view this note." + ) return note @@ -32,28 +36,40 @@ async def get_my_notes(info: Info) -> List[NoteModel]: async def search_notes(info: Info, contentContains: str) -> List[NoteModel]: """ Resolver pour 'searchNotes' """ - # ... (partie A05) ... + # --- FIX A05 (Injection) --- + # La logique de recherche est déléguée à la DAL (fake_db_repo) + # qui est supposée utiliser des requêtes paramétrées. notes = await fake_db_repo.search_notes_by_content(contentContains) - # TODO 2.2: Récupérer le current_user (get_current_user_from_info) + # --- FIX A01 (Broken Access Control) --- + # ATTENTION: la recherche peut retourner des notes d'autres utilisateurs ! + # Il faut filtrer les résultats après coup. + current_user = get_current_user_from_info(info) + is_admin = "admin" in current_user.roles - # TODO 2.2: Filtrer la liste 'notes' (A01) - # Problème : 'notes' contient les résultats de TOUS les utilisateurs. - # Règle : ne retourner que les notes où l'utilisateur est admin OU propriétaire. + allowed_notes = [] + for n in notes: + if is_admin or n.owner_id == current_user.id: + allowed_notes.append(n) - return notes # INSECURE: retourne tout + return allowed_notes async def get_notes_for_user_resolver(root: UserModel, info: Info) -> List[NoteModel]: """ Resolver pour le champ imbriqué 'User.notes' """ + current_user = get_current_user_from_info(info) + # 'root' est l'objet 'User' parent user_id_to_view = root.id - # TODO 2.3: Récupérer le current_user (get_current_user_from_info) + # --- FIX A01 (Broken Access Control) --- + is_admin = "admin" in current_user.roles + is_self = current_user.id == user_id_to_view - # TODO 2.3: Implémenter le contrôle d'accès (A01) - # Règle : admin OU 'is_self' (l'utilisateur connecté regarde son propre profil) - # Si accès refusé, retourner une liste vide []. + if not (is_admin or is_self): + # On ne lève pas d'erreur, on retourne juste une liste vide + # pour ne pas bloquer la requête parente. + return [] notes = await fake_db_repo.get_notes_for_user(user_id_to_view) - return notes + return notes \ No newline at end of file diff --git a/src/app/graphql/resolvers/user_resolver.py b/src/app/graphql/resolvers/user_resolver.py index fedac0f..0421d6b 100644 --- a/src/app/graphql/resolvers/user_resolver.py +++ b/src/app/graphql/resolvers/user_resolver.py @@ -8,30 +8,53 @@ from app.models.notemodel import NoteModel from app.models.usermodel import UserModel async def get_user_by_id(info: Info, id: strawberry.ID) -> UserModel: - """ Resolver pour le champ 'user(id: ID!)'. """ - # TODO 2.2: Récupérer le current_user (get_current_user_from_info) - + """ + Resolver pour le champ 'user(id: ID!)'. + """ + current_user = get_current_user_from_info(info) user_to_view = await fake_db_repo.get_user_by_id(int(id)) if not user_to_view: raise NotFoundBLLException(resource_name="User", resource_id=id) - # TODO 2.2: Implémenter le contrôle d'accès (A01) - # Règle : admin OU 'is_self' - # Lever AuthZException si l'accès est refusé + # --- FIX A01 (Broken Access Control) --- + # On vérifie si l'utilisateur connecté est admin OU + # s'il demande son propre profil. + is_admin = "admin" in current_user.roles + is_self = current_user.id == user_to_view.id + if not (is_admin or is_self): + raise AuthZException( + "Access denied: You do not have permission to view this user's details." + ) + + # Si la vérification passe, on retourne l'objet return user_to_view async def get_note_owner_resolver(root: NoteModel, info: Info) -> UserModel: - """ Resolver pour le champ imbriqué 'Note.owner'. """ + """ + Resolver pour le champ imbriqué 'Note.owner'. + """ # 'root' est l'objet 'Note' parent owner = await fake_db_repo.get_user_by_id(root.owner_id) if not owner: raise NotFoundBLLException(resource_name="User", resource_id=root.owner_id) - # TODO 2.3: (Défense en profondeur) - # Problème : 'owner' contient des champs sensibles (email). - # Retourner un UserModel, mais avec l'email masqué (ex: "[Redacted]"). + # Note : Pas besoin de check A01 ici car on retourne un type 'User' + # et ses champs sensibles (email) sont protégés par LEURS propres resolvers/permissions. + # ...Cependant, notre 'user' type ne protège 'email' que + # via le resolver 'get_user_by_id'. + # + # Pour un TP, le plus simple est de ne retourner que les champs publics. - return owner # INSECURE: retourne l'objet complet avec l'email \ No newline at end of file + # FIX A01 (Défense en profondeur) + # Pour éviter de fuiter l'email, on ne le retourne QUE si l'utilisateur + # a les droits via le resolver 'get_user_by_id'. + # Ici, on ne retourne pas l'email. + return UserModel( + id=owner.id, + username=owner.username, + email="[Redacted]", # Ou lever une erreur si on tente d'y accéder + roles=owner.roles + ) \ No newline at end of file diff --git a/src/app/main.py b/src/app/main.py index a705e54..5a894b0 100644 --- a/src/app/main.py +++ b/src/app/main.py @@ -6,9 +6,19 @@ from strawberry.fastapi import GraphQLRouter from app.core.config import settings from app.graphql.context import get_context +from app.graphql.extensions import SecurityErrorExtension from app.graphql.mutations import Mutation from app.graphql.queries import Query +# --- FIX A06 (Insecure Design - DoS) --- +from strawberry.extensions import QueryDepthLimiter +from strawberry.extensions import MaxTokensLimiter +from strawberry.extensions import MaxAliasesLimiter +from strawberry.extensions import AddValidationRules + +# --- FIX A06 (Insecure Design - Data Exposure) --- +from graphql.validation import NoSchemaIntrospectionCustomRule + # Crée l'application FastAPI app = FastAPI( title="Secure GraphQL API", @@ -30,11 +40,19 @@ app.add_middleware( # Crée le schéma GraphQL extensions=[ - # TODO : ajouter ici notre extension personnalisé de formatage des exceptions de sécurité (authz et authn) + SecurityErrorExtension, # Notre extension de gestion d'erreurs ] if IS_PRODUCTION: production_extensions = [ - # TODO : ajouter ici les extensions de sécurité à activer en production + # --- FIX A06 (Insecure Design - DoS) --- + # On applique l'extension qui bloque les + # requêtes trop profondes (ex: > 7 niveaux). + QueryDepthLimiter(max_depth=7), + # On empêche l'introspection + AddValidationRules([NoSchemaIntrospectionCustomRule]), + # Autres règles de limitation + MaxAliasesLimiter(max_alias_count=15), + MaxTokensLimiter(max_token_count=1000), ] extensions.extend(production_extensions) @@ -49,7 +67,9 @@ graphql_app = GraphQLRouter( schema, context_getter=get_context, - # TODO : désactiver GraphiQL si on est en production + # --- FIX A02 (Security Misconfiguration) --- + # On désactive l'interface GraphiQL si on est en production. + graphiql=not IS_PRODUCTION ) # Ajoute le routeur à l'application