from __future__ import annotations from datetime import datetime from django.conf import settings from django.contrib import messages from django.contrib.auth import authenticate, get_user_model, login, update_session_auth_hash from django.contrib.auth.decorators import login_required from django.contrib.auth.views import LogoutView, PasswordChangeView, PasswordResetCompleteView, PasswordResetConfirmView, PasswordResetDoneView, PasswordResetView from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse_lazy from django.utils import timezone from django.views.decorators.http import require_http_methods, require_POST from .decorators import super_admin_required from .forms import ( AdminUserUpdateForm, OTPVerificationForm, RegistrationForm, SignInForm, StyledPasswordChangeForm, StyledPasswordResetForm, StyledSetPasswordForm, ) from .services import clear_pending_otp_session, issue_login_otp, latest_active_otp, send_login_otp_email, set_session_expiry from .utils import ( clear_failed_attempts, get_client_ip, get_failed_attempt_count, is_rate_limited, register_failed_attempt, should_enable_otp, ) User = get_user_model() def _managed_users_queryset(): return User.objects.order_by("-created_at", "email") @require_http_methods(["GET"]) def home_redirect(request): if request.user.is_authenticated: return redirect("accounts:dashboard") return redirect("accounts:signin") @super_admin_required @require_http_methods(["GET", "POST"]) def register_view(request): form = RegistrationForm(request.POST or None) if request.method == "POST" and form.is_valid(): user = form.save() messages.success(request, f"User account '{user.email}' created successfully.") return redirect("accounts:register") return render( request, "accounts/register.html", { "form": form, "page_title": "Users", "managed_user": None, "users": _managed_users_queryset(), "submit_label": "Create User", }, ) @super_admin_required @require_http_methods(["GET", "POST"]) def user_edit_view(request, pk: int): managed_user = get_object_or_404(_managed_users_queryset(), pk=pk) form = AdminUserUpdateForm(request.POST or None, instance=managed_user) if request.method == "POST" and form.is_valid(): if managed_user.pk == request.user.pk and not form.cleaned_data["is_active"]: form.add_error("is_active", "You cannot deactivate your own account from this page.") else: updated_user = form.save() if updated_user.pk == request.user.pk and form.cleaned_data.get("password1"): update_session_auth_hash(request, updated_user) messages.success(request, f"User account '{updated_user.email}' updated successfully.") return redirect("accounts:user_edit", pk=updated_user.pk) return render( request, "accounts/register.html", { "form": form, "page_title": "Users", "managed_user": managed_user, "users": _managed_users_queryset(), "submit_label": "Save Changes", }, ) @require_http_methods(["GET", "POST"]) def signin_view(request): if request.user.is_authenticated: return redirect("accounts:dashboard") form = SignInForm(request.POST or None) if request.method == "POST": ip_address = get_client_ip(request) raw_identifier = request.POST.get("identifier", "").strip().lower() or "unknown" if is_rate_limited( scope="login", identifier=raw_identifier, ip_address=ip_address, max_attempts=settings.LOGIN_RATE_LIMIT_ATTEMPTS, window_seconds=settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS, ): failed_attempt_count = get_failed_attempt_count( scope="login", identifier=raw_identifier, ip_address=ip_address, window_seconds=settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS, ) form.add_error( None, f"Too many failed attempts ({failed_attempt_count}/{settings.LOGIN_RATE_LIMIT_ATTEMPTS}). Try again in a few minutes.", ) elif form.is_valid(): identifier = form.cleaned_data["identifier"].strip() password = form.cleaned_data["password"] remember_me = form.cleaned_data["remember_me"] user = authenticate(request, username=identifier, password=password) if user is None: failed_attempt_count = register_failed_attempt( scope="login", identifier=raw_identifier, ip_address=ip_address, window_seconds=settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS, ) form.add_error( None, f"Invalid email or password. Failed attempts: {failed_attempt_count}/{settings.LOGIN_RATE_LIMIT_ATTEMPTS}.", ) else: clear_failed_attempts(scope="login", identifier=raw_identifier, ip_address=ip_address) if should_enable_otp(): otp_record = issue_login_otp(user) send_login_otp_email( user=user, otp_code=otp_record.raw_code, expires_minutes=settings.OTP_EXPIRY_MINUTES, ) request.session["pending_otp_user_id"] = user.pk request.session["pending_otp_remember"] = remember_me request.session["pending_otp_last_sent"] = timezone.now().isoformat() messages.info(request, "OTP sent to your registered email address.") return redirect("accounts:verify_otp") login(request, user) set_session_expiry(request, remember_me) messages.success(request, "Welcome back.") return redirect("accounts:dashboard") return render(request, "accounts/signin.html", {"form": form}) def _masked_email(email: str) -> str: if "@" not in email: return email username, domain = email.split("@", 1) if len(username) <= 2: obfuscated = username[0] + "*" if username else "*" else: obfuscated = username[0] + ("*" * (len(username) - 2)) + username[-1] return f"{obfuscated}@{domain}" @require_http_methods(["GET", "POST"]) def verify_otp_view(request): pending_user_id = request.session.get("pending_otp_user_id") if not pending_user_id: messages.info(request, "Your login session expired. Please sign in again.") return redirect("accounts:signin") try: user = User.objects.get(pk=pending_user_id, is_active=True) except User.DoesNotExist: clear_pending_otp_session(request) messages.error(request, "Unable to continue OTP verification. Please sign in again.") return redirect("accounts:signin") form = OTPVerificationForm(request.POST or None) if request.method == "POST": ip_address = get_client_ip(request) scope_identifier = f"user-{user.pk}" if is_rate_limited( scope="otp-verify", identifier=scope_identifier, ip_address=ip_address, max_attempts=settings.OTP_RATE_LIMIT_ATTEMPTS, window_seconds=settings.OTP_RATE_LIMIT_WINDOW_SECONDS, ): form.add_error(None, "Too many invalid OTP attempts. Please sign in again.") elif form.is_valid(): otp_code = form.cleaned_data["otp_code"] otp_record = latest_active_otp(user) if otp_record and otp_record.validate(otp_code): otp_record.mark_used() clear_failed_attempts(scope="otp-verify", identifier=scope_identifier, ip_address=ip_address) login(request, user) remember_me = request.session.get("pending_otp_remember", False) set_session_expiry(request, remember_me) clear_pending_otp_session(request) messages.success(request, "OTP verified. You are signed in.") return redirect("accounts:dashboard") register_failed_attempt( scope="otp-verify", identifier=scope_identifier, ip_address=ip_address, window_seconds=settings.OTP_RATE_LIMIT_WINDOW_SECONDS, ) if otp_record: otp_record.attempts += 1 otp_record.save(update_fields=["attempts", "updated_at"]) form.add_error("otp_code", "Invalid or expired OTP code.") now = timezone.now() otp_record = latest_active_otp(user) otp_seconds_remaining = settings.OTP_EXPIRY_MINUTES * 60 if otp_record: otp_seconds_remaining = max(0, int((otp_record.expires_at - now).total_seconds())) resend_seconds_remaining = 0 last_sent_raw = request.session.get("pending_otp_last_sent") if last_sent_raw: try: last_sent = datetime.fromisoformat(last_sent_raw) if timezone.is_naive(last_sent): last_sent = timezone.make_aware(last_sent) elapsed = int((now - last_sent).total_seconds()) resend_seconds_remaining = max(0, settings.OTP_RESEND_COOLDOWN_SECONDS - elapsed) except ValueError: resend_seconds_remaining = 0 context = { "form": form, "masked_email": _masked_email(user.email), "otp_seconds_remaining": otp_seconds_remaining, "resend_seconds_remaining": resend_seconds_remaining, } return render(request, "accounts/verify_otp.html", context) @require_POST def resend_otp_view(request): pending_user_id = request.session.get("pending_otp_user_id") if not pending_user_id: messages.info(request, "Your login session expired. Please sign in again.") return redirect("accounts:signin") try: user = User.objects.get(pk=pending_user_id, is_active=True) except User.DoesNotExist: clear_pending_otp_session(request) messages.error(request, "Unable to resend OTP. Please sign in again.") return redirect("accounts:signin") ip_address = get_client_ip(request) scope_identifier = f"user-{user.pk}" if is_rate_limited( scope="otp-resend", identifier=scope_identifier, ip_address=ip_address, max_attempts=settings.OTP_RESEND_RATE_LIMIT_ATTEMPTS, window_seconds=settings.OTP_RESEND_RATE_LIMIT_WINDOW_SECONDS, ): messages.error(request, "Too many OTP resend requests. Try again later.") return redirect("accounts:verify_otp") now = timezone.now() last_sent_raw = request.session.get("pending_otp_last_sent") if last_sent_raw: try: last_sent = datetime.fromisoformat(last_sent_raw) if timezone.is_naive(last_sent): last_sent = timezone.make_aware(last_sent) elapsed = (now - last_sent).total_seconds() cooldown = settings.OTP_RESEND_COOLDOWN_SECONDS if elapsed < cooldown: remaining = int(cooldown - elapsed) messages.warning(request, f"Please wait {remaining}s before requesting another OTP.") return redirect("accounts:verify_otp") except ValueError: pass register_failed_attempt( scope="otp-resend", identifier=scope_identifier, ip_address=ip_address, window_seconds=settings.OTP_RESEND_RATE_LIMIT_WINDOW_SECONDS, ) otp_record = issue_login_otp(user) send_login_otp_email(user=user, otp_code=otp_record.raw_code, expires_minutes=settings.OTP_EXPIRY_MINUTES) request.session["pending_otp_last_sent"] = now.isoformat() messages.success(request, "A new OTP code has been sent to your email.") return redirect("accounts:verify_otp") class CustomLogoutView(LogoutView): next_page = reverse_lazy("accounts:signin") class CustomPasswordChangeView(PasswordChangeView): form_class = StyledPasswordChangeForm template_name = "accounts/change_password.html" success_url = reverse_lazy("accounts:dashboard") def form_valid(self, form): response = super().form_valid(form) messages.success(self.request, "Password updated successfully.") return response class CustomPasswordResetView(PasswordResetView): form_class = StyledPasswordResetForm template_name = "accounts/forgot_password.html" email_template_name = "accounts/password_reset_email.txt" subject_template_name = "accounts/password_reset_subject.txt" success_url = reverse_lazy("accounts:password_reset_done") def form_valid(self, form): messages.info( self.request, "If an account exists with that email, a password reset link has been sent.", ) return super().form_valid(form) class CustomPasswordResetDoneView(PasswordResetDoneView): template_name = "accounts/password_reset_done.html" class CustomPasswordResetConfirmView(PasswordResetConfirmView): form_class = StyledSetPasswordForm template_name = "accounts/reset_password.html" success_url = reverse_lazy("accounts:password_reset_complete") class CustomPasswordResetCompleteView(PasswordResetCompleteView): template_name = "accounts/password_reset_complete.html" def custom_404_view(request, exception): return render(request, "accounts/pages-error404.html", status=404)