feat: Implement authentication and authorization with user management and admin-only debug endpoints.

This commit is contained in:
Michael committed 2025-11-24 07:40:22 +01:00
1 parent 25b0a824c2
commit 5f4d3afbae
3 files changed
+73 -54

No files matched your search

+69
View File
@@ -0,0 +1,69 @@
"""
Shared dependencies for the application.
"""
from typing import Annotated
from fastapi import Depends, Header, HTTPException, status
from sqlalchemy.orm import Session
from app import models
from app.database import SessionLocal
from app.services import auth_service
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_current_user(
authorization: Annotated[str | None, Header()] = None,
db: Session = Depends(get_db),
) -> models.User:
"""Dependency to get current authenticated user from JWT token."""
if not authorization:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Non authentifié",
)
# Extract token from "Bearer <token>" format
parts = authorization.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Format de token invalide",
)
token = parts[1]
payload = auth_service.decode_token(token)
if not payload:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token invalide ou expiré",
)
user = auth_service.get_user_by_id(db, payload["user_id"])
if not user or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Utilisateur non trouvé ou désactivé",
)
return user
def get_admin_user(
current_user: models.User = Depends(get_current_user),
) -> models.User:
"""Dependency to require admin privileges."""
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Accès administrateur requis",
)
return current_user
+2 -53
View File
@@ -11,18 +11,14 @@ from sqlalchemy.orm import Session
from app import models
from app.database import SessionLocal
from app.services import auth_service
from app.dependencies import get_db, get_current_user, get_admin_user
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/auth", tags=["auth"])
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Pydantic models
@@ -64,54 +60,7 @@ class AdminChangePasswordRequest(BaseModel):
new_password: str
def get_current_user(
authorization: Annotated[str | None, Header()] = None,
db: Session = Depends(get_db),
) -> models.User:
"""Dependency to get current authenticated user from JWT token."""
if not authorization:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Non authentifié",
)
# Extract token from "Bearer <token>" format
parts = authorization.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Format de token invalide",
)
token = parts[1]
payload = auth_service.decode_token(token)
if not payload:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token invalide ou expiré",
)
user = auth_service.get_user_by_id(db, payload["user_id"])
if not user or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Utilisateur non trouvé ou désactivé",
)
return user
def get_admin_user(
current_user: models.User = Depends(get_current_user),
) -> models.User:
"""Dependency to require admin privileges."""
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Accès administrateur requis",
)
return current_user
@router.post("/login", response_model=LoginResponse)
+2 -1
View File
@@ -8,11 +8,12 @@ from fastapi.responses import FileResponse
from pydantic import BaseModel
from app.services import auth_service
from app.dependencies import get_admin_user
router = APIRouter(
prefix="/debug",
tags=["debug"],
dependencies=[Depends(auth_service.get_current_active_admin)],
dependencies=[Depends(get_admin_user)],
)
logger = logging.getLogger(__name__)