127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import jwt
|
|
from fastapi import Cookie, Depends, HTTPException, Request, status
|
|
from passlib.context import CryptContext
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import models
|
|
from .database import get_db
|
|
|
|
|
|
AUTH_COOKIE_NAME = "professeur_top_session"
|
|
AUTH_ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "720"))
|
|
AUTH_SECRET_KEY = os.getenv("AUTH_SECRET_KEY") or os.getenv("SECRET_KEY") or "dev-secret-change-me"
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
return pwd_context.verify(password, password_hash)
|
|
|
|
|
|
def create_access_token(user: models.User) -> str:
|
|
expires_at = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
payload: dict[str, Any] = {
|
|
"sub": str(user.id),
|
|
"username": user.username,
|
|
"role": user.role,
|
|
"student_id": user.student_id,
|
|
"exp": expires_at,
|
|
}
|
|
return jwt.encode(payload, AUTH_SECRET_KEY, algorithm=AUTH_ALGORITHM)
|
|
|
|
|
|
def decode_access_token(token: str) -> dict[str, Any]:
|
|
try:
|
|
return jwt.decode(token, AUTH_SECRET_KEY, algorithms=[AUTH_ALGORITHM])
|
|
except jwt.PyJWTError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Session invalide ou expiree",
|
|
) from exc
|
|
|
|
|
|
def get_token_from_request(
|
|
request: Request,
|
|
session_cookie: str | None = Cookie(default=None, alias=AUTH_COOKIE_NAME),
|
|
) -> str | None:
|
|
if session_cookie:
|
|
return session_cookie
|
|
|
|
authorization = request.headers.get("Authorization", "")
|
|
scheme, _, value = authorization.partition(" ")
|
|
if scheme.lower() == "bearer" and value:
|
|
return value
|
|
return None
|
|
|
|
|
|
def get_current_user(
|
|
request: Request,
|
|
session_cookie: str | None = Cookie(default=None, alias=AUTH_COOKIE_NAME),
|
|
db: Session = Depends(get_db),
|
|
) -> models.User:
|
|
token = get_token_from_request(request, session_cookie)
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Authentification requise",
|
|
)
|
|
|
|
payload = decode_access_token(token)
|
|
user_id = payload.get("sub")
|
|
if not user_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Session invalide",
|
|
)
|
|
|
|
user = db.query(models.User).filter_by(id=int(user_id)).first()
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Utilisateur introuvable",
|
|
)
|
|
return user
|
|
|
|
|
|
def require_roles(*roles: str):
|
|
def dependency(current_user: models.User = Depends(get_current_user)) -> models.User:
|
|
if current_user.role not in roles:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Droits insuffisants",
|
|
)
|
|
return current_user
|
|
|
|
return dependency
|
|
|
|
|
|
def seed_admin_user(db: Session) -> None:
|
|
username = os.getenv("ADMIN_USERNAME")
|
|
password = os.getenv("ADMIN_PASSWORD")
|
|
if not username or not password:
|
|
return
|
|
|
|
existing = db.query(models.User).filter_by(username=username).first()
|
|
if existing:
|
|
if existing.role != "teacher":
|
|
existing.role = "teacher"
|
|
db.commit()
|
|
return
|
|
|
|
db.add(
|
|
models.User(
|
|
username=username,
|
|
password_hash=hash_password(password),
|
|
role="teacher",
|
|
)
|
|
)
|
|
db.commit()
|