fix(user): password change silently lost when notification email fails to send

nextcloudappstore/urls.py

@@ -28,6 +28,7 @@
     AppScaffoldingView,
     IntegrationScaffoldingView,
 )
+from nextcloudappstore.user.views import PasswordResetFromKeyView
 
 admin.site.login = login_required(admin.site.login)
 
@@ -36,6 +37,11 @@
     path("featured", CategoryAppListView.as_view(), {"id": None, "is_featured_category": True}, name="featured"),
     path("signup/", csp_update(**settings.CSP_SIGNUP)(signup), name="account_signup"),
     path("social/signup/", csp_update(**settings.CSP_SIGNUP)(social_signup), name="socialaccount_signup"),
+    re_path(
+        r"^password/reset/key/(?P<uidb36>[0-9A-Za-z]+)-(?P<key>.+)/$",
+        PasswordResetFromKeyView.as_view(),
+        name="account_reset_password_from_key",
+    ),
     path("", include("allauth.urls")),
     re_path(r"^categories/(?P<id>[\w]*)/?$", CategoryAppListView.as_view(), name="category-app-list"),
     re_path(r"^developer/apps/generate/?$", AppScaffoldingView.as_view(), name="app-scaffold"),

nextcloudappstore/user/signals.py

@@ -8,6 +8,7 @@
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import User
+from django.core.exceptions import ValidationError
 from django.core.mail import send_mail
 from django.db.models.signals import post_save, pre_save
 from django.dispatch import receiver
@@ -30,7 +31,11 @@ def post_save_create_token(sender, **kwargs):
 @receiver(pre_save, sender=get_user_model())
 def password_changed_signal(sender, instance, **kwargs):
     """
-    Regenerate token on password change
+    Notify the user by email and regenerate their token on password change.
+    This is a pre_save hook: raising here aborts the save, so the password
+    change is only persisted once the notification email is confirmed sent.
+    If it can't be sent (e.g. a mail outage), we'd rather block the change
+    than silently persist it without notifying the user.
     :param sender:
     :param instance:
     :param kwargs:
@@ -41,14 +46,23 @@ def password_changed_signal(sender, instance, **kwargs):
 
     try:
         user = get_user_model().objects.get(pk=instance.pk)
-        if user.password != instance.password and instance._password is not None:
-            # make sure we only send an email when user changed their
-            # password and _password is set and not when password hash
-            # was changed due to a new default hashing algorithm
-            update_token(user.username)
-            send_mail(mail_subect, mail_message, settings.DEFAULT_FROM_EMAIL, [user.email], False)
     except User.DoesNotExist:
-        pass
+        return
+
+    if user.password == instance.password or instance._password is None:
+        # only act when the user changed their password and _password is
+        # set, not when the password hash changed due to a new default
+        # hashing algorithm
+        return
+
+    try:
+        send_mail(mail_subect, mail_message, settings.DEFAULT_FROM_EMAIL, [user.email], False)
+    except Exception as exc:
+        raise ValidationError(
+            _("Could not send the password change notification email. Your password was not changed.")
+        ) from exc
+
+    update_token(user.username)
 
 
 @receiver(email_confirmed)

nextcloudappstore/user/tests.py

@@ -3,15 +3,19 @@
 SPDX-License-Identifier: AGPL-3.0-or-later
 """
 
+import re
 from unittest.mock import patch
 
 from allauth.account.models import EmailAddress
 from allauth.core import ratelimit
 from django.contrib.auth import get_user_model
+from django.core import mail
 from django.core.cache import cache
+from django.core.exceptions import ValidationError
 from django.test import RequestFactory, TestCase, override_settings
 from django.urls import reverse
 from email_validator import validate_email as _real_validate_email
+from rest_framework.authtoken.models import Token
 
 from nextcloudappstore.user.facades import create_user
 
@@ -184,3 +188,129 @@ def test_budget_shared_with_allauth_email_view(self):
         email_url = reverse("account_email")
         r = self.client.post(email_url, {"email": "viaemail@nextcloud.com", "action_add": ""})
         self.assertEqual(r.status_code, 429)
+
+
+class PasswordChangedSignalTest(TestCase):
+    def setUp(self):
+        self.user = create_user(username="pwuser", password=PASSWORD, email=PRIMARY_EMAIL)
+
+    def test_password_change_persists_when_email_succeeds(self):
+        old_hash = self.user.password
+        self.user.set_password("New-Sup3rS3cret!")
+        self.user.save()
+
+        refreshed = get_user_model().objects.get(pk=self.user.pk)
+        self.assertNotEqual(old_hash, refreshed.password)
+
+    def test_password_change_not_persisted_when_email_fails(self):
+        old_hash = self.user.password
+        self.user.set_password("New-Sup3rS3cret!")
+
+        with patch("nextcloudappstore.user.signals.send_mail", side_effect=Exception("SMTP down")):
+            with self.assertRaises(ValidationError):
+                self.user.save()
+
+        refreshed = get_user_model().objects.get(pk=self.user.pk)
+        self.assertEqual(old_hash, refreshed.password)
+
+    def test_token_not_regenerated_when_email_fails(self):
+        old_token = Token.objects.get(user=self.user).key
+        self.user.set_password("New-Sup3rS3cret!")
+
+        with patch("nextcloudappstore.user.signals.send_mail", side_effect=Exception("SMTP down")):
+            with self.assertRaises(ValidationError):
+                self.user.save()
+
+        self.assertEqual(old_token, Token.objects.get(user=self.user).key)
+
+    def test_no_email_sent_when_password_unchanged(self):
+        with patch("nextcloudappstore.user.signals.send_mail") as mock_send_mail:
+            self.user.first_name = "Changed"
+            self.user.save()
+
+        mock_send_mail.assert_not_called()
+
+
+class PasswordChangeViewFriendlyErrorTest(TestCase):
+    """The logged-in /account/password/ form must show a warning, not crash,
+    when password_changed_signal aborts the save."""
+
+    def setUp(self):
+        self.user = create_user(username="pwchangeuser", password=PASSWORD, email=PRIMARY_EMAIL)
+        self.client.login(username="pwchangeuser", password=PASSWORD)
+        self.url = reverse("user:account-password")
+
+    def _post(self, new_password="New-Sup3rS3cret!"):
+        return self.client.post(
+            self.url,
+            {
+                "oldpassword": PASSWORD,
+                "password1": new_password,
+                "password2": new_password,
+            },
+        )
+
+    def test_change_persists_when_email_succeeds(self):
+        response = self._post()
+        self.assertEqual(response.status_code, 302)
+        refreshed = get_user_model().objects.get(pk=self.user.pk)
+        self.assertTrue(refreshed.check_password("New-Sup3rS3cret!"))
+
+    def test_change_shows_warning_instead_of_crashing_when_email_fails(self):
+        with patch("nextcloudappstore.user.signals.send_mail", side_effect=Exception("SMTP down")):
+            response = self._post()
+
+        self.assertEqual(response.status_code, 200)
+        shown_messages = [str(m) for m in response.context["messages"]]
+        self.assertTrue(any("not changed" in m for m in shown_messages))
+        refreshed = get_user_model().objects.get(pk=self.user.pk)
+        self.assertTrue(refreshed.check_password(PASSWORD))
+
+
+class PasswordResetFromKeyViewFriendlyErrorTest(TestCase):
+    """The forgot-password flow must show a warning, not crash, when
+    password_changed_signal aborts the save."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls._validate_patcher = patch(
+            "nextcloudappstore.user.adapters.validate_email",
+            side_effect=_validate_email_no_dns,
+        )
+        cls._validate_patcher.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._validate_patcher.stop()
+        super().tearDownClass()
+
+    def setUp(self):
+        cache.clear()
+        self.user = create_user(username="resetuser", password=PASSWORD, email=PRIMARY_EMAIL)
+
+    def _get_set_password_url(self):
+        self.client.post(reverse("account_reset_password"), {"email": PRIMARY_EMAIL})
+        body = mail.outbox[-1].body
+        match = re.search(r"https?://[^\s]+(/password/reset/key/\S+?)/?(?:\s|$)", body)
+        response = self.client.get(match.group(1) + "/", follow=True)
+        self.assertEqual(response.status_code, 200)
+        return response.redirect_chain[-1][0]
+
+    def test_reset_persists_when_email_succeeds(self):
+        url = self._get_set_password_url()
+        response = self.client.post(url, {"password1": "New-Sup3rS3cret!", "password2": "New-Sup3rS3cret!"})
+        self.assertEqual(response.status_code, 302)
+        refreshed = get_user_model().objects.get(pk=self.user.pk)
+        self.assertTrue(refreshed.check_password("New-Sup3rS3cret!"))
+
+    def test_reset_shows_warning_instead_of_crashing_when_email_fails(self):
+        url = self._get_set_password_url()
+        with patch("nextcloudappstore.user.signals.send_mail", side_effect=Exception("SMTP down")):
+            response = self.client.post(url, {"password1": "New-Sup3rS3cret!", "password2": "New-Sup3rS3cret!"})
+
+        self.assertEqual(response.status_code, 200)
+        shown_messages = [str(m) for m in response.context["messages"]]
+        self.assertTrue(any("not changed" in m for m in shown_messages))
+        refreshed = get_user_model().objects.get(pk=self.user.pk)
+        self.assertTrue(refreshed.check_password(PASSWORD))

nextcloudappstore/user/views.py

@@ -5,10 +5,14 @@
 
 from allauth.account.models import EmailAddress
 from allauth.account.views import PasswordChangeView
+from allauth.account.views import (
+    PasswordResetFromKeyView as AllauthPasswordResetFromKeyView,
+)
 from allauth.core import ratelimit
 from django.contrib import messages
 from django.contrib.auth import logout
 from django.contrib.auth.mixins import LoginRequiredMixin
+from django.core.exceptions import ValidationError
 from django.db.models import Q
 from django.http import HttpResponseRedirect
 from django.shortcuts import get_object_or_404, redirect, render
@@ -193,12 +197,34 @@ class PasswordView(LoginRequiredMixin, PasswordChangeView):
     template_name = "user/password.html"
     success_url = reverse_lazy("user:account-password")
 
+    def form_valid(self, form):
+        try:
+            return super().form_valid(form)
+        except ValidationError as exc:
+            # Raised by password_changed_signal when the change couldn't be
+            # persisted (e.g. the notification email failed to send). Show
+            # it as a warning instead of an unhandled error page.
+            messages.error(self.request, " ".join(exc.messages))
+            return self.form_invalid(form)
+
     def get_context_data(self, **kwargs):
         context = super().get_context_data(**kwargs)
         context["acc_page"] = "password"
         return context
 
 
+class PasswordResetFromKeyView(AllauthPasswordResetFromKeyView):
+    """Same as allauth's view, but shows a friendly warning instead of an
+    unhandled error page if password_changed_signal aborts the save."""
+
+    def form_valid(self, form):
+        try:
+            return super().form_valid(form)
+        except ValidationError as exc:
+            messages.error(self.request, " ".join(exc.messages))
+            return self.form_invalid(form)
+
+
 @method_decorator(never_cache, name="dispatch")
 class APITokenView(LoginRequiredMixin, TemplateView):
     """Display the user's API token, and allow it to be regenerated."""