Add address verification step in the "Change Email" flow

A similar issue has come up multiple times: the user
changes account's email address, enters a bad address
by mistake, and gets locked out of their account.

This commit adds an extra step in the "Change Email" flow:

* In "Account Settings", user clicks on [Change Email]
* User gets a prompt for a 6-digit confirmation code, which
  has been sent to their old address. This is to prevent
  account takeover when Eve sits down at a computer where Alice
  is logged in.
* The user enters the confirmation code, and a "Change Email"
  form loads.
* The user enters their new email address.
* (The new step!) Instead of changing the email right away,
  we send a special login link to user's specified new address.
* (The new step, continued) The user clicks on the login link,
  their account's email address gets updated, and they get
  logged in.

The additional step makes sure the user can receive email
at their new address. If they cannot receive email there,
they cannot complete the "Change Email" procedure.

CHANGELOG.md

@@ -3,6 +3,9 @@ All notable changes to this project will be documented in this file.
 
 ## V2.2 - Unreleased
 
+### Improvements
+- Add address verification step in the "Change Email" flow
+
 ### Bug Fixes
 - Update hc.front.views.channels to handle empty strings in settings (#635)
 - Add logic to handle ContentDecodingError exceptions

hc/accounts/models.py

@@ -118,6 +118,21 @@ class Profile(models.Model):
         }
         emails.login(self.user.email, ctx)
 
+    def send_change_email_link(self, new_email):
+        payload = {
+            "u": self.user.username,
+            "t": self.prepare_token("login"),
+            "e": new_email,
+        }
+        signed_payload = TimestampSigner().sign_object(payload)
+        path = reverse("hc-change-email-verify", args=[signed_payload])
+
+        ctx = {
+            "button_text": "Sign In",
+            "button_url": settings.SITE_ROOT + path,
+        }
+        emails.login(new_email, ctx)
+
     def send_transfer_request(self, project):
         token = self.prepare_token("login")
         settings_path = reverse("hc-project-settings", args=[project.code])

hc/accounts/tests/test_change_email.py

@@ -1,3 +1,6 @@
+from django.conf import settings
+from django.core import mail
+from django.test.utils import override_settings
 from hc.test import BaseTestCase
 
 
@@ -15,21 +18,27 @@ class ChangeEmailTestCase(BaseTestCase):
         r = self.client.get("/accounts/change_email/")
         self.assertContains(r, "Change Account's Email Address")
 
-    def test_it_updates_email(self):
+    @override_settings(SITE_ROOT="http://testserver")
+    def test_it_sends_link(self):
         self.client.login(username="alice@example.org", password="password")
         self.set_sudo_flag()
 
         payload = {"email": "alice2@example.org"}
         r = self.client.post("/accounts/change_email/", payload, follow=True)
-        self.assertRedirects(r, "/accounts/change_email/done/")
-        self.assertContains(r, "Email Address Updated")
+        self.assertRedirects(r, "/accounts/change_email/")
+        self.assertContains(r, "One Last Step")
 
+        # The email addess should have not changed yet
         self.alice.refresh_from_db()
-        self.assertEqual(self.alice.email, "alice2@example.org")
-        self.assertFalse(self.alice.has_usable_password())
-
-        # The user should have been logged out:
-        self.assertNotIn("_auth_user_id", self.client.session)
+        self.assertEqual(self.alice.email, "alice@example.org")
+        self.assertTrue(self.alice.has_usable_password())
+
+        # And email should have been sent
+        self.assertEqual(len(mail.outbox), 1)
+        message = mail.outbox[0]
+        self.assertEqual(message.subject, f"Log in to {settings.SITE_NAME}")
+        html = message.alternatives[0][0]
+        self.assertIn("http://testserver/accounts/change_email/", html)
 
     def test_it_requires_unique_email(self):
         self.client.login(username="alice@example.org", password="password")

hc/accounts/tests/test_change_email_verify.py

@@ -0,0 +1,65 @@
+from unittest.mock import patch
+import time
+
+from django.contrib.auth.hashers import make_password
+from django.core.signing import TimestampSigner
+from hc.test import BaseTestCase
+
+
+class ChangeEmailVerifyTestCase(BaseTestCase):
+    def setUp(self):
+        super().setUp()
+        self.profile.token = make_password("secret-token", "login")
+        self.profile.save()
+
+        self.checks_url = "/projects/%s/checks/" % self.project.code
+
+    def _url(self, expired=False):
+        payload = {
+            "u": self.alice.username,
+            "t": "secret-token",
+            "e": "alice+new@example.org",
+        }
+
+        if expired:
+            with patch("django.core.signing.TimestampSigner.timestamp") as mock_ts:
+                mock_ts.return_value = "1kHR5c"
+                signed = TimestampSigner().sign_object(payload)
+        else:
+            signed = TimestampSigner().sign_object(payload)
+
+        return f"/accounts/change_email/{signed}/"
+
+    def test_it_works(self):
+        r = self.client.post(self._url())
+        self.assertRedirects(r, self.checks_url)
+
+        # Alice's email should have been updated, and password cleared
+        self.alice.refresh_from_db()
+        self.assertEqual(self.alice.email, "alice+new@example.org")
+        self.assertFalse(self.alice.has_usable_password())
+
+        # After login, token should be blank
+        self.profile.refresh_from_db()
+        self.assertEqual(self.profile.token, "")
+
+    def test_it_handles_get(self):
+        r = self.client.get(self._url())
+        self.assertContains(r, "You are about to log into")
+
+        # Alice's email should have *not* been changed yet
+        self.alice.refresh_from_db()
+        self.assertEqual(self.alice.email, "alice@example.org")
+
+    def test_it_handles_get_with_cookie(self):
+        self.client.cookies["auto-login"] = "1"
+        r = self.client.get(self._url())
+        self.assertRedirects(r, self.checks_url)
+
+    def test_it_handles_expired_link(self):
+        r = self.client.post(self._url(expired=True))
+        self.assertContains(r, "The link you just used is incorrect.")
+
+    def test_it_handles_bad_payload(self):
+        r = self.client.post("/accounts/change_email/bad-payload/")
+        self.assertContains(r, "The link you just used is incorrect.")

hc/accounts/urls.py

@@ -23,8 +23,12 @@ urlpatterns = [
         name="hc-unsubscribe-reports",
     ),
     path("set_password/", views.set_password, name="hc-set-password"),
-    path("change_email/done/", views.change_email_done, name="hc-change-email-done"),
     path("change_email/", views.change_email, name="hc-change-email"),
+    path(
+        "change_email/<str:signed_payload>/",
+        views.change_email_verify,
+        name="hc-change-email-verify",
+    ),
     path("two_factor/webauthn/", views.add_webauthn, name="hc-add-webauthn"),
     path("two_factor/totp/", views.add_totp, name="hc-add-totp"),
     path("two_factor/totp/remove/", views.remove_totp, name="hc-remove-totp"),

hc/accounts/views.py

@@ -1,7 +1,7 @@
 import base64
 from datetime import timedelta as td
 from secrets import token_bytes, token_urlsafe
-from urllib.parse import urlparse
+from urllib.parse import urlencode, urlparse
 import time
 import uuid
 
@@ -13,7 +13,7 @@ from django.contrib.auth import logout as auth_logout
 from django.contrib.auth import authenticate, update_session_auth_hash
 from django.contrib.auth.decorators import login_required
 from django.contrib.auth.models import User
-from django.core import signing
+from django.core.signing import TimestampSigner, BadSignature, SignatureExpired
 from django.http import HttpResponse, HttpResponseForbidden, HttpResponseBadRequest
 from django.shortcuts import get_object_or_404, redirect, render
 from django.utils.timezone import now
@@ -100,7 +100,7 @@ def _make_user(email, tz=None, with_project=True):
 
 
 def _redirect_after_login(request):
-    """ Redirect to the URL indicated in ?next= query parameter. """
+    """Redirect to the URL indicated in ?next= query parameter."""
 
     redirect_url = request.GET.get("next")
     if _allow_redirect(redirect_url):
@@ -167,7 +167,7 @@ def login(request):
                 profile.send_instant_login_link(redirect_url=redirect_url)
                 response = redirect("hc-login-link-sent")
 
-                # check_token_submit looks for this cookie to decide if
+                # check_token looks for this cookie to decide if
                 # it needs to do the extra POST step.
                 response.set_cookie("auto-login", "1", max_age=300, httponly=True)
                 return response
@@ -221,10 +221,9 @@ def login_link_sent(request):
     return render(request, "accounts/login_link_sent.html")
 
 
-def check_token(request, username, token):
-    if request.user.is_authenticated and request.user.username == username:
-        # User is already logged in
-        return _redirect_after_login(request)
+def check_token(request, username, token, new_email=None):
+    if request.user.is_authenticated:
+        auth_logout(request)
 
     # Some email servers open links in emails to check for malicious content.
     # To work around this, we sign user in if the method is POST
@@ -232,18 +231,22 @@ def check_token(request, username, token):
     #
     # If the method is GET and the auto-login cookie isn't present, we serve
     # a HTML form with a submit button.
+    if request.method != "POST" and "auto-login" not in request.COOKIES:
+        return render(request, "accounts/check_token_submit.html")
 
-    if request.method == "POST" or "auto-login" in request.COOKIES:
-        user = authenticate(username=username, token=token)
-        if user is not None and user.is_active:
-            user.profile.token = ""
-            user.profile.save()
-            return _check_2fa(request, user)
+    user = authenticate(username=username, token=token)
+    if user is not None and user.is_active:
+        if new_email:
+            user.email = new_email
+            user.set_unusable_password()
+            user.save()
 
-        request.session["bad_link"] = True
-        return redirect("hc-login")
+        user.profile.token = ""
+        user.profile.save()
+        return _check_2fa(request, user)
 
-    return render(request, "accounts/check_token_submit.html")
+    request.session["bad_link"] = True
+    return redirect("hc-login")
 
 
 @login_required
@@ -552,25 +555,40 @@ def set_password(request):
 @login_required
 @require_sudo_mode
 def change_email(request):
+    if "sent" in request.session:
+        ctx = {"email": request.session.pop("sent")}
+        return render(request, "accounts/change_email_instructions.html", ctx)
+
     if request.method == "POST":
         form = forms.ChangeEmailForm(request.POST)
         if form.is_valid():
-            request.user.email = form.cleaned_data["email"]
-            request.user.set_unusable_password()
-            request.user.save()
-
-            request.profile.token = ""
-            request.profile.save()
-
-            return redirect("hc-change-email-done")
+            # The user has entered a valid-looking new email address.
+            # Send a special login link to the new address. When the user
+            # clicks the special login link, hc.accounts.views.change_email_verify
+            # unpacks the payload, and passes it to hc.accounts.views.check_token,
+            # which finally updates user's email address.
+            email = form.cleaned_data["email"]
+            request.profile.send_change_email_link(email)
+            request.session["sent"] = email
+
+            response = redirect(reverse("hc-change-email"))
+            # check_token looks for this cookie to decide if
+            # it needs to do the extra POST step.
+            response.set_cookie("auto-login", "1", max_age=900, httponly=True)
+            return response
     else:
         form = forms.ChangeEmailForm()
 
     return render(request, "accounts/change_email.html", {"form": form})
 
 
-def change_email_done(request):
-    return render(request, "accounts/change_email_done.html")
+def change_email_verify(request, signed_payload):
+    try:
+        payload = TimestampSigner().unsign_object(signed_payload, max_age=900)
+    except BadSignature:
+        return render(request, "bad_link.html")
+
+    return check_token(request, payload["u"], payload["t"], payload["e"])
 
 
 @csrf_exempt
@@ -580,11 +598,11 @@ def unsubscribe_reports(request, signed_username):
     # If the signature is more than 5 minutes old, we also include JS code to
     # auto-submit the form.
 
-    signer = signing.TimestampSigner(salt="reports")
+    signer = TimestampSigner(salt="reports")
     # First, check the signature without looking at the timestamp:
     try:
         username = signer.unsign(signed_username)
-    except signing.BadSignature:
+    except BadSignature:
         return render(request, "bad_link.html")
 
     try:
@@ -600,7 +618,7 @@ def unsubscribe_reports(request, signed_username):
         try:
             autosubmit = False
             username = signer.unsign(signed_username, max_age=300)
-        except signing.SignatureExpired:
+        except SignatureExpired:
             autosubmit = True
 
         ctx = {"autosubmit": autosubmit}
@@ -650,7 +668,7 @@ def remove_project(request, code):
 
 
 def _get_credential_data(request, form):
-    """ Complete WebAuthn registration, return binary credential data.
+    """Complete WebAuthn registration, return binary credential data.
 
     This function is an interface to the fido2 library. It is separated
     out so that we don't need to mock ClientData, AttestationObject,
@@ -794,7 +812,7 @@ def remove_credential(request, code):
 
 
 def _check_credential(request, form, credentials):
-    """ Complete WebAuthn authentication, return True on success.
+    """Complete WebAuthn authentication, return True on success.
 
     This function is an interface to the fido2 library. It is separated
     out so that we don't need to mock ClientData, AuthenticatorData,
@@ -859,7 +877,6 @@ def login_webauthn(request):
     totp_url = None
     if user.profile.totp:
         totp_url = reverse("hc-login-totp")
-
         redirect_url = request.GET.get("next")
         if _allow_redirect(redirect_url):
             totp_url += "?next=%s" % redirect_url

templates/accounts/change_email_done.html → templates/accounts/change_email_instructions.html

@@ -4,12 +4,12 @@
 <div class="row">
     <div class="col-sm-6 col-sm-offset-3">
         <div class="hc-dialog">
-            <h1>Email Address Updated</h1>
+            <h1>One Last Step&hellip;</h1>
             <br />
             <p>
-                Your account's email address has been updated.
-                You can now <a href="{% url 'hc-login' %}">sign in</a>
-                with the new email address.
+                We've sent a sign in link to <strong>{{ email }}</strong>.
+                To finalize the email address change, please log
+                in using that link.
             </p>
         </div>
     </div>