feat: implémentation des sessions JWT avec cookies sécurisés (expiration 60 min)
Build and Push Docker Image / build-and-push (push) Successful in 1m24s
Build and Push Docker Image / build-and-push (push) Successful in 1m24s
This commit is contained in:
1 parent
b48853d186
commit
c3b10508d9
3 files changed
+58
-8
No files matched your search
+27
-1
@@ -21,4 +21,30 @@ def get_totp_uri(secret: str, username: str) -> str:
|
|||||||
def verify_totp(secret: str, code: str) -> bool:
|
def verify_totp(secret: str, code: str) -> bool:
|
||||||
"""Vérifie le code à 6 chiffres soumis par l'utilisateur."""
|
"""Vérifie le code à 6 chiffres soumis par l'utilisateur."""
|
||||||
totp = pyotp.TOTP(secret)
|
totp = pyotp.TOTP(secret)
|
||||||
return totp.verify(code)
|
return totp.verify(code)
|
||||||
|
|
||||||
|
import jwt
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
import os
|
||||||
|
|
||||||
|
# Dans l'idéal, cette clé secrète sera à placer dans votre fichier .env plus tard
|
||||||
|
SECRET_KEY = os.getenv("SECRET_KEY", "`*\$Tu*CCc5hglT$mX'HykxERDljBz")
|
||||||
|
ALGORITHM = "HS256"
|
||||||
|
ACCESS_TOKEN_EXPIRE_MINUTES = 60
|
||||||
|
|
||||||
|
def create_access_token(data: dict):
|
||||||
|
"""Crée un jeton JWT valable 60 minutes."""
|
||||||
|
to_encode = data.copy()
|
||||||
|
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||||
|
to_encode.update({"exp": expire})
|
||||||
|
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||||
|
|
||||||
|
def verify_token(token: str):
|
||||||
|
"""Vérifie la validité du jeton JWT."""
|
||||||
|
try:
|
||||||
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||||
|
return payload
|
||||||
|
except jwt.ExpiredSignatureError:
|
||||||
|
return None # Le jeton a expiré (60 min écoulées)
|
||||||
|
except jwt.InvalidTokenError:
|
||||||
|
return None # Le jeton est invalide
|
||||||
+29
-6
@@ -1,4 +1,4 @@
|
|||||||
from fastapi import FastAPI, Depends, HTTPException, status
|
from fastapi import FastAPI, Depends, HTTPException, status, Request, Response
|
||||||
from fastapi.responses import RedirectResponse, FileResponse
|
from fastapi.responses import RedirectResponse, FileResponse
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
import os
|
import os
|
||||||
@@ -6,7 +6,7 @@ import os
|
|||||||
# Importation de nos modules locaux
|
# Importation de nos modules locaux
|
||||||
from .database import engine, Base, get_db
|
from .database import engine, Base, get_db
|
||||||
from .models import User
|
from .models import User
|
||||||
from .auth import get_password_hash, generate_totp_secret, get_totp_uri, verify_password, verify_totp
|
from .auth import get_password_hash, generate_totp_secret, get_totp_uri, verify_password, verify_totp, create_access_token, verify_token
|
||||||
from .schemas import AdminCreate, LoginRequest
|
from .schemas import AdminCreate, LoginRequest
|
||||||
|
|
||||||
# Création des tables dans la base de données
|
# Création des tables dans la base de données
|
||||||
@@ -85,8 +85,8 @@ def create_admin(admin_data: AdminCreate, db: Session = Depends(get_db)):
|
|||||||
}
|
}
|
||||||
|
|
||||||
@app.post("/api/login")
|
@app.post("/api/login")
|
||||||
def login(login_data: LoginRequest, db: Session = Depends(get_db)):
|
def login(login_data: LoginRequest, response: Response, db: Session = Depends(get_db)):
|
||||||
"""Vérifie les identifiants et le code 2FA."""
|
"""Vérifie les identifiants et crée une session de 60 minutes."""
|
||||||
user = db.query(User).filter(User.username == login_data.username).first()
|
user = db.query(User).filter(User.username == login_data.username).first()
|
||||||
|
|
||||||
# Vérification de l'utilisateur et du mot de passe
|
# Vérification de l'utilisateur et du mot de passe
|
||||||
@@ -103,8 +103,31 @@ def login(login_data: LoginRequest, db: Session = Depends(get_db)):
|
|||||||
detail="Code 2FA invalide."
|
detail="Code 2FA invalide."
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Création du jeton de session
|
||||||
|
access_token = create_access_token(data={"sub": user.username})
|
||||||
|
|
||||||
|
# Injection du jeton dans un cookie sécurisé (invisible pour le JavaScript côté client)
|
||||||
|
response.set_cookie(
|
||||||
|
key="session_token",
|
||||||
|
value=access_token,
|
||||||
|
httponly=True,
|
||||||
|
max_age=3600, # 3600 secondes = 60 minutes
|
||||||
|
samesite="lax"
|
||||||
|
)
|
||||||
|
|
||||||
return {"message": "Connexion réussie"}
|
return {"message": "Connexion réussie"}
|
||||||
|
|
||||||
@app.get("/dashboard")
|
@app.get("/dashboard")
|
||||||
def dashboard():
|
def dashboard(request: Request):
|
||||||
return {"message": "Bienvenue sur le tableau de bord de l'Orchestrateur AETHAS38 ! (Interface à venir)"}
|
"""Route protégée : nécessite un cookie de session valide."""
|
||||||
|
token = request.cookies.get("session_token")
|
||||||
|
if not token:
|
||||||
|
return RedirectResponse(url="/login")
|
||||||
|
|
||||||
|
payload = verify_token(token)
|
||||||
|
if not payload:
|
||||||
|
# Si le token a expiré (60 min) ou est invalide, retour au login
|
||||||
|
return RedirectResponse(url="/login")
|
||||||
|
|
||||||
|
username = payload.get("sub")
|
||||||
|
return {"message": f"Bienvenue sur le tableau de bord sécurisé, {username} !"}
|
||||||
+2
-1
@@ -9,4 +9,5 @@ psutil
|
|||||||
python-dotenv
|
python-dotenv
|
||||||
python-multipart
|
python-multipart
|
||||||
bcrypt
|
bcrypt
|
||||||
email-validator
|
email-validator
|
||||||
|
PyJWT
|
||||||
Reference in new issue
Block a user