152 lines
5.1 KiB
Python
152 lines
5.1 KiB
Python
"""API d'administration pour l'interface React (authentification Django par session)."""
|
|
import json
|
|
|
|
from django.contrib.auth import authenticate, get_user_model, login, logout
|
|
from django.contrib.auth.password_validation import validate_password
|
|
from django.core.exceptions import ValidationError
|
|
from django.db.models import Q
|
|
from django.http import JsonResponse
|
|
from django.middleware.csrf import get_token
|
|
from django.views.decorators.csrf import ensure_csrf_cookie
|
|
from django.views.decorators.http import require_GET, require_POST
|
|
|
|
|
|
def _error(message, status=400, **extra):
|
|
return JsonResponse({"detail": message, **extra}, status=status)
|
|
|
|
|
|
def _json_body(request):
|
|
try:
|
|
payload = json.loads(request.body or "{}")
|
|
except (ValueError, UnicodeDecodeError):
|
|
return None
|
|
return payload if isinstance(payload, dict) else None
|
|
|
|
|
|
def _admin_only(request, superuser=False):
|
|
if not request.user.is_authenticated:
|
|
return _error("Connexion requise.", 401)
|
|
if not request.user.is_active or not request.user.is_staff:
|
|
return _error("Accès administrateur requis.", 403)
|
|
if superuser and not request.user.is_superuser:
|
|
return _error("Action réservée aux superutilisateurs.", 403)
|
|
return None
|
|
|
|
|
|
def _user_data(user):
|
|
return {
|
|
"id": user.pk,
|
|
"username": user.get_username(),
|
|
"first_name": user.first_name,
|
|
"last_name": user.last_name,
|
|
"email": user.email,
|
|
"is_active": user.is_active,
|
|
"is_staff": user.is_staff,
|
|
"is_superuser": user.is_superuser,
|
|
"last_login": user.last_login.isoformat() if user.last_login else None,
|
|
}
|
|
|
|
|
|
@require_GET
|
|
@ensure_csrf_cookie
|
|
def csrf(request):
|
|
# Prépare un cookie CSRF avant la première requête POST (connexion incluse).
|
|
get_token(request)
|
|
return JsonResponse({"detail": "CSRF prêt."})
|
|
|
|
|
|
@require_GET
|
|
def me(request):
|
|
if not request.user.is_authenticated:
|
|
return _error("Connexion requise.", 401)
|
|
return JsonResponse({"user": _user_data(request.user)})
|
|
|
|
|
|
@require_POST
|
|
def sign_in(request):
|
|
data = _json_body(request)
|
|
if data is None:
|
|
return _error("Données JSON invalides.")
|
|
username = str(data.get("username", "")).strip()
|
|
password = data.get("password", "")
|
|
if not isinstance(password, str):
|
|
return _error("Mot de passe invalide.")
|
|
user = authenticate(request, username=username, password=password)
|
|
if user is None:
|
|
return _error("Identifiants incorrects.", 401)
|
|
if not user.is_staff:
|
|
return _error("Accès administrateur requis.", 403)
|
|
login(request, user)
|
|
return JsonResponse({"user": _user_data(user)})
|
|
|
|
|
|
@require_POST
|
|
def sign_out(request):
|
|
logout(request)
|
|
return JsonResponse({"detail": "Déconnexion effectuée."})
|
|
|
|
|
|
@require_GET
|
|
def dashboard(request):
|
|
denied = _admin_only(request)
|
|
if denied:
|
|
return denied
|
|
User = get_user_model()
|
|
users = User.objects.all()
|
|
return JsonResponse({
|
|
"stats": {
|
|
"total": users.count(),
|
|
"active": users.filter(is_active=True).count(),
|
|
"staff": users.filter(is_staff=True).count(),
|
|
},
|
|
"users": [_user_data(user) for user in users.order_by("-date_joined", "-pk")[:200]],
|
|
})
|
|
|
|
|
|
@require_POST
|
|
def create_user(request):
|
|
denied = _admin_only(request, superuser=True)
|
|
if denied:
|
|
return denied
|
|
data = _json_body(request)
|
|
if data is None:
|
|
return _error("Données JSON invalides.")
|
|
username = str(data.get("username", "")).strip()
|
|
password = data.get("password")
|
|
email = str(data.get("email", "")).strip()
|
|
if not username or len(username) > 150:
|
|
return _error("Nom d'utilisateur invalide.")
|
|
if not isinstance(password, str) or not password:
|
|
return _error("Mot de passe requis.")
|
|
User = get_user_model()
|
|
if User.objects.filter(username__iexact=username).exists():
|
|
return _error("Ce nom d'utilisateur existe déjà.", 409)
|
|
temp_user = User(username=username, email=email)
|
|
try:
|
|
temp_user.full_clean(exclude=["password"], validate_unique=False)
|
|
validate_password(password, user=temp_user)
|
|
except ValidationError as exc:
|
|
return _error(" ".join(exc.messages))
|
|
user = User.objects.create_user(username=username, password=password, email=email)
|
|
return JsonResponse({"user": _user_data(user)}, status=201)
|
|
|
|
|
|
@require_POST
|
|
def set_user_active(request, user_id):
|
|
denied = _admin_only(request, superuser=True)
|
|
if denied:
|
|
return denied
|
|
data = _json_body(request)
|
|
if data is None or type(data.get("is_active")) is not bool:
|
|
return _error("La propriété is_active doit être un booléen.")
|
|
User = get_user_model()
|
|
try:
|
|
target = User.objects.get(pk=user_id)
|
|
except User.DoesNotExist:
|
|
return _error("Utilisateur introuvable.", 404)
|
|
if target.pk == request.user.pk or target.is_superuser:
|
|
return _error("Impossible de modifier ce compte.", 403)
|
|
target.is_active = data["is_active"]
|
|
target.save(update_fields=["is_active"])
|
|
return JsonResponse({"user": _user_data(target)})
|