From 764f840d4d508725d741f115ba3732206691f2ee Mon Sep 17 00:00:00 2001 From: Ali-Nawaz-devt Date: Sun, 25 Oct 2026 01:24:49 +0500 Subject: [PATCH] feat(accounts): implement Phase 3 email change workflow and OTP verification --- accounts/forms.py | 66 +++ accounts/services/email_change_service.py | 12 + accounts/views.py | 115 +++++ static/css/settings.css | 71 +++ static/js/settings.js | 101 ++++ templates/accounts/settings.html | 141 +++++- .../test_account_settings_email_change.py | 442 ++++++++++++++++++ 7 files changed, 925 insertions(+), 23 deletions(-) create mode 100644 tests/unit/test_account_settings_email_change.py diff --git a/accounts/forms.py b/accounts/forms.py index cc7bf39..1d467c5 100644 --- a/accounts/forms.py +++ b/accounts/forms.py @@ -299,3 +299,69 @@ def clean_new_username(self): raw = self.cleaned_data.get("new_username", "").strip() return validate_username_format(raw, current_user=self.user) + + +class EmailChangeRequestForm(forms.Form): + """Form to initiate email address change with password confirmation.""" + + new_email = forms.EmailField( + required=True, + widget=forms.EmailInput( + attrs={ + "class": "form-input", + "placeholder": "new.email@example.com", + "autocomplete": "email", + "id": "settings-new-email", + } + ), + ) + current_password = forms.CharField( + required=True, + widget=forms.PasswordInput( + attrs={ + "class": "form-input", + "placeholder": "Enter current password to authorize", + "autocomplete": "current-password", + "id": "settings-email-password", + } + ), + ) + + def __init__(self, user, *args, **kwargs): + self.user = user + super().__init__(*args, **kwargs) + + def clean_new_email(self): + raw = self.cleaned_data.get("new_email", "").strip().lower() + current_email = (self.user.email or "").strip().lower() + if raw == current_email: + raise ValidationError("The new email address cannot be the same as your current address.") + return raw + + +class EmailChangeConfirmForm(forms.Form): + """Form to confirm 6-digit OTP code sent to the new email address.""" + + code = forms.CharField( + max_length=6, + min_length=6, + required=True, + widget=forms.TextInput( + attrs={ + "class": "form-input font-mono-data", + "placeholder": "123456", + "id": "email-otp-combined-input", + "autocomplete": "one-time-code", + "inputmode": "numeric", + "pattern": "[0-9]{6}", + "maxlength": "6", + "style": "letter-spacing: 0.25em; text-align: center; font-size: 1.15rem; max-width: 200px;", + } + ), + ) + + def clean_code(self): + raw = self.cleaned_data.get("code", "").strip() + if not raw.isdigit() or len(raw) != 6: + raise ValidationError("Please enter a valid 6-digit numerical code.") + return raw diff --git a/accounts/services/email_change_service.py b/accounts/services/email_change_service.py index 7c9f805..f997b61 100644 --- a/accounts/services/email_change_service.py +++ b/accounts/services/email_change_service.py @@ -212,3 +212,15 @@ def confirm_email_change( ) return True, "Your email address has been successfully updated and verified." + + +def cancel_email_change(user, request=None) -> None: + """ + Cancels any active pending email change request and invalidates unused change_email codes. + """ + PendingEmailChange.objects.filter(user=user).delete() + EmailVerificationCode.objects.filter( + user=user, + purpose=EmailVerificationCode.PURPOSE_CHANGE_EMAIL, + is_used=False, + ).update(is_used=True) diff --git a/accounts/views.py b/accounts/views.py index c7d1074..b575cec 100644 --- a/accounts/views.py +++ b/accounts/views.py @@ -24,11 +24,14 @@ ProfileForm, AvatarUploadForm, UsernameChangeForm, + EmailChangeRequestForm, + EmailChangeConfirmForm, ) from .models import ( UserProfile, EmailVerificationCode, PasswordResetToken, + PendingEmailChange, LoginLockout, AccountEvent, ) @@ -42,6 +45,13 @@ ) from .services.avatar_service import process_and_save_avatar, remove_avatar from .services.username_service import change_username, get_username_cooldown_status +from .services.email_change_service import ( + request_email_change, + confirm_email_change, + cancel_email_change, +) +from .services.otp_service import issue_otp +from .services.email_service import send_templated_email from documents.models import Document from sharing.models import SharePass from audit.models import get_client_ip @@ -523,17 +533,122 @@ def account_settings_view(request): messages.error(request, "Could not update username. Please review the form.") return redirect(reverse("account_settings") + "#username") + elif action == "request_email_change": + email_request_form = EmailChangeRequestForm(user, request.POST) + if email_request_form.is_valid(): + try: + success, msg = request_email_change( + user=user, + new_email=email_request_form.cleaned_data["new_email"], + current_password=email_request_form.cleaned_data["current_password"], + request=request, + ) + if success: + messages.success(request, msg) + else: + messages.error(request, msg) + return redirect(reverse("account_settings") + "#email") + except ValidationError as exc: + messages.error(request, exc.message if hasattr(exc, "message") else str(exc)) + return redirect(reverse("account_settings") + "#email") + else: + err_msg = email_request_form.errors.get("new_email") or email_request_form.errors.get( + "current_password" + ) + if err_msg: + messages.error(request, err_msg[0]) + else: + messages.error(request, "Could not process email change. Please review the form.") + return redirect(reverse("account_settings") + "#email") + + elif action == "confirm_email_change": + email_confirm_form = EmailChangeConfirmForm(request.POST) + if email_confirm_form.is_valid(): + success, msg = confirm_email_change( + user=user, + submitted_code=email_confirm_form.cleaned_data["code"], + request=request, + ) + if success: + messages.success(request, msg) + else: + messages.error(request, msg) + return redirect(reverse("account_settings") + "#email") + else: + err_msg = email_confirm_form.errors.get("code") + if err_msg: + messages.error(request, err_msg[0]) + else: + messages.error(request, "Please enter a valid 6-digit verification code.") + return redirect(reverse("account_settings") + "#email") + + elif action == "resend_email_change_otp": + pending = PendingEmailChange.objects.filter(user=user).first() + if not pending: + messages.error(request, "No pending email change request found.") + return redirect(reverse("account_settings") + "#email") + + try: + _, raw_otp = issue_otp( + user=user, + email=pending.new_email, + purpose=EmailVerificationCode.PURPOSE_CHANGE_EMAIL, + ) + send_templated_email( + subject="Passli Vault: Verify Your New Email Address", + template_base="email_change_otp", + context={ + "username": user.username, + "new_email": pending.new_email, + "otp_code": raw_otp, + "ttl_minutes": 15, + }, + recipient_list=[pending.new_email], + fail_silently=True, + ) + messages.success(request, f"A fresh verification code has been dispatched to {pending.new_email}.") + except ValidationError as exc: + messages.warning(request, exc.message if hasattr(exc, "message") else str(exc)) + return redirect(reverse("account_settings") + "#email") + + elif action == "cancel_email_change": + cancel_email_change(user, request=request) + messages.info(request, "Pending email change request cancelled.") + return redirect(reverse("account_settings") + "#email") + # GET request form instances and context profile_form = ProfileForm(initial={"display_name": profile.display_name}) avatar_form = AvatarUploadForm() username_form = UsernameChangeForm(user=user) + email_request_form = EmailChangeRequestForm(user=user) + email_confirm_form = EmailChangeConfirmForm() + pending_email_change = PendingEmailChange.objects.filter(user=user).first() recent_activity = AccountEvent.objects.filter(user=user).order_by("-created_at")[:10] + resend_cooldown_seconds = 0 + if pending_email_change: + latest_code = ( + EmailVerificationCode.objects.filter( + user=user, + purpose=EmailVerificationCode.PURPOSE_CHANGE_EMAIL, + ) + .order_by("-created_at") + .first() + ) + if latest_code: + elapsed = (timezone.now() - latest_code.created_at).total_seconds() + if elapsed < 60: + resend_cooldown_seconds = max(0, int(60 - elapsed)) + context = { "profile": profile, "profile_form": profile_form, "avatar_form": avatar_form, "username_form": username_form, + "email_request_form": email_request_form, + "email_confirm_form": email_confirm_form, + "pending_email_change": pending_email_change, + "resend_cooldown_seconds": resend_cooldown_seconds, "cooldown_status": cooldown_status, "recent_activity": recent_activity, "has_avatar": bool(profile.avatar_path), diff --git a/static/css/settings.css b/static/css/settings.css index a49f804..9940d4d 100644 --- a/static/css/settings.css +++ b/static/css/settings.css @@ -355,6 +355,67 @@ outline-offset: 2px; } +/* Pending Alert Banner */ +.settings-pending-alert { + background-color: var(--bg-subtle); + border: 1px solid var(--border-light); + border-radius: 8px; + padding: 1rem; + margin-bottom: 1.25rem; + display: flex; + align-items: flex-start; + gap: 0.75rem; +} + +.settings-pending-alert-icon { + color: var(--brand-blue); + display: flex; + align-items: center; + padding-top: 0.1rem; +} + +/* 6-Digit OTP Box UI & Progressive Enhancement */ +.settings-otp-wrapper { + display: none; + gap: 8px; + margin: 0.5rem 0 0.5rem; + align-items: center; +} + +.js-ready .settings-otp-wrapper { + display: flex; +} + +.js-ready .settings-otp-fallback { + display: none; +} + +.settings-otp-box { + width: 48px; + height: 54px; + font-family: 'JetBrains Mono', monospace; + font-size: 1.5rem; + font-weight: 600; + text-align: center; + border: 1.5px solid var(--border-light); + border-radius: 8px; + background: var(--bg-surface); + color: var(--text-main); + outline: none; + transition: border-color 150ms ease, box-shadow 150ms ease; + -webkit-appearance: none; + margin: 0; +} + +.settings-otp-box:focus { + border-color: var(--brand-navy); + box-shadow: 0 0 0 3px rgba(15, 23, 42, 0.1); +} + +.settings-otp-box.filled { + border-color: #94a3b8; +} + /* Mobile Responsiveness */ @media (max-width: 768px) { .settings-layout { @@ -383,6 +444,16 @@ .settings-card { padding: 1.25rem; } + + .settings-otp-wrapper { + gap: 6px; + } + + .settings-otp-box { + width: 40px; + height: 48px; + font-size: 1.25rem; + } } @media (prefers-reduced-motion: reduce) { diff --git a/static/js/settings.js b/static/js/settings.js index 0521d19..4916f37 100644 --- a/static/js/settings.js +++ b/static/js/settings.js @@ -3,9 +3,13 @@ * Progressive enhancement for avatar preview, tab routing, and responsive interactions. */ +document.documentElement.classList.add('js-ready'); + document.addEventListener('DOMContentLoaded', () => { initNavigationTabs(); initAvatarUploader(); + initOtpInputs(); + initResendTimer(); }); function initNavigationTabs() { @@ -110,3 +114,100 @@ function initAvatarUploader() { reader.readAsDataURL(file); }); } + +function initOtpInputs() { + const container = document.getElementById('email-otp-boxes-container'); + const fallbackInput = document.getElementById('email-otp-combined-input'); + if (!container || !fallbackInput) return; + + const boxes = Array.from(container.querySelectorAll('.settings-otp-box')); + if (boxes.length !== 6) return; + + function syncToFallback() { + const fullVal = boxes.map(b => b.value.trim()).join(''); + fallbackInput.value = fullVal; + boxes.forEach(b => { + b.classList.toggle('filled', b.value.length > 0); + }); + } + + boxes.forEach((box, index) => { + box.addEventListener('input', () => { + const val = box.value.replace(/[^0-9]/g, ''); + box.value = val ? val[0] : ''; + syncToFallback(); + if (box.value && index < 5) { + boxes[index + 1].focus(); + boxes[index + 1].select(); + } + }); + + box.addEventListener('keydown', (e) => { + if (e.key === 'Backspace') { + if (!box.value && index > 0) { + boxes[index - 1].focus(); + boxes[index - 1].value = ''; + syncToFallback(); + e.preventDefault(); + } else if (box.value) { + box.value = ''; + syncToFallback(); + e.preventDefault(); + } + } else if (e.key === 'ArrowLeft' && index > 0) { + boxes[index - 1].focus(); + boxes[index - 1].select(); + e.preventDefault(); + } else if (e.key === 'ArrowRight' && index < 5) { + boxes[index + 1].focus(); + boxes[index + 1].select(); + e.preventDefault(); + } + }); + + box.addEventListener('paste', (e) => { + e.preventDefault(); + const pasteData = (e.clipboardData || window.clipboardData).getData('text'); + const cleanDigits = pasteData.replace(/[^0-9]/g, '').slice(0, 6); + if (!cleanDigits) return; + + cleanDigits.split('').forEach((digit, i) => { + if (boxes[i]) { + boxes[i].value = digit; + } + }); + syncToFallback(); + const focusIndex = Math.min(cleanDigits.length, 5); + boxes[focusIndex].focus(); + }); + + box.addEventListener('focus', () => { + box.select(); + }); + }); +} + +function initResendTimer() { + const resendBtn = document.getElementById('btn-resend-email-otp'); + const timerText = document.getElementById('resend-email-timer-text'); + if (!resendBtn || !timerText) return; + + let cooldownSeconds = parseInt(resendBtn.getAttribute('data-cooldown') || '0', 10); + if (isNaN(cooldownSeconds) || cooldownSeconds <= 0) return; + + resendBtn.disabled = true; + timerText.textContent = `Resend in ${cooldownSeconds}s`; + + const timer = setInterval(() => { + cooldownSeconds -= 1; + if (cooldownSeconds <= 0) { + clearInterval(timer); + resendBtn.disabled = false; + timerText.textContent = 'Resend Code'; + resendBtn.removeAttribute('data-cooldown'); + } else { + timerText.textContent = `Resend in ${cooldownSeconds}s`; + } + }, 1000); +} + diff --git a/templates/accounts/settings.html b/templates/accounts/settings.html index 5929d29..f115d24 100644 --- a/templates/accounts/settings.html +++ b/templates/accounts/settings.html @@ -210,35 +210,130 @@

Email Address

- {% if not user.email %} -

- Your account currently has no email address configured. Adding a verified email enables self-service password recovery and critical security alerts. -

- {% elif not profile.is_email_verified %} -

- Your email address is unverified. Please verify your address to activate full account recovery capabilities. -

- - verified Enter Verification Code - - {% endif %} - - -
-
+ {% if pending_email_change %} + +
+
+ mark_email_unread +
-
Change Email Address
- Requires entering your current password and confirming a 6-digit code sent to your new address. +
Verification in Progress
+

+ A 6-digit confirmation code was sent to {{ pending_email_change.new_email }}. Your email stays unchanged until you enter it. +

+
+
+ + +
+ {% csrf_token %} + + +
+ + + +
+ + + + + + +
+ + +
+ {{ email_confirm_form.code }} +
+ + Code expires in 15 minutes. 5 attempts maximum. + {% if email_confirm_form.code.errors %} + + {% endif %}
- +
+ +
+ {% csrf_token %} + + +
+ +
+ {% csrf_token %} + + +
- + {% endif %} diff --git a/tests/unit/test_account_settings_email_change.py b/tests/unit/test_account_settings_email_change.py new file mode 100644 index 0000000..e70ec9b --- /dev/null +++ b/tests/unit/test_account_settings_email_change.py @@ -0,0 +1,442 @@ +""" +Unit tests for Account Settings Phase 3: Email Change Workflow. +Verifies anti-enumeration, dual notifications, OTP purpose-scoping, +database-backed rate limits/cooldowns, and atomic collision handling. +""" + +from datetime import timedelta +from django.test import TestCase +from django.urls import reverse +from django.contrib.auth import get_user_model +from django.core import mail +from django.utils import timezone + +from accounts.models import ( + UserProfile, + EmailVerificationCode, + PendingEmailChange, + AccountEvent, +) +from accounts.services.otp_service import issue_otp +from tests.unit import get_test_password + +User = get_user_model() + + +class EmailChangeWorkflowTests(TestCase): + def setUp(self): + self.password = get_test_password() + self.user = User.objects.create_user( + username="testuser", + email="primary@example.com", + password=self.password, + ) + self.profile, _ = UserProfile.objects.get_or_create( + user=self.user, + defaults={"is_email_verified": True}, + ) + self.profile.is_email_verified = True + self.profile.save() + + self.other_user = User.objects.create_user( + username="otheruser", + email="existing@example.com", + password=self.password, + ) + + self.client.login(username="testuser", password=self.password) + + def test_user_email_unchanged_until_otp_confirmed(self): + """User email remains primary@example.com until OTP is confirmed.""" + response = self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "brandnew@example.com", + "current_password": self.password, + }, + follow=True, + ) + self.assertEqual(response.status_code, 200) + + # 1. Assert user.email is unchanged + self.user.refresh_from_db() + self.assertEqual(self.user.email, "primary@example.com") + + # 2. Assert pending record exists + pending = PendingEmailChange.objects.filter(user=self.user).first() + self.assertIsNotNone(pending) + self.assertEqual(pending.new_email, "brandnew@example.com") + + # 3. Retrieve raw OTP from verification code record + otp_record = EmailVerificationCode.objects.filter( + user=self.user, + purpose=EmailVerificationCode.PURPOSE_CHANGE_EMAIL, + is_used=False, + ).first() + self.assertIsNotNone(otp_record) + + # Find the OTP code in outgoing mail + self.assertEqual(len(mail.outbox), 2) # 1 to new_email, 1 notice to old_email + new_mail = [m for m in mail.outbox if "brandnew@example.com" in m.to][0] + # Extract 6-digit code from email body + raw_code = None + for word in new_mail.body.split(): + if word.isdigit() and len(word) == 6: + raw_code = word + break + self.assertIsNotNone(raw_code) + + # 4. Confirm with wrong code -> still unchanged + fail_resp = self.client.post( + reverse("account_settings"), + { + "action": "confirm_email_change", + "code": "000000", + }, + follow=True, + ) + self.assertEqual(fail_resp.status_code, 200) + self.assertTrue(any("invalid" in m.message.lower() for m in fail_resp.context["messages"])) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "primary@example.com") + + # 5. Confirm with valid code -> email updated & verified + success_resp = self.client.post( + reverse("account_settings"), + { + "action": "confirm_email_change", + "code": raw_code, + }, + follow=True, + ) + self.assertEqual(success_resp.status_code, 200) + self.user.refresh_from_db() + self.profile.refresh_from_db() + self.assertEqual(self.user.email, "brandnew@example.com") + self.assertTrue(self.profile.is_email_verified) + self.assertFalse(PendingEmailChange.objects.filter(user=self.user).exists()) + + def test_otp_goes_to_new_address_only(self): + """OTP code is delivered exclusively to the new address.""" + self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "target@example.com", + "current_password": self.password, + }, + ) + self.assertEqual(len(mail.outbox), 2) + new_email_msg = [m for m in mail.outbox if "target@example.com" in m.to][0] + old_email_msg = [m for m in mail.outbox if "primary@example.com" in m.to][0] + + # New email contains the verification code + self.assertIn("Verify Your New Email Address", new_email_msg.subject) + # Old email is a security advisory without the OTP code + self.assertIn("Security Alert", old_email_msg.subject) + self.assertNotIn("Verify Your New Email Address", old_email_msg.subject) + + def test_old_address_gets_notices_on_request_and_completion(self): + """Old address receives security alert on request and final confirmation on completion.""" + # 1. Request change + self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "forward@example.com", + "current_password": self.password, + }, + ) + self.assertEqual(len(mail.outbox), 2) + alert_msg = [m for m in mail.outbox if "primary@example.com" in m.to][0] + self.assertIn("Security Alert: Email Change Requested", alert_msg.subject) + + # Retrieve OTP code + new_msg = [m for m in mail.outbox if "forward@example.com" in m.to][0] + raw_code = [w for w in new_msg.body.split() if w.isdigit() and len(w) == 6][0] + + mail.outbox.clear() + + # 2. Confirm change + self.client.post( + reverse("account_settings"), + { + "action": "confirm_email_change", + "code": raw_code, + }, + ) + self.assertEqual(len(mail.outbox), 2) + confirmed_old_msg = [m for m in mail.outbox if "primary@example.com" in m.to][0] + self.assertIn("Security Confirmation: Passli Account Email Address Changed", confirmed_old_msg.subject) + + new_confirmed_msg = [m for m in mail.outbox if "forward@example.com" in m.to][0] + self.assertIn("Email Address Verified Successfully", new_confirmed_msg.subject) + + def test_admin_with_empty_email_can_add_and_verify(self): + """Admins and users without an email on file can add and verify one smoothly.""" + admin_user = User.objects.create_superuser( + username="passliadmin", + email="", + password="AdminPassword123!", + ) + self.client.login(username="passliadmin", password="AdminPassword123!") + + response = self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "admin@passli.org", + "current_password": "AdminPassword123!", + }, + follow=True, + ) + self.assertEqual(response.status_code, 200) + + # Only 1 email sent (to admin@passli.org) because there was no prior old email + self.assertEqual(len(mail.outbox), 1) + self.assertIn("admin@passli.org", mail.outbox[0].to) + + raw_code = [w for w in mail.outbox[0].body.split() if w.isdigit() and len(w) == 6][0] + + # Confirm + confirm_resp = self.client.post( + reverse("account_settings"), + { + "action": "confirm_email_change", + "code": raw_code, + }, + follow=True, + ) + self.assertEqual(confirm_resp.status_code, 200) + admin_user.refresh_from_db() + self.assertEqual(admin_user.email, "admin@passli.org") + admin_profile = UserProfile.objects.get(user=admin_user) + self.assertTrue(admin_profile.is_email_verified) + + def test_identical_response_for_taken_and_free_addresses(self): + """Anti-enumeration: taken address and free address produce identical response.""" + # 1. Request existing taken address + resp_taken = self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "existing@example.com", + "current_password": self.password, + }, + follow=True, + ) + self.assertEqual(resp_taken.status_code, 200) + messages_taken = [m.message for m in resp_taken.context["messages"]] + self.assertIn("If this address can be used, a verification code has been sent.", messages_taken) + # No email sent to taken address or user + self.assertEqual(len(mail.outbox), 0) + + # 2. Request fresh available address + resp_free = self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "fresh@example.com", + "current_password": self.password, + }, + follow=True, + ) + self.assertEqual(resp_free.status_code, 200) + messages_free = [m.message for m in resp_free.context["messages"]] + self.assertIn("If this address can be used, a verification code has been sent.", messages_free) + # Outgoing OTP email was sent + self.assertGreater(len(mail.outbox), 0) + + def test_rejection_of_expired_code(self): + """Codes older than 15 minutes are rejected and email remains unchanged.""" + self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "expirytest@example.com", + "current_password": self.password, + }, + ) + otp_record = EmailVerificationCode.objects.filter( + user=self.user, + purpose=EmailVerificationCode.PURPOSE_CHANGE_EMAIL, + ).first() + + # Fast forward expiration to 1 minute in the past + otp_record.expires_at = timezone.now() - timedelta(minutes=1) + otp_record.save() + + raw_code = [w for w in mail.outbox[0].body.split() if w.isdigit() and len(w) == 6][0] + + response = self.client.post( + reverse("account_settings"), + { + "action": "confirm_email_change", + "code": raw_code, + }, + follow=True, + ) + self.assertEqual(response.status_code, 200) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "primary@example.com") + messages = [m.message for m in response.context["messages"]] + self.assertTrue(any("expired" in msg.lower() for msg in messages)) + + def test_rejection_of_locked_code_after_5_attempts(self): + """Code is permanently locked after 5 failed submission attempts.""" + self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "lockout@example.com", + "current_password": self.password, + }, + ) + raw_code = [w for w in mail.outbox[0].body.split() if w.isdigit() and len(w) == 6][0] + + # 5 wrong attempts + for _ in range(5): + self.client.post( + reverse("account_settings"), + {"action": "confirm_email_change", "code": "111111"}, + ) + + # 6th attempt with CORRECT code is rejected because code is locked + response = self.client.post( + reverse("account_settings"), + {"action": "confirm_email_change", "code": raw_code}, + follow=True, + ) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "primary@example.com") + messages = [m.message for m in response.context["messages"]] + self.assertTrue(any("exceeded" in msg.lower() or "locked" in msg.lower() for msg in messages)) + + def test_rejection_of_reused_code(self): + """A code that has already been verified cannot be used again.""" + self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "reusetest@example.com", + "current_password": self.password, + }, + ) + raw_code = [w for w in mail.outbox[0].body.split() if w.isdigit() and len(w) == 6][0] + + # First use -> success + self.client.post( + reverse("account_settings"), + {"action": "confirm_email_change", "code": raw_code}, + ) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "reusetest@example.com") + + # Second attempt to use code -> fails + response = self.client.post( + reverse("account_settings"), + {"action": "confirm_email_change", "code": raw_code}, + follow=True, + ) + messages = [m.message for m in response.context["messages"]] + self.assertTrue(any("invalid" in msg.lower() or "no pending" in msg.lower() for msg in messages)) + + def test_cross_purpose_code_isolation(self): + """Codes issued for verify_email cannot be used for change_email.""" + PendingEmailChange.objects.create(user=self.user, new_email="spoofed@example.com") + _, verify_otp_code = issue_otp( + user=self.user, + email="spoofed@example.com", + purpose=EmailVerificationCode.PURPOSE_VERIFY_EMAIL, + ) + + response = self.client.post( + reverse("account_settings"), + { + "action": "confirm_email_change", + "code": verify_otp_code, + }, + follow=True, + ) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "primary@example.com") + messages = [m.message for m in response.context["messages"]] + self.assertTrue(any("invalid" in msg.lower() or "expired" in msg.lower() for msg in messages)) + + def test_race_condition_taken_email_between_request_and_confirm(self): + """If another user claims the pending address before confirmation, abort gracefully.""" + self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "race@example.com", + "current_password": self.password, + }, + ) + raw_code = [w for w in mail.outbox[0].body.split() if w.isdigit() and len(w) == 6][0] + + # Another user registers or claims race@example.com + User.objects.create_user( + username="racer", + email="race@example.com", + password=get_test_password(), + ) + + # Now testuser attempts to confirm + response = self.client.post( + reverse("account_settings"), + { + "action": "confirm_email_change", + "code": raw_code, + }, + follow=True, + ) + self.assertEqual(response.status_code, 200) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "primary@example.com") + self.assertFalse(PendingEmailChange.objects.filter(user=self.user).exists()) + messages = [m.message for m in response.context["messages"]] + self.assertTrue(any("no longer available" in msg.lower() for msg in messages)) + + def test_cancel_pending_email_change(self): + """User can cancel an active pending email change request.""" + self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "cancelme@example.com", + "current_password": self.password, + }, + ) + self.assertTrue(PendingEmailChange.objects.filter(user=self.user).exists()) + + cancel_resp = self.client.post( + reverse("account_settings"), + {"action": "cancel_email_change"}, + follow=True, + ) + self.assertEqual(cancel_resp.status_code, 200) + self.assertFalse(PendingEmailChange.objects.filter(user=self.user).exists()) + self.user.refresh_from_db() + self.assertEqual(self.user.email, "primary@example.com") + + def test_wrong_password_on_email_change_request(self): + """Entering incorrect current password aborts request and logs reauth_failed event.""" + response = self.client.post( + reverse("account_settings"), + { + "action": "request_email_change", + "new_email": "badpass@example.com", + "current_password": get_test_password(), + }, + follow=True, + ) + self.assertEqual(response.status_code, 200) + self.assertFalse(PendingEmailChange.objects.filter(user=self.user).exists()) + self.assertTrue( + AccountEvent.objects.filter( + user=self.user, + event_type=AccountEvent.EventType.REAUTH_FAILED, + ).exists() + )