254 lines
10 KiB
Python
254 lines
10 KiB
Python
"""Accounts-module: login/logout, wie-ben-ik, gebruikersbeheer en wachtwoordreset.
|
|
|
|
De login-endpoints horen hier thuis (de module is in de praktijk altijd aan).
|
|
"""
|
|
import logging
|
|
import re
|
|
import secrets
|
|
import unicodedata
|
|
|
|
from django.contrib.auth import authenticate, get_user_model
|
|
from django.contrib.auth import login as auth_login
|
|
from django.contrib.auth import logout as auth_logout
|
|
from django.contrib.auth.password_validation import validate_password
|
|
from django.contrib.auth.tokens import default_token_generator
|
|
from django.core.exceptions import ValidationError as DjangoValidationError
|
|
from django.core.mail import send_mail
|
|
from django.db.models import Q
|
|
from django.utils.encoding import force_bytes, force_str
|
|
from django.utils.http import urlsafe_base64_decode, urlsafe_base64_encode
|
|
from django.views.decorators.csrf import ensure_csrf_cookie
|
|
from rest_framework import status
|
|
from rest_framework.decorators import api_view, permission_classes
|
|
from rest_framework.permissions import AllowAny
|
|
from rest_framework.response import Response
|
|
|
|
from core.audit import AuditedModelViewSet, audit_log
|
|
from core.permissions import AdminRolePermission, module_permission_class
|
|
|
|
from .models import Gebruikersprofiel, Gebruikersrol
|
|
from .serializers import GebruikerSerializer
|
|
|
|
logger = logging.getLogger("roosterwijs.auth")
|
|
User = get_user_model()
|
|
AccountsModuleEnabled = module_permission_class("accounts")
|
|
|
|
|
|
def _password_errors(password, user):
|
|
try:
|
|
validate_password(password, user)
|
|
except DjangoValidationError as exc:
|
|
return list(exc.messages)
|
|
return None
|
|
|
|
|
|
def _user_payload(user):
|
|
profiel = getattr(user, "gebruikersprofiel", None)
|
|
persoon = getattr(profiel, "persoon", None) if profiel else None
|
|
return {
|
|
"authenticated": True,
|
|
"username": user.get_username(),
|
|
"is_staff": user.is_staff,
|
|
"role": profiel.rol if profiel else None,
|
|
"role_label": profiel.get_rol_display() if profiel else None,
|
|
"person": persoon.id if persoon else None,
|
|
"person_name": persoon.volledige_naam if persoon else None,
|
|
}
|
|
|
|
|
|
@api_view(["GET"])
|
|
@permission_classes([AllowAny])
|
|
@ensure_csrf_cookie
|
|
def auth_me(request):
|
|
"""Wie ben ik? Zet tegelijk de CSRF-cookie voor de frontend."""
|
|
if request.user.is_authenticated:
|
|
return Response(_user_payload(request.user))
|
|
return Response({"authenticated": False})
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AllowAny])
|
|
def auth_login_view(request):
|
|
"""Inloggen met gebruikersnaam of e-mailadres + wachtwoord (sessie-cookie)."""
|
|
naam = str(request.data.get("username", "")).strip()
|
|
wachtwoord = str(request.data.get("password", ""))
|
|
if not naam or not wachtwoord:
|
|
return Response({"detail": "Vul gebruikersnaam en wachtwoord in."}, status=status.HTTP_400_BAD_REQUEST)
|
|
# Sta ook inloggen met e-mailadres toe: zoek de bijbehorende gebruikersnaam op.
|
|
gebruikersnaam = naam
|
|
if "@" in naam:
|
|
match = User.objects.filter(email__iexact=naam).first()
|
|
if match:
|
|
gebruikersnaam = match.get_username()
|
|
user = authenticate(request, username=gebruikersnaam, password=wachtwoord)
|
|
if user is None:
|
|
logger.warning("Mislukte inlogpoging voor %r", naam)
|
|
return Response({"detail": "Onjuiste gebruikersnaam/e-mail of wachtwoord."}, status=status.HTTP_401_UNAUTHORIZED)
|
|
auth_login(request, user)
|
|
logger.info("Gebruiker %r ingelogd", user.get_username())
|
|
return Response(_user_payload(user))
|
|
|
|
|
|
@api_view(["POST"])
|
|
def auth_logout_view(request):
|
|
"""Uitloggen: sessie beëindigen."""
|
|
logger.info("Gebruiker %r uitgelogd", request.user.get_username())
|
|
auth_logout(request)
|
|
return Response({"authenticated": False})
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AllowAny])
|
|
def wachtwoord_vergeten(request):
|
|
"""Stuur (indien het account bestaat) een resetlink per e-mail."""
|
|
ident = str(request.data.get("identifier", "")).strip()
|
|
if ident:
|
|
user = User.objects.filter(Q(username__iexact=ident) | Q(email__iexact=ident)).first()
|
|
if user and user.email and user.is_active:
|
|
uid = urlsafe_base64_encode(force_bytes(user.pk))
|
|
token = default_token_generator.make_token(user)
|
|
from django.conf import settings
|
|
basis = (getattr(settings, "FRONTEND_BASE_URL", "") or request.build_absolute_uri("/")).rstrip("/")
|
|
link = f"{basis}/?reset=1&uid={uid}&token={token}"
|
|
send_mail(
|
|
"Wachtwoord opnieuw instellen — Roosterwijs",
|
|
f"Hallo {user.get_username()},\n\n"
|
|
f"Stel je wachtwoord opnieuw in via deze link:\n{link}\n\n"
|
|
f"Heb je dit niet aangevraagd? Dan kun je deze e-mail negeren.",
|
|
None, [user.email], fail_silently=True,
|
|
)
|
|
# Altijd hetzelfde antwoord: lek niet of een account bestaat.
|
|
return Response({"detail": "Als het account bestaat, is er een e-mail met instructies verstuurd."})
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AllowAny])
|
|
def wachtwoord_reset(request):
|
|
"""Zet een nieuw wachtwoord op basis van een geldige resetlink (uid + token)."""
|
|
uidb64 = request.data.get("uid")
|
|
token = request.data.get("token")
|
|
nieuw = str(request.data.get("wachtwoord", ""))
|
|
if len(nieuw) < 8:
|
|
return Response({"detail": "Het wachtwoord moet minstens 8 tekens zijn."}, status=400)
|
|
user = None
|
|
try:
|
|
user = User.objects.get(pk=force_str(urlsafe_base64_decode(uidb64)))
|
|
except Exception:
|
|
user = None
|
|
if not user or not default_token_generator.check_token(user, token):
|
|
return Response({"detail": "Deze resetlink is ongeldig of verlopen."}, status=400)
|
|
fouten = _password_errors(nieuw, user)
|
|
if fouten:
|
|
return Response({"detail": fouten}, status=400)
|
|
user.set_password(nieuw)
|
|
user.save()
|
|
logger.info("Wachtwoord gereset voor %r", user.get_username())
|
|
audit_log(request, "password_reset", f"User:{user.pk}")
|
|
return Response({"detail": "Je wachtwoord is aangepast. Je kunt nu inloggen."})
|
|
|
|
|
|
class GebruikerViewSet(AuditedModelViewSet):
|
|
"""Gebruikersbeheer — alleen voor beheerders (is_staff)."""
|
|
|
|
serializer_class = GebruikerSerializer
|
|
permission_classes = [AdminRolePermission, AccountsModuleEnabled]
|
|
|
|
def get_queryset(self):
|
|
return User.objects.select_related("gebruikersprofiel", "gebruikersprofiel__persoon").order_by("username")
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AdminRolePermission, AccountsModuleEnabled])
|
|
def wachtwoord_zet(request, pk):
|
|
"""Beheerder zet/genereert een (tijdelijk) wachtwoord; geeft het terug."""
|
|
try:
|
|
user = User.objects.get(pk=pk)
|
|
except User.DoesNotExist:
|
|
return Response({"detail": "Gebruiker niet gevonden."}, status=404)
|
|
nieuw = str(request.data.get("wachtwoord", "")).strip() or secrets.token_urlsafe(9)
|
|
if len(nieuw) < 8:
|
|
return Response({"detail": "Het wachtwoord moet minstens 8 tekens zijn."}, status=400)
|
|
fouten = _password_errors(nieuw, user)
|
|
if fouten:
|
|
return Response({"detail": fouten}, status=400)
|
|
user.set_password(nieuw)
|
|
user.save()
|
|
logger.info("Beheerder %r zette nieuw wachtwoord voor %r", request.user.get_username(), user.get_username())
|
|
audit_log(request, "password_set", f"User:{user.pk}")
|
|
return Response({"wachtwoord": nieuw, "gebruiker": user.get_username()})
|
|
|
|
|
|
_WW_ALFABET = "abcdefghjkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789"
|
|
|
|
|
|
def _genereer_wachtwoord(lengte: int = 10) -> str:
|
|
return "".join(secrets.choice(_WW_ALFABET) for _ in range(lengte))
|
|
|
|
|
|
def _slug_gebruikersnaam(voornaam: str, achternaam: str) -> str:
|
|
basis = f"{voornaam}.{achternaam}".strip(".").lower()
|
|
basis = unicodedata.normalize("NFKD", basis).encode("ascii", "ignore").decode()
|
|
basis = re.sub(r"[^a-z0-9.]+", "", basis).strip(".")
|
|
return basis or "gebruiker"
|
|
|
|
|
|
def _uniek(basis: str, gebruikt: set) -> str:
|
|
naam = basis
|
|
i = 1
|
|
while naam in gebruikt or User.objects.filter(username__iexact=naam).exists():
|
|
i += 1
|
|
naam = f"{basis}{i}"
|
|
gebruikt.add(naam)
|
|
return naam
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AdminRolePermission, AccountsModuleEnabled])
|
|
def gebruikers_import(request):
|
|
"""Bulk: maak meerdere gebruikers met automatisch gegenereerd wachtwoord.
|
|
|
|
Per rij optioneel een bestaande persoon koppelen (persoon-id) of een nieuwe
|
|
persoon aanmaken (maak_persoon). Geeft de gegenereerde wachtwoorden terug.
|
|
"""
|
|
from core.models import Persoon, Rol
|
|
|
|
rijen = request.data.get("gebruikers") or []
|
|
if not isinstance(rijen, list):
|
|
return Response({"detail": "gebruikers moet een lijst zijn."}, status=400)
|
|
geldige_rollen = {r.value for r in Gebruikersrol}
|
|
resultaat = []
|
|
gebruikt = set()
|
|
for r in rijen:
|
|
vn = str(r.get("voornaam") or "").strip()
|
|
an = str(r.get("achternaam") or "").strip()
|
|
if not vn and not an:
|
|
continue
|
|
email = str(r.get("email") or "").strip()
|
|
rol = r.get("rol") if r.get("rol") in geldige_rollen else Gebruikersrol.MEDEWERKER
|
|
gn = str(r.get("gebruikersnaam") or "").strip() or _slug_gebruikersnaam(vn, an)
|
|
gn = _uniek(gn, gebruikt)
|
|
wachtwoord = _genereer_wachtwoord()
|
|
user = User(username=gn, email=email, first_name=vn, last_name=an)
|
|
fouten = _password_errors(wachtwoord, user)
|
|
if fouten:
|
|
return Response({"detail": fouten, "gebruiker": gn}, status=400)
|
|
user.set_password(wachtwoord)
|
|
user.save()
|
|
persoon = None
|
|
pid = r.get("persoon")
|
|
if pid:
|
|
persoon = Persoon.objects.filter(pk=pid).first()
|
|
elif r.get("maak_persoon"):
|
|
prol = Rol.LEERLING if rol == Gebruikersrol.LEERLING else Rol.LEERKRACHT
|
|
persoon = Persoon.objects.create(voornaam=vn, achternaam=an, rol=prol)
|
|
Gebruikersprofiel.objects.create(user=user, rol=rol, persoon=persoon)
|
|
resultaat.append({
|
|
"gebruikersnaam": gn,
|
|
"wachtwoord": wachtwoord,
|
|
"naam": f"{vn} {an}".strip(),
|
|
"email": email,
|
|
"rol": rol,
|
|
"persoon_naam": persoon.volledige_naam if persoon else None,
|
|
})
|
|
audit_log(request, "import", "User", {"aangemaakt": len(resultaat)})
|
|
return Response({"aangemaakt": len(resultaat), "gebruikers": resultaat})
|