DEV Community: PEMPEME MOHAMED CHAMSOUDINE

Django PayPal Payment API

PEMPEME MOHAMED CHAMSOUDINE — Thu, 06 Feb 2025 14:01:44 +0000

Django PayPal Payment API

A robust Django REST API for handling PayPal payments with comprehensive error handling, logging, and proper separation of concerns.

🌟 Features

Complete PayPal payment integration
RESTful API endpoints
Comprehensive error handling
Detailed logging system
Database tracking of payment status
Support for payment execution and refunds
Clean architecture with separation of concerns

🛠 Technology Stack

Python 3.8+
Django 4.2+
Django REST Framework
PayPal REST SDK
PostgreSQL (recommended) / SQLite

📋 Prerequisites

Before you begin, ensure you have:

Python 3.8 or higher installed
A PayPal Developer account
PayPal API credentials (Client ID and Secret)
PostgreSQL (recommended) or SQLite

⚙️ Installation

Clone the repository:

git clone https://github.com/pemochamdev/PayPal-Payment-using-django.git
cd PayPal-Payment-using-django

Create and activate a virtual environment:

python -m venv env
source env/bin/activate  # On Windows: env\Scripts\activate

Install dependencies:

pip install -r requirements.txt

Create environment variables file (.env):

DJANGO_SECRET_KEY=your_secret_key
PAYPAL_CLIENT_ID=your_paypal_client_id
PAYPAL_CLIENT_SECRET=your_paypal_client_secret
PAYPAL_MODE=sandbox  # or 'live' for production

Run migrations:

python manage.py makemigrations
python manage.py migrate

🚀 Usage

Starting the Server

python manage.py runserver

API Endpoints

Create Payment

POST /api/payments/

Request body:

{
    "amount": 10.00,
    "description": "Payment description"
}

Response:

{
    "id": "uuid",
    "paypal_payment_id": "PAYID-...",
    "approval_url": "https://www.sandbox.paypal.com/..."
}

Execute Payment

POST /api/payments/execute/

Request body:

{
    "paymentId": "PAYID-...",
    "PayerID": "BUYER-ID"
}

Get Payment Details

GET /api/payments/{payment_id}/

Request Refund

POST /api/payments/{payment_id}/refund/

Request body:

{
    "amount": 10.00,
    "reason": "Customer request"
}

📁 Project Structure

paypal-payment-using-django/
    -config(project name)
    -payments(app)
    -requirements.txt
    -READMED.md

💡 Implementation Details

Models

The project uses two main models:

Payment: Tracks payment information and status
PaymentRefund: Handles refund information

Service Layer

The PayPalService class handles all PayPal-related operations:

Payment creation
Payment execution
Refund processing
Error handling
Logging

Error Handling

Custom exceptions for different scenarios:

PaymentError: Base payment exception
PaymentValidationError: For validation issues
PaymentProcessError: For processing errors
RefundError: For refund-related issues

🔒 Security Considerations

All sensitive credentials are stored in environment variables
Input validation for all API endpoints
CSRF protection enabled
Proper error handling to prevent information leakage
Logging of sensitive data is sanitized

📝 Logging

The application implements comprehensive logging:

Payment creation attempts
Execution status
Error tracking
PayPal API responses
Database operations

🌐 Environment Configuration

Development

PAYPAL_MODE = "sandbox"
DEBUG = True

Production

PAYPAL_MODE = "live"
DEBUG = False

🔧 Testing

Run the test suite:

python manage.py test

🤝 Contributing

Fork the repository
Create a feature branch
Commit your changes
Push to the branch
Create a Pull Request

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

👥 Author

Your Name

GitHub: @pemochamdev
LinkedIn: pempeme mohamed chamsoudine

🙏 Acknowledgments

PayPal Developer Documentation
Django Documentation
Django REST Framework Documentation

📞 Support

For support, email pemochamdev@gmail.com or open an issue on GitHub.

Comprendre l'Authentification JWT : De la Théorie à la Pratique

PEMPEME MOHAMED CHAMSOUDINE — Thu, 06 Feb 2025 10:04:37 +0000

Introduction

Bienvenue dans un voyage en profondeur à travers le monde de l'authentification JSON Web Token (JWT). Que vous soyez débutant, intermédiaire ou expert dans le développement web, cet article vous offrira une compréhension exhaustive des JWT, depuis leurs bases jusqu'à leur implémentation dans des applications modernes. Si vous cherchez à sécuriser vos APIs ou applications, comprendre JWT est essentiel.

Qu'est-ce que le JWT ?

JSON Web Token (JWT) est un standard ouvert (RFC 7519) qui définit une méthode compacte et sécurisée pour transmettre des informations entre les parties sous forme de JSON. JWT est principalement utilisé pour l'authentification, mais il peut aussi servir à l'autorisation et à l'échange d'informations.

Structure d'un JWT

Un JWT se compose de trois parties, séparées par des points (.):

En-tête (Header):

{
  "alg": "HS256",
  "typ": "JWT"
}

alg : Algorithme de signature (ici, HMAC-SHA256).
typ : Type de token (JWT).

Charge utile (Payload):

{
  "sub": "1234567890",
  "name": "John Doe",
  "iat": 1516239022,
  "exp": 1516242622
}

sub : Sujet, souvent l'ID de l'utilisateur.
iat : Issued At, heure d'émission du token.
exp : Expiration du token.

Signature:

Calculée en signant l'en-tête et la charge utile avec une clé secrète ou une paire de clés publique/privée.

L'ensemble donne quelque chose comme :

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

Comment ça fonctionne ?

Génération : Lors de l'authentification, un serveur crée un JWT avec les informations de l'utilisateur, le signe avec une clé secrète, puis l'envoie au client.
*Stockage *: Le client stocke le JWT, généralement dans le localStorage, sessionStorage, ou dans des cookies HttpOnly.
Utilisation : Pour chaque requête nécessitant une authentification, le client inclut le JWT dans l'en-tête HTTP, typiquement sous la forme Authorization: Bearer .
*Validation *: Le serveur vérifie le JWT en :
- Décodant le token.
- Vérifiant la signature pour s'assurer que le token n'a pas été altéré.
- Vérifiant les claims comme l'expiration pour autoriser ou refuser l'accès.

Avantages et Défis

Avantages

Stateless : Aucun besoin de session côté serveur, améliorant la scalabilité.
Sécurité : La signature assure l'intégrité du token.
Portable : Le token peut être utilisé sur différents systèmes sans nécessiter de validation centralisée.

Défis :

Révocation : Les JWT sont auto-suffisants; la révocation n'est pas directe.
Sécurité des Clés : La sécurité repose sur la confidentialité de la clé secrète.
Expiration : Nécessite des stratégies de rafraîchissement ou d'expiration pour gérer la durée de vie des tokens.

Implémentation Pratique avec Python Flask et FastAPI

Pour l'implementation complete consulter l'article: implementation-dune-authentification-jwt-complete-en-fastapi

Conclusion
L'authentification JWT est une puissante méthode pour sécuriser les applications web modernes. En comprenant et en implémentant correctement JWT, vous pouvez offrir une expérience utilisateur fluide tout en assurant un haut niveau de sécurité. Que vous débutiez ou que vous soyez un développeur expérimenté, les JWT offrent une solution scalable et flexible pour la gestion de l'authentification.

N'oubliez pas que la sécurité est un voyage continu; restez informé des meilleures pratiques et des nouvelles vulnérabilités.

Ressources
JWT.io - Pour encoder/décoder des JWT et comprendre leur structure.

Flask-JWT-Extended - Docs pour l'extension Flask utilisée dans l'exemple.

FastAPI Security - Docs sur la sécurité avec FastAPI.

En suivant ces conseils et en utilisant les exemples fournis, vous êtes prêt à intégrer JWT dans vos projets pour une authentification robuste et efficace.

Intégration de Stripe dans Django pour la Gestion des Paiements : Un Guide Complet

PEMPEME MOHAMED CHAMSOUDINE — Wed, 05 Feb 2025 02:02:27 +0000

Introduction

L'intégration de systèmes de paiement comme Stripe dans vos applications web est cruciale pour gérer les transactions en ligne de manière sécurisée et efficace. Dans cet article, nous allons explorer comment utiliser Django pour structurer une API de paiement avec Stripe, en se concentrant sur la gestion des clients, des méthodes de paiement, des paiements uniques, des remboursements et des abonnements.

Le code source est disponible sur github via ce lien: api-payment-stripe-in-drf

Modèles Django

Voici les modèles définis dans core/models.py pour gérer les entités de paiement :

Customer

class Customer(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE)
    stripe_customer_id = models.CharField(max_length=255, unique=True)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f"Customer {self.user.email}"

Utilisation: Chaque utilisateur a un client Stripe associé pour gérer ses informations de paiement.

PaymentMethod

class PaymentMethod(models.Model):
    TYPES = (
        ('card', 'Carte bancaire'),
        ('sepa_debit', 'Prélèvement SEPA'),
        ('bancontact', 'Bancontact'),
        ('giropay', 'Giropay'),
        ('ideal', 'iDEAL'),
        ('p24', 'P24'),
        ('sofort', 'Sofort'),
        ('apple_pay', 'Apple Pay'),
        ('google_pay', 'Google Pay'),
    )

    customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
    stripe_payment_method_id = models.CharField(max_length=255)
    type = models.CharField(max_length=20, choices=TYPES)
    last4 = models.CharField(max_length=4, null=True, blank=True)
    is_default = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

Utilisation: Pour stocker différentes méthodes de paiement pour chaque client.

Payment

class Payment(models.Model):
    STATUS_CHOICES = (
        ('pending', 'En attente'),
        ('processing', 'En cours'),
        ('succeeded', 'Réussi'),
        ('failed', 'Échoué'),
        ('canceled', 'Annulé'),
        ('refunded', 'Remboursé'),
        ('partially_refunded', 'Partiellement remboursé'),
    )

    customer = models.ForeignKey(Customer, on_delete=models.SET_NULL, null=True)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    currency = models.CharField(max_length=3, default='EUR')
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
    payment_method = models.ForeignKey(PaymentMethod, on_delete=models.SET_NULL, null=True)
    stripe_payment_intent_id = models.CharField(max_length=255, unique=True)
    description = models.TextField(blank=True)
    metadata = models.JSONField(default=dict)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

Utilisation: Pour enregistrer chaque transaction avec son statut et détails.

Refund

class Refund(models.Model):
    payment = models.ForeignKey(Payment, on_delete=models.CASCADE)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    stripe_refund_id = models.CharField(max_length=255, unique=True)
    reason = models.CharField(max_length=255)
    status = models.CharField(max_length=20)
    created_at = models.DateTimeField(auto_now_add=True)

Utilisation: Gère les remboursements liés à des paiements.

Subscription

class Subscription(models.Model):
    customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
    stripe_subscription_id = models.CharField(max_length=255, unique=True)
    status = models.CharField(max_length=20)
    current_period_start = models.DateTimeField()
    current_period_end = models.DateTimeField()
    cancel_at_period_end = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

Utilisation: Pour les abonnements récurrents.

Services Stripe Dans core/services.py, nous avons des méthodes pour interagir avec l'API Stripe :

Création d'un Client

import stripe
from django.conf import settings
import stripe.error
from core.models import Customer,Payment,PaymentMethod,Refund,Subscription

stripe.api_key = settings.STRIPE_SECRET_KEY

class StripeService:

    @staticmethod
    def create_account(user, token=None):
        try:
            customer = stripe.Customer.create(
                email = user.email,
                source=token
            )

            return Customer.objects.create(
                user = user,
                stripe_customer_id = customer.id
            )
        except stripe.error.StripeError as e:
            raise Exception(f"Erreur lors de la création du client: {str(e)}")



    @staticmethod
    def add_payment_method(customer: Customer,payment_method_id):
        try:
            payment_method = stripe.PaymentMethod.attach(
                payment_method_id,
                customer=customer.stripe_customer_id
            )

            return PaymentMethod.objects.create(
                stripe_payment_method_id = payment_method.id,
                customer = customer,
                type = payment_method.type,
                last4 = payment_method.card.last4 if payment_method.type =="card" else None
            )
        except stripe.error.StripeError as e:
            raise Exception(f"Erreur lors de l'ajout du moyen de paiement: {str(e)}")


    @staticmethod
    def create_payment_intent(customer, amount, currency, payment_method_id=None):

        try:
            intent_params = {
                'amount': int(amount * 100),
                'currency': currency,
                'customer': customer.stripe_customer_id,
                'payment_method_types': ['card', 'sepa_debit', 'bancontact', 
                                       'giropay', 'ideal', 'p24', 'sofort'],
                'setup_future_usage': 'off_session',
            }

            if payment_method_id:
                intent_params['payment_method'] = payment_method_id

                intent = stripe.PaymentIntent.create(**intent_params)

                return Payment.objects.create(
                    customer=customer,
                    currency=currency,
                    stripe_payment_intent_id = intent.id

                ), intent.client_secret
        except stripe.error.StripeError as e:
            raise Exception(f"Erreur lors de la création du paiement: {str(e)}")


    @staticmethod
    def process_refund(payment, amount=None, reason=None):
        try:
            refund_params = {
                'payment_intent': payment.stripe_payment_intent_id,
            }

            if amount:
                refund_params['amount'] = int(amount * 100)
            if reason:
                refund_params['reason'] = reason

            refund = stripe.Refund.create(**refund_params)

            return Refund.objects.create(
                payment=payment,
                amount=amount or payment.amount,
                stripe_refund_id=refund.id,
                reason=reason or 'requested_by_customer',
                status=refund.status
            )

        except stripe.error.StripeError as e:
            raise Exception(f"Erreur lors du remboursement: {str(e)}")


    @staticmethod
    def create_subscription(customer, price_id, payment_method_id=None):
        try:
            subscription = stripe.Subscription.create(
                customer=customer.stripe_customer_id,
                items=[{'price': price_id}],
                payment_method=payment_method_id,
                expand=['latest_invoice.payment_intent']
            )

            return Subscription.objects.create(
                customer=customer,
                stripe_subscription_id=subscription.id,
                status=subscription.status,
                current_period_start=subscription.current_period_start,
                current_period_end=subscription.current_period_end
            )
        except stripe.error.StripeError as e:
            raise Exception(f"Erreur lors de la création de l'abonnement: {str(e)}")

Seralizers


from rest_framework import serializers
from core.models import Customer, PaymentMethod, Payment, Refund, Subscription

class CustomerSerializer(serializers.ModelSerializer):
    class Meta:
        model = Customer
        fields = ['id', 'stripe_customer_id', 'created_at']

class PaymentMethodSerializer(serializers.ModelSerializer):
    class Meta:
        model = PaymentMethod
        fields = ['id', 'type', 'last4', 'is_default', 'created_at']

class PaymentSerializer(serializers.ModelSerializer):
    class Meta:
        model = Payment
        fields = ['id', 'amount', 'currency', 'status', 'description', 
                 'metadata', 'created_at', 'updated_at']

class RefundSerializer(serializers.ModelSerializer):
    class Meta:
        model = Refund
        fields = ['id', 'payment', 'amount', 'reason', 'status', 'created_at']

class SubscriptionSerializer(serializers.ModelSerializer):
    class Meta:
        model = Subscription
        fields = ['id', 'status', 'current_period_start', 
                 'current_period_end', 'cancel_at_period_end']

API Views Les ViewSets dans core/views.py facilitent l'interaction RESTful avec les modèles :

CustomerViewSet, PaymentViewSet, SubscriptionViewSet: Offrent des points d'entrée pour CRUD sur les clients, paiements, et abonnements.

Suivez le lien github pour avoir le code source complet:api-payment-stripe-in-drf.git

Conclusion
Cette structure permet de construire une API de paiement robuste avec Django et Stripe. Assurez-vous de sécuriser votre application en utilisant HTTPS, de gérer les exceptions correctement et de tester chaque composant avec des scénarios réels. Pour une sécurité accrue, utilisez des environnements de staging avant de passer en production.

Authentification JWT Complete en FastAPI [link](https://dev.to/pemochamdev/implementation-dune-authentification-jwt-complete-en-fastapi-5gjb)

PEMPEME MOHAMED CHAMSOUDINE — Tue, 04 Feb 2025 16:01:24 +0000

Implementation d'une Authentification JWT Complete en FastAPI https://dev.to/pemochamdev/implementation-dune-authentification-jwt-complete-en-fastapi-5gjb

PEMPEME MOHAMED CHAMSOUDINE — Tue, 04 Feb 2025 15:58:25 +0000

Implémentation d'une Authentification JWT Complète en FastAPI

PEMPEME MOHAMED CHAMSOUDINE — Tue, 04 Feb 2025 15:05:30 +0000

Introduction

Dans cet article, nous allons explorer comment mettre en place une authentification JWT complète en utilisant FastAPI, incluant la gestion des tokens d'accès et de rafraîchissement, des rôles d'utilisateurs, et la sécurisation des données sensibles avec des variables d'environnement.
le code completest disponible sur le auth-jwt-fastapi

Pré-requis
Python 3.8+
FastAPI
SQLAlchemy pour la gestion de la base de données
Pydantic pour la modélisation des données
python-jose pour JWT
passlib pour le hachage des mots de passe
python-dotenv pour charger les variables d'environnement

Installation des Dépendances

pip install fastapi[all] "python-jose[cryptography]" "passlib[bcrypt]" python-multipart email-validator python-dotenv

Configuration avec Variables d'Environnement Pour sécuriser les données sensibles:

-Créer un fichier .env:

SECRET_KEY=your-very-secure-secret-key
MAIL_USERNAME=your-mail-username
MAIL_PASSWORD=your-mail-password
MAIL_FROM=your-from-email@example.com

Fichier config.py pour gérer les configurations:

from pydantic_settings import BaseSettings
from dotenv import load_dotenv

load_dotenv()

class Settings(BaseSettings):
    SECRET_KEY: str  
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 15
    REFRESH_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 7
    MAIL_USERNAME: str 
    MAIL_PASSWORD: str
    MAIL_FROM: str 
    MAIL_PORT: int = 587
    MAIL_SERVER: str = "smtp.gmail.com"
    MAIL_TLS: bool = True
    MAIL_SSL: bool = False
    USE_CREDENTIALS: bool = True
    VALIDATE_CERTS: bool = True
    DATABASE_URL: str = ""

    class Config:
        env_file = ".env"

settings = Settings()

Modélisation des Données Les modeles

from pydantic import BaseModel
from typing import List
from passlib.context import CryptContext
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from db import Base

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class Role(BaseModel):
    name: str

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    roles = relationship("Role", secondary="user_roles")
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    def verify_password(self, plain_password):
        return pwd_context.verify(plain_password, self.hashed_password)

    def get_password_hash(password):
        return pwd_context.hash(password)

class UserInDB(User):
    hashed_password: str

class UserCreate(BaseModel):
    username: str
    email: str
    password: str

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str

class UserRoles(BaseModel):
    roles: List[Role] = []

Gestion des Tokens JWT Génération et Décodage des Tokens

from jose import JWTError, jwt
from datetime import datetime, timedelta
from config import settings
from typing import Optional
from fastapi import HTTPException

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt



def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.REFRESH_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt



def decode_token(token: str):
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Implémentation des Routes avec FastAPI

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from db import SessionLocal, engine
from models import User,UserRoles
from db import Base
import crud
from jose import JWTError
from models import UserCreate, Token
from tokens import decode_token,create_access_token,create_refresh_token

app = FastAPI()
Base.metadata.create_all(bind=engine)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    payload = decode_token(token)
    user = crud.get_user_by_username(db, username=payload.get("sub"))
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = crud.authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    refresh_token = create_refresh_token(data={"sub": user.username})
    return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"}

@app.post("/refresh", response_model=Token)
async def refresh_token(refresh_token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    try:
        payload = decode_token(refresh_token)
        user = crud.get_user_by_username(db, username=payload.get("sub"))
        if not user:
            raise HTTPException(status_code=401, detail="User not found")
        access_token = create_access_token(data={"sub": user.username})
        return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"}
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

@app.post("/users/", response_model=User)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = crud.create_user(db=db, user=user)
    return db_user

@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

@app.get("/users/me/roles/", response_model=UserRoles)
async def read_user_roles(current_user: User = Depends(get_current_active_user)):
    return {"roles": current_user.roles}

CRUD Operations

from sqlalchemy.orm import Session
from models import User
from models import UserCreate
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_user(db: Session, user: UserCreate):
    hashed_password = pwd_context.hash(user.password)
    db_user = User(username=user.username, email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user_by_username(db: Session, username: str):
    return db.query(User).filter(User.username == username).first()

def authenticate_user(db: Session, username: str, password: str):
    user = get_user_by_username(db, username)
    if not user or not user.verify_password(password):
        return False
    return user

database

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from app.config import settings

SQLALCHEMY_DATABASE_URL = settings.DATABASE_URL

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Conclusion
Ce tutoriel vous a guidé à travers une implémentation détaillée d'une authentification JWT en FastAPI, couvrant la gestion sécurisée des configurations, la modélisation des utilisateurs et des rôles, et les opérations de base sur ces données. En utilisant ces concepts, vous pouvez construire des applications web sécurisées et évolutives.

Points Clés
Utilisation de variables d'environnement pour sécuriser les données sensibles.
Gestion des tokens d'accès et de rafraîchissement pour une expérience utilisateur améliorée.
Prise en charge des rôles pour une gestion fine des permissions.

Tags: #Python #FastAPI #JWT #Authentication #Security #Pydantic

AUTHENTIFICATION JWT AVEC DJANGO

PEMPEME MOHAMED CHAMSOUDINE — Tue, 04 Feb 2025 13:38:52 +0000

Développeurs, découvrez notre avancée en sécurité et gestion d'utilisateurs !

J'ai récemment implémenté un système de gestion d'utilisateurs et de sécurité robuste avec Django et Django REST Framework. Voici ce que notre solution apporte :

Inscription et Authentification Sécurisée:

Utilisation de APIView et TokenObtainPairView pour une gestion JWT avancée avec RefreshToken.
Protection contre les attaques par force brute grâce à throttle_scope pour les vues RegisterView et CustomTokenObtainPairView.
Enregistrement de l'historique des connexions pour chaque utilisateur avec LoginHistory pour une traçabilité complète.

Gestion des Mots de Passe:

Vue ChangePasswordView avec IsAuthenticated pour garantir que seul l'utilisateur peut changer son mot de passe, incluant la mise à jour de password_changed_at et l'invalidation des anciens tokens.

Profil Utilisateur et Historique de Connexion:

UserProfileView pour une mise à jour sécurisée du profil.
LoginHistoryView pour permettre aux utilisateurs de consulter leur historique de connexion, optimisé avec select_related pour minimiser les requêtes SQL.

Déconnexion Sécurisée:

LogoutView avec gestion de RefreshToken pour blacklister le token lors de la déconnexion.

Notre approche combine :

Sécurité: Utilisation de AllowAny pour l'inscription mais IsAuthenticated pour les actions sensibles.

Performances: Gestion des tentatives de connexion pour éviter les blocages par DoS, et optimisation des requêtes avec Django ORM.

Expérience Utilisateur: Notifications claires des succès ou des erreurs grâce à des messages traduits avec gettext_lazy.

Trêve de bavardage passons a l'implémentation.

manage.py

from django.contrib.auth.models import BaseUserManager
from django.utils.translation import gettext_lazy as _


class CustomUserManager(BaseUserManager):
    def create_user(self,email,password=None,**extra_fields):
        if not email:
            raise ValueError(_("L'email est obligatoire"))

        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.set_password(password)
        user.save
        return user

    def create_superuser(self,email,password,**extra_fields):
        extra_fields.setdefault('is_staff',True)
        extra_fields.setdefault('is_superuser',True)
        extra_fields.setdefault('is_active',True)

        return self.create_user(email,password,**extra_fields)

models.py

from django.db import models
from django.contrib.auth.models import AbstractUser
from django.core.validators import RegexValidator
from django.utils.translation import gettext_lazy as _
import uuid
from users.managers import CustomUserManager

class User(AbstractUser):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    username = None
    email = models.EmailField(
        _('adresse email'), 
        unique=True,
        error_messages={
            'unique': _("Un utilisateur avec cette adresse email existe déjà."),
        }
    )
    first_name = models.CharField(_('prénom'), max_length=150)
    last_name = models.CharField(_('nom'), max_length=150)
    phone_regex = RegexValidator(
        regex=r'^\+?1?\d{9,15}$',
        message=_("Le numéro de téléphone doit être au format: '+999999999'. 15 chiffres maximum.")
    )
    phone = models.CharField(
        _('téléphone'), 
        validators=[phone_regex], 
        max_length=15, 
        blank=True
    )
    date_of_birth = models.DateField(_('date de naissance'), null=True, blank=True)
    is_verified = models.BooleanField(_('vérifié'), default=False)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    last_login_ip = models.GenericIPAddressField(null=True, blank=True)
    failed_login_attempts = models.IntegerField(default=0)
    is_locked = models.BooleanField(default=False)
    lock_timestamp = models.DateTimeField(null=True, blank=True)
    password_changed_at = models.DateTimeField(null=True, blank=True)

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['first_name', 'last_name']

    objects = CustomUserManager()

    class Meta:
        verbose_name = _('user')
        verbose_name_plural = _('users')
        ordering = ['-created_at']

    def __str__(self):
        return self.email

    def lock_account(self):
        from django.utils import timezone
        self.is_locked = True
        self.lock_timestamp = timezone.now()
        self.save()

    def reset_login_attempts(self):
        self.failed_login_attempts = 0
        self.is_locked = False
        self.lock_timestamp = None
        self.save()

class UserProfile(models.Model):
    GENDER_CHOICES = [
        ('M', _('Masculin')),
        ('F', _('Féminin')),
        ('O', _('Autre')),
    ]

    user = models.OneToOneField(
        User, 
        on_delete=models.CASCADE, 
        related_name='profile'
    )
    avatar = models.ImageField(
        upload_to='avatars/%Y/%m/', 
        null=True, 
        blank=True
    )
    bio = models.TextField(max_length=500, blank=True)
    location = models.CharField(max_length=100, blank=True)
    preferences = models.JSONField(
        default=dict,
        help_text=_("Préférences utilisateur en format JSON")
    )
    gender = models.CharField(
        max_length=1,
        choices=GENDER_CHOICES,
        blank=True
    )
    website = models.URLField(blank=True)

    class Meta:
        verbose_name = _('profil utilisateur')
        verbose_name_plural = _('profils utilisateurs')

    def __str__(self):
        return f"Profil de {self.user.email}"

class LoginHistory(models.Model):
    LOGIN_STATUS_CHOICES = [
        ('success', _('Succès')),
        ('failed', _('Échec')),
        ('locked', _('Compte verrouillé')),
    ]

    user = models.ForeignKey(
        User, 
        on_delete=models.CASCADE,
        related_name='login_history'
    )
    login_datetime = models.DateTimeField(auto_now_add=True)
    ip_address = models.GenericIPAddressField()
    user_agent = models.CharField(max_length=255)
    status = models.CharField(
        max_length=20,
        choices=LOGIN_STATUS_CHOICES,
        default='failed'
    )
    location_city = models.CharField(max_length=100, blank=True)
    location_country = models.CharField(max_length=100, blank=True)

    class Meta:
        verbose_name = _('historique de connexion')
        verbose_name_plural = _('historiques de connexion')
        ordering = ['-login_datetime']
        indexes = [
            models.Index(fields=['user', '-login_datetime']),
        ]

    def __str__(self):
        return f"{self.user.email} - {self.login_datetime} - {self.status}"

serializers.py

from rest_framework import serializers
from django.contrib.auth import get_user_model, password_validation
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
from users.models import UserProfile, LoginHistory
from django.utils.translation import gettext_lazy as _
from django.core.exceptions import ValidationError
from django.utils import timezone
import re

User = get_user_model()

class UserProfileSerializer(serializers.ModelSerializer):
    avatar_url = serializers.SerializerMethodField()

    class Meta:
        model = UserProfile
        fields = ('avatar', 'avatar_url', 'bio', 'location', 'preferences', 'gender', 'website')
        extra_kwargs = {
            'avatar': {'write_only': True}
        }

    def get_avatar_url(self, obj):
        if obj.avatar:
            return self.context['request'].build_absolute_uri(obj.avatar.url)
        return None

    def validate_website(self, value):
        if value and not value.startswith(('http://', 'https://')):
            value = 'https://' + value
        return value

class UserSerializer(serializers.ModelSerializer):
    profile = UserProfileSerializer(required=False)
    password = serializers.CharField(write_only=True, required=True)
    password_confirm = serializers.CharField(write_only=True, required=True)
    full_name = serializers.SerializerMethodField()

    class Meta:
        model = User
        fields = ('id', 'email', 'password', 'password_confirm', 'first_name', 
                 'last_name', 'phone', 'date_of_birth', 'profile', 'full_name',
                 'created_at', 'is_verified')
        extra_kwargs = {
            'first_name': {'required': True},
            'last_name': {'required': True},
            'email': {'required': True},
            'created_at': {'read_only': True},
            'is_verified': {'read_only': True}
        }

    def get_full_name(self, obj):
        return f"{obj.first_name} {obj.last_name}"

    def validate_email(self, value):
        if User.objects.filter(email__iexact=value).exists():
            raise serializers.ValidationError(_("Un utilisateur avec cette adresse email existe déjà."))
        return value.lower()

    def validate_phone(self, value):
        if value:
            pattern = r'^\+?1?\d{9,15}$'
            if not re.match(pattern, value):
                raise serializers.ValidationError(
                    _("Format de téléphone invalide. Utilisez le format: '+999999999'")
                )
        return value

    def validate_password(self, value):
        try:
            password_validation.validate_password(value)
        except ValidationError as e:
            raise serializers.ValidationError(list(e.messages))
        return value

    def validate(self, attrs):
        if attrs['password'] != attrs.pop('password_confirm'):
            raise serializers.ValidationError({"password": _("Les mots de passe ne correspondent pas")})

        if attrs.get('date_of_birth'):
            if attrs['date_of_birth'] > timezone.now().date():
                raise serializers.ValidationError(
                    {"date_of_birth": _("La date de naissance ne peut pas être dans le futur")}
                )
        return attrs

    def create(self, validated_data):
        profile_data = validated_data.pop('profile', None)
        password = validated_data.pop('password')

        user = User.objects.create(**validated_data)
        user.set_password(password)
        user.password_changed_at = timezone.now()
        user.save()

        if profile_data:
            UserProfile.objects.create(user=user, **profile_data)
        return user

    def update(self, instance, validated_data):
        profile_data = validated_data.pop('profile', None)
        password = validated_data.pop('password', None)

        for attr, value in validated_data.items():
            setattr(instance, attr, value)

        if password:
            instance.set_password(password)
            instance.password_changed_at = timezone.now()

        instance.save()

        if profile_data and hasattr(instance, 'profile'):
            for attr, value in profile_data.items():
                setattr(instance.profile, attr, value)
            instance.profile.save()
        elif profile_data:
            UserProfile.objects.create(user=instance, **profile_data)

        return instance

class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
    def validate(self, attrs):
        try:
            data = super().validate(attrs)
        except Exception as e:
            user = User.objects.filter(email=attrs.get('email')).first()
            if user:
                user.failed_login_attempts += 1
                if user.failed_login_attempts >= 5:  # Configurable
                    user.lock_account()
                user.save()
            raise

        user = self.user
        if user.is_locked:
            raise serializers.ValidationError({
                "error": _("Compte verrouillé. Veuillez contacter le support."),
                "locked_since": user.lock_timestamp
            })

        data.update({
            'user': {
                'id': str(user.id),
                'email': user.email,
                'full_name': f"{user.first_name} {user.last_name}",
                'is_verified': user.is_verified,
                'profile': UserProfileSerializer(
                    user.profile,
                    context=self.context
                ).data if hasattr(user, 'profile') else None
            }
        })

        user.reset_login_attempts()
        return data

class ChangePasswordSerializer(serializers.Serializer):
    old_password = serializers.CharField(required=True)
    new_password = serializers.CharField(required=True)
    new_password_confirm = serializers.CharField(required=True)

    def validate_old_password(self, value):
        user = self.context['request'].user
        if not user.check_password(value):
            raise serializers.ValidationError(_("Ancien mot de passe incorrect"))
        return value

    def validate_new_password(self, value):
        try:
            password_validation.validate_password(value, self.context['request'].user)
        except ValidationError as e:
            raise serializers.ValidationError(list(e.messages))
        return value

    def validate(self, attrs):
        if attrs['new_password'] != attrs['new_password_confirm']:
            raise serializers.ValidationError({
                "new_password": _("Les nouveaux mots de passe ne correspondent pas")
            })
        if attrs['old_password'] == attrs['new_password']:
            raise serializers.ValidationError({
                "new_password": _("Le nouveau mot de passe doit être différent de l'ancien")
            })
        return attrs

class LoginHistorySerializer(serializers.ModelSerializer):
    user_email = serializers.CharField(source='user.email', read_only=True)

    class Meta:
        model = LoginHistory
        fields = ('id', 'user_email', 'login_datetime', 'ip_address', 
                 'user_agent', 'status', 'location_city', 'location_country')
        read_only_fields = ('id', 'login_datetime')

views.py

from rest_framework import status, generics
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework_simplejwt.views import TokenObtainPairView
from rest_framework_simplejwt.tokens import RefreshToken, TokenError
from django.contrib.auth import get_user_model
from django.utils.translation import gettext_lazy as _
from django.core.cache import cache
from django.utils import timezone
from users.serializers import (
    UserSerializer, 
    CustomTokenObtainPairSerializer,
    ChangePasswordSerializer,
    LoginHistorySerializer
)
from .models import LoginHistory
import logging

logger = logging.getLogger(__name__)
User = get_user_model()

class RegisterView(APIView):
    permission_classes = (AllowAny,)
    throttle_scope = 'registration'  # Protection contre les attaques par force brute

    def post(self, request):
        try:
            serializer = UserSerializer(data=request.data, context={'request': request})
            if serializer.is_valid():
                user = serializer.save()

                # Créer les tokens JWT
                refresh = RefreshToken.for_user(user)

                # Enregistrer l'historique
                LoginHistory.objects.create(
                    user=user,
                    ip_address=self._get_client_ip(request),
                    user_agent=request.META.get('HTTP_USER_AGENT', ''),
                    status='success'
                )

                return Response({
                    'user': UserSerializer(user, context={'request': request}).data,
                    'refresh': str(refresh),
                    'access': str(refresh.access_token),
                }, status=status.HTTP_201_CREATED)
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
        except Exception as e:
            logger.error(f"Error in registration: {str(e)}", exc_info=True)
            return Response(
                {'error': _("Une erreur est survenue lors de l'inscription")},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

    def _get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0].strip()
        return request.META.get('REMOTE_ADDR')



class CustomTokenObtainPairView(TokenObtainPairView):
    serializer_class = CustomTokenObtainPairSerializer
    throttle_scope = 'login'

    def post(self, request, *args, **kwargs):
        try:
            # Vérifier le nombre de tentatives par IP
            ip = self._get_client_ip(request)
            attempts_key = f"login_attempts_{ip}"
            attempts = cache.get(attempts_key, 0)

            if attempts >= 5:
                return Response(
                    {'error': _("Trop de tentatives. Réessayez plus tard.")},
                    status=status.HTTP_429_TOO_MANY_REQUESTS
                )

            serializer = self.get_serializer(data=request.data)

            try:
                serializer.is_valid(raise_exception=True)
            except ValidationError as e:
                # Incrémenter le compteur de tentatives
                cache.set(attempts_key, attempts + 1, timeout=300)
                return Response(e.detail, status=status.HTTP_400_BAD_REQUEST)

            user = serializer.user
            cache.delete(attempts_key)  # Réinitialiser les tentatives en cas de succès

            # Enregistrer l'historique de connexion
            LoginHistory.objects.create(
                user=user,
                ip_address=ip,
                user_agent=request.META.get('HTTP_USER_AGENT', ''),
                status='success',
                location_city=self._get_location_city(ip),
                location_country=self._get_location_country(ip)
            )

            # Mettre à jour les informations de connexion
            user.last_login_ip = ip
            user.last_login = timezone.now()
            user.save(update_fields=['last_login_ip', 'last_login'])

            return Response(serializer.validated_data)

        except Exception as e:
            logger.error(f"Error in login: {str(e)}", exc_info=True)
            return Response(
                {'error': _("Une erreur est survenue lors de la connexion")},
                status=status.HTTP_400_BAD_REQUEST
            )


    def _get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0].strip()
        return request.META.get('REMOTE_ADDR')

    def _get_location_city(self, ip):
        # Implémenter la géolocalisation IP ici
        return ""

    def _get_location_country(self, ip):
        # Implémenter la géolocalisation IP ici
        return ""

class ChangePasswordView(generics.UpdateAPIView):
    permission_classes = (IsAuthenticated,)
    serializer_class = ChangePasswordSerializer
    throttle_scope = 'password_change'

    def update(self, request, *args, **kwargs):
        try:
            serializer = self.get_serializer(data=request.data)
            if serializer.is_valid():
                user = request.user

                # Changer le mot de passe
                user.set_password(serializer.validated_data['new_password'])
                user.password_changed_at = timezone.now()
                user.save()

                # Blacklister tous les tokens existants
                RefreshToken.for_user(user)

                # Créer de nouveaux tokens
                refresh = RefreshToken.for_user(user)

                return Response({
                    'message': _("Mot de passe modifié avec succès"),
                    'refresh': str(refresh),
                    'access': str(refresh.access_token)
                }, status=status.HTTP_200_OK)

            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

        except Exception as e:
            logger.error(f"Error in password change: {str(e)}", exc_info=True)
            return Response(
                {'error': _("Une erreur est survenue lors du changement de mot de passe")},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

class UserProfileView(generics.RetrieveUpdateAPIView):
    permission_classes = (IsAuthenticated,)
    serializer_class = UserSerializer

    def get_object(self):
        return self.request.user

    def update(self, request, *args, **kwargs):
        try:
            partial = kwargs.pop('partial', False)
            instance = self.get_object()
            serializer = self.get_serializer(
                instance, 
                data=request.data, 
                partial=partial,
                context={'request': request}
            )

            if serializer.is_valid():
                self.perform_update(serializer)
                return Response(serializer.data)

            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

        except Exception as e:
            logger.error(f"Error updating profile: {str(e)}", exc_info=True)
            return Response(
                {'error': _("Une erreur est survenue lors de la mise à jour du profil")},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

class LoginHistoryView(generics.ListAPIView):
    permission_classes = (IsAuthenticated,)
    serializer_class = LoginHistorySerializer
    pagination_class = None  # Ou définir une pagination personnalisée

    def get_queryset(self):
        return (LoginHistory.objects
                .filter(user=self.request.user)
                .select_related('user')
                .order_by('-login_datetime'))

class LogoutView(APIView):
    permission_classes = (IsAuthenticated,)

    def post(self, request):
        try:
            refresh_token = request.data.get("refresh_token")
            if not refresh_token:
                return Response(
                    {'error': _("Le token de rafraîchissement est requis")},
                    status=status.HTTP_400_BAD_REQUEST
                )

            token = RefreshToken(refresh_token)
            token.blacklist()

            # Enregistrer la déconnexion
            LoginHistory.objects.create(
                user=request.user,
                ip_address=self._get_client_ip(request),
                user_agent=request.META.get('HTTP_USER_AGENT', ''),
                status='logout'
            )

            return Response(status=status.HTTP_204_NO_CONTENT)

        except TokenError:
            return Response(
                {'error': _("Token invalide ou expiré")},
                status=status.HTTP_400_BAD_REQUEST
            )
        except Exception as e:
            logger.error(f"Error in logout: {str(e)}", exc_info=True)
            return Response(
                {'error': _("Une erreur est survenue lors de la déconnexion")},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

    def _get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0].strip()
        return request.META.get('REMOTE_ADDR')

urls.py


from django.urls import path
from rest_framework_simplejwt.views import TokenRefreshView
from users.views import (
    RegisterView,
    CustomTokenObtainPairView,
    ChangePasswordView,
    UserProfileView,
    LoginHistoryView,
    LogoutView
)

urlpatterns = [
    path('register/', RegisterView.as_view(), name='register'),
    path('login/', CustomTokenObtainPairView.as_view(), name='token_obtain_pair'),
    path('token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
    path('change-password/', ChangePasswordView.as_view(), name='change_password'),
    path('profile/', UserProfileView.as_view(), name='user_profile'),
    path('login-history/', LoginHistoryView.as_view(), name='login_history'),
    path('logout/', LogoutView.as_view(), name='logout'),
]