import io from urllib.parse import urlencode import segno from django.conf import settings from django.contrib import messages from django.contrib.auth import get_user_model, logout from django.contrib.auth.decorators import login_required from django.contrib.auth.views import ( LoginView, PasswordResetCompleteView, PasswordResetConfirmView, PasswordResetDoneView, PasswordResetView, ) from django.http import Http404 from django.shortcuts import redirect, render from django.urls import reverse, reverse_lazy from django.utils import timezone from django.views.decorators.http import require_POST from django_otp.plugins.otp_totp.models import TOTPDevice from .forms import AccountForm, DisableTwoFactorForm, OTPTokenForm, RegenerateRecoveryCodesForm, UnifiedAuthenticationForm from .models import RecoveryCode, SecurityEvent from .services import ( SESSION_2FA_VERIFIED, SESSION_PREAUTH_BACKEND, SESSION_PREAUTH_NEXT, SESSION_PREAUTH_USER, complete_login, confirmed_device, consume_recovery_code, generate_recovery_codes, get_profile, log_event, manual_totp_secret, pending_device, remember_preauth, safe_next, two_factor_enabled, two_factor_required, user_destination, ) User = get_user_model() class K4WPasswordResetView(PasswordResetView): template_name = "security/password_reset_form.html" email_template_name = "security/password_reset_email.txt" html_email_template_name = "security/password_reset_email.html" subject_template_name = "security/password_reset_subject.txt" success_url = reverse_lazy("security:password_reset_done") def form_valid(self, form): email = (form.cleaned_data.get("email") or "").strip() log_event( self.request, SecurityEvent.EventType.PASSWORD_RESET_REQUESTED, username_entered=email, details={"delivery": "requested"}, ) return super().form_valid(form) class K4WPasswordResetDoneView(PasswordResetDoneView): template_name = "security/password_reset_done.html" class K4WPasswordResetConfirmView(PasswordResetConfirmView): template_name = "security/password_reset_confirm.html" success_url = reverse_lazy("security:password_reset_complete") def form_valid(self, form): response = super().form_valid(form) if getattr(self, "user", None): log_event( self.request, SecurityEvent.EventType.PASSWORD_RESET_COMPLETED, user=self.user, ) return response class K4WPasswordResetCompleteView(PasswordResetCompleteView): template_name = "security/password_reset_complete.html" class UnifiedLoginView(LoginView): template_name = "security/login.html" authentication_form = UnifiedAuthenticationForm redirect_authenticated_user = False def dispatch(self, request, *args, **kwargs): if request.user.is_authenticated: return redirect(user_destination(request.user)) return super().dispatch(request, *args, **kwargs) def form_invalid(self, form): username = self.request.POST.get("username", "") log_event( self.request, SecurityEvent.EventType.LOGIN_FAILED, username_entered=username, details={"reason": "invalid_credentials"}, ) return super().form_invalid(form) def form_valid(self, form): user = form.get_user() destination = safe_next(self.request, user_destination(user)) device = confirmed_device(user) if device: remember_preauth(self.request, user, destination) return redirect("security:verify") backend = getattr(user, "backend", settings.AUTHENTICATION_BACKENDS[0]) complete_login(self.request, user, backend=backend, next_url=destination) self.request.session[SESSION_2FA_VERIFIED] = not two_factor_required(user) log_event(self.request, SecurityEvent.EventType.LOGIN_SUCCESS, user=user) if two_factor_required(user): messages.warning(self.request, "A política de segurança exige que configure o segundo fator antes de continuar.") return redirect("security:setup") return redirect(destination) @require_POST def unified_logout(request): logout(request) return redirect("security:login") def _preauth_user(request): user_id = request.session.get(SESSION_PREAUTH_USER) if user_id: return User.objects.filter(pk=user_id, is_active=True).first() if request.user.is_authenticated: return request.user return None def verify(request): user = _preauth_user(request) if not user: messages.info(request, "Inicie sessão para continuar.") return redirect("security:login") device = confirmed_device(user) if not device: return redirect("security:setup") form = OTPTokenForm(request.POST or None) if request.method == "POST" and form.is_valid(): token = form.cleaned_data["token"].replace(" ", "").strip() valid = token.isdigit() and device.verify_token(token) used_recovery = False if not valid: used_recovery = consume_recovery_code(user, token) valid = used_recovery if valid: next_url = request.session.get(SESSION_PREAUTH_NEXT) or safe_next(request, user_destination(user)) backend = request.session.get(SESSION_PREAUTH_BACKEND) or settings.AUTHENTICATION_BACKENDS[0] if not request.user.is_authenticated: complete_login(request, user, backend=backend, next_url=next_url) else: request.session[SESSION_2FA_VERIFIED] = True profile = get_profile(user) profile.last_two_factor_at = timezone.now() profile.save(update_fields=["last_two_factor_at", "updated_at"]) log_event( request, SecurityEvent.EventType.RECOVERY_USED if used_recovery else SecurityEvent.EventType.OTP_SUCCESS, user=user, ) return redirect(next_url) log_event(request, SecurityEvent.EventType.OTP_FAILED, user=user) form.add_error("token", "O código não é válido ou já foi utilizado.") return render( request, "security/verify.html", { "form": form, "target_user": user, "remaining_recovery_codes": RecoveryCode.objects.filter(user=user, used_at__isnull=True).count(), }, ) @login_required def overview(request): device = confirmed_device(request.user) profile = get_profile(request.user) return render( request, "security/overview.html", { "device": device, "profile": profile, "recovery_remaining": RecoveryCode.objects.filter(user=request.user, used_at__isnull=True).count(), "staff_enforced": bool(request.user.is_staff and settings.K4W_REQUIRE_2FA_FOR_STAFF), "recent_events": request.user.k4w_security_events.all()[:8], }, ) @login_required def setup(request): existing = confirmed_device(request.user) if existing: messages.info(request, "A autenticação de dois fatores já está ativa nesta conta.") return redirect("security:overview") device = pending_device(request.user) if not device: device = TOTPDevice.objects.create(user=request.user, name="Kreate4Web", confirmed=False) form = OTPTokenForm(request.POST or None) recovery_codes = None if request.method == "POST" and form.is_valid(): token = form.cleaned_data["token"].replace(" ", "").strip() if token.isdigit() and device.verify_token(token): device.confirmed = True device.save(update_fields=["confirmed"]) recovery_codes = generate_recovery_codes(request.user) request.session[SESSION_2FA_VERIFIED] = True profile = get_profile(request.user) profile.last_two_factor_at = timezone.now() profile.save(update_fields=["last_two_factor_at", "updated_at"]) log_event(request, SecurityEvent.EventType.OTP_ENABLED, user=request.user) return render( request, "security/recovery_codes.html", {"recovery_codes": recovery_codes, "first_setup": True}, ) form.add_error("token", "O código não foi aceite. Confirme a hora do telemóvel e tente novamente.") output = io.BytesIO() segno.make(device.config_url, error="m").save(output, kind="svg", scale=5, border=1) qr_svg = output.getvalue().decode("utf-8") return render( request, "security/setup.html", { "form": form, "qr_svg": qr_svg, "manual_secret": manual_totp_secret(device), }, ) @login_required @require_POST def disable(request): if request.user.is_staff and settings.K4W_REQUIRE_2FA_FOR_STAFF: messages.error(request, "A política de segurança do backoffice não permite desativar o segundo fator.") return redirect("security:overview") form = DisableTwoFactorForm(request.user, request.POST) device = confirmed_device(request.user) if not device: return redirect("security:overview") if form.is_valid(): token = form.cleaned_data["token"].replace(" ", "").strip() valid = token.isdigit() and device.verify_token(token) if not valid: valid = consume_recovery_code(request.user, token) if valid: TOTPDevice.objects.filter(user=request.user).delete() RecoveryCode.objects.filter(user=request.user).delete() request.session[SESSION_2FA_VERIFIED] = True log_event(request, SecurityEvent.EventType.OTP_DISABLED, user=request.user) messages.success(request, "A autenticação de dois fatores foi desativada.") else: messages.error(request, "O código indicado não é válido.") else: messages.error(request, "Não foi possível confirmar a desativação.") return redirect("security:overview") @login_required def recovery_codes(request): device = confirmed_device(request.user) if not device: return redirect("security:setup") form = RegenerateRecoveryCodesForm(request.user, request.POST or None) if request.method == "POST" and form.is_valid(): token = form.cleaned_data["token"].replace(" ", "").strip() if token.isdigit() and device.verify_token(token): codes = generate_recovery_codes(request.user) log_event(request, SecurityEvent.EventType.RECOVERY_REGENERATED, user=request.user) return render(request, "security/recovery_codes.html", {"recovery_codes": codes, "first_setup": False}) form.add_error("token", "O código da aplicação autenticadora não é válido.") return render( request, "security/regenerate_codes.html", { "form": form, "remaining": RecoveryCode.objects.filter(user=request.user, used_at__isnull=True).count(), }, ) @login_required def account(request): profile = get_profile(request.user) form = AccountForm(request.POST or None, request.FILES or None, instance=profile, user=request.user) if request.method == "POST" and form.is_valid(): form.save() log_event(request, SecurityEvent.EventType.PROFILE_UPDATED, user=request.user) messages.success(request, "A sua conta foi atualizada.") return redirect("security:account") return render( request, "security/account.html", { "form": form, "profile": profile, "two_factor_enabled": two_factor_enabled(request.user), }, )