feat(accounts): Add notifications for account events

allauth/account/adapter.py

@@ -693,6 +693,9 @@ def get_client_ip(self, request):
             ip = request.META.get("REMOTE_ADDR")
         return ip
 
+    def get_http_user_agent(self, request):
+        return request.META["HTTP_USER_AGENT"]
+
     def generate_emailconfirmation_key(self, email):
         key = get_random_string(64).lower()
         return key
@@ -729,6 +732,22 @@ def get_reauthentication_methods(self, user):
                 )
         return ret
 
+    def send_notification_mail(self, template_prefix, user, context):
+        from allauth.account.models import EmailAddress
+
+        if app_settings.ACCOUNT_EMAIL_NOTIFICATIONS:
+            context.update(
+                {
+                    "current_site": get_current_site(self.request),
+                    "timestamp": timezone.now(),
+                    "ip": self.get_client_ip(self.request),
+                    "user_agent": self.get_http_user_agent(self.request),
+                }
+            )
+            email = EmailAddress.objects.get_primary(user)
+            if email:
+                self.send_mail(template_prefix, email.email, context)
+
 
 def get_adapter(request=None):
     return import_attribute(app_settings.ADAPTER)(request)

allauth/account/app_settings.py

@@ -390,6 +390,10 @@ def PASSWORD_RESET_TOKEN_GENERATOR(self):
     def REAUTHENTICATION_TIMEOUT(self):
         return self._setting("REAUTHENTICATION_TIMEOUT", 300)
 
+    @property
+    def ACCOUNT_EMAIL_NOTIFICATIONS(self):
+        return self._setting("ACCOUNT_EMAIL_NOTIFICATIONS", False)
+
     @property
     def REAUTHENTICATION_REQUIRED(self):
         return self._setting("REAUTHENTICATION_REQUIRED", False)

allauth/account/tests/test_change_email.py

@@ -2,6 +2,7 @@
 from unittest.mock import patch
 
 from django.contrib.auth import get_user_model
+from django.core import mail
 from django.urls import reverse
 
 import pytest
@@ -380,3 +381,17 @@ def test_dont_lookup_invalid_email(auth_client, email, did_look_up):
             {"action_remove": "", "email": email},
         )
         assert gfu_mock.called == did_look_up
+
+
+# @patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+# def test_notification_on_email_change(user, client, settings):
+#     secondary = EmailAddress.objects.create(
+#         email="[email protected]", user=user, verified=False, primary=True
+#     )
+#     resp = client.post(
+#         reverse("account_email"),
+#         {"action_primary": "", "email": secondary.email},
+#     )
+#     assert resp.status_code == 302
+#     assert len(mail.outbox) == 1
+#     assert "Your email has been changed.{" in mail.outbox[0].body

allauth/account/tests/test_confirm_email.py

@@ -362,3 +362,36 @@ def test_confirm_email_with_same_user_logged_in(self):
         )
 
         self.assertEqual(user, resp.wsgi_request.user)
+
+
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+def test_notification_on_email_add(auth_client, user, client):
+    settings.ACCOUNT_MAX_EMAIL_ADDRESSES = 2
+    client.force_login(user)
+    response = client.post(
+        reverse("account_email"),
+        {"email": "[email protected]", "action_add": ""},
+        **{
+            "HTTP_USER_AGENT": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
+        }
+    )
+    assert response.status_code == 302
+    assert len(mail.outbox) == 2
+    assert "Email address has been added." in mail.outbox[1].body
+
+
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+def test_notification_on_email_remove(auth_client, user):
+    secondary = EmailAddress.objects.create(
+        email="[email protected]", user=user, verified=False, primary=False
+    )
+    resp = auth_client.post(
+        reverse("account_email"),
+        {"action_remove": "", "email": secondary.email},
+        **{
+            "HTTP_USER_AGENT": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
+        }
+    )
+    assert resp.status_code == 302
+    assert len(mail.outbox) == 1
+    assert "Following email has been removed" in mail.outbox[0].body

allauth/account/tests/test_reset_password.py

@@ -1,4 +1,5 @@
 import json
+from unittest.mock import patch
 
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import AnonymousUser
@@ -298,3 +299,52 @@ def _create_user_and_login(self, usable_password=True):
     def _password_set_or_change_redirect(self, urlname, usable_password):
         self._create_user_and_login(usable_password)
         return self.client.get(reverse(urlname))
+
+
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+def test_notification_on_password_change(user_factory, client):
+    user = user_factory(
+        email="[email protected]",
+        password="password",
+        email_verified=True,
+    )
+    client.force_login(user)
+
+    client.post(
+        reverse("account_change_password"),
+        data={
+            "oldpassword": "password",
+            "password1": "change_password",
+            "password2": "change_password",
+        },
+        **{
+            "HTTP_USER_AGENT": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
+        }
+    )
+    assert len(mail.outbox) == 1
+    print(mail.outbox[0].body)
+    assert "Your password has been changed" in mail.outbox[0].body
+
+
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+def test_notification_on_password_reset(user_factory, client, settings):
+    user = user_factory(
+        email="[email protected]",
+        password="password",
+        email_verified=True,
+    )
+
+    client.post(reverse("account_reset_password"), data={"email": user.email})
+    body = mail.outbox[0].body
+    url = body[body.find("/password/reset/") :].split()[0]
+    resp = client.get(url)
+    resp = client.post(
+        resp.url,
+        {"password1": "newpass123", "password2": "newpass123"},
+        **{
+            "HTTP_USER_AGENT": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
+        }
+    )
+
+    assert len(mail.outbox) == 2
+    assert "Your password has been reset" in mail.outbox[1].body

allauth/account/views.py

@@ -485,7 +485,8 @@ def get_form_kwargs(self):
 
     def form_valid(self, form):
         email_address = form.save(self.request)
-        get_adapter(self.request).add_message(
+        adapter = get_adapter(self.request)
+        adapter.add_message(
             self.request,
             messages.INFO,
             "account/messages/email_confirmation_sent.txt",
@@ -497,6 +498,9 @@ def form_valid(self, form):
             user=self.request.user,
             email_address=email_address,
         )
+        adapter.send_notification_mail(
+            "account/email/email_added", self.request.user, {"email": email_address}
+        )
         return super(EmailView, self).form_valid(form)
 
     def post(self, request, *args, **kwargs):
@@ -563,6 +567,11 @@ def _action_remove(self, request, *args, **kwargs):
                     "account/messages/email_deleted.txt",
                     {"email": email_address.email},
                 )
+                adapter.send_notification_mail(
+                    "account/email/email_removed",
+                    request.user,
+                    {"email": email_address},
+                )
                 return HttpResponseRedirect(self.get_success_url())
 
     def _action_primary(self, request, *args, **kwargs):
@@ -593,7 +602,8 @@ def _action_primary(self, request, *args, **kwargs):
                 except EmailAddress.DoesNotExist:
                     from_email_address = None
                 email_address.set_as_primary()
-                get_adapter().add_message(
+                adapter = get_adapter()
+                adapter.add_message(
                     request,
                     messages.SUCCESS,
                     "account/messages/primary_email_set.txt",
@@ -605,6 +615,14 @@ def _action_primary(self, request, *args, **kwargs):
                     from_email_address=from_email_address,
                     to_email_address=email_address,
                 )
+                adapter.send_notification_mail(
+                    "account/email/email_changed",
+                    request.user,
+                    {
+                        "from_email_address": from_email_address,
+                        "to_email_address": email_address,
+                    },
+                )
                 return HttpResponseRedirect(self.get_success_url())
 
     def get_context_data(self, **kwargs):
@@ -689,11 +707,15 @@ def get_form_kwargs(self):
     def form_valid(self, form):
         form.save()
         logout_on_password_change(self.request, form.user)
-        get_adapter(self.request).add_message(
+        adapter = get_adapter(self.request)
+        adapter.add_message(
             self.request,
             messages.SUCCESS,
             "account/messages/password_changed.txt",
         )
+        adapter.send_notification_mail(
+            "account/email/password_changed", self.request.user, {}
+        )
         signals.password_changed.send(
             sender=self.request.user.__class__,
             request=self.request,
@@ -745,14 +767,18 @@ def get_form_kwargs(self):
     def form_valid(self, form):
         form.save()
         logout_on_password_change(self.request, form.user)
-        get_adapter(self.request).add_message(
+        adapter = get_adapter(self.request)
+        adapter.add_message(
             self.request, messages.SUCCESS, "account/messages/password_set.txt"
         )
         signals.password_set.send(
             sender=self.request.user.__class__,
             request=self.request,
             user=self.request.user,
         )
+        adapter.send_notification_mail(
+            "account/email/password_set", self.request.user, {}
+        )
         return super(PasswordSetView, self).form_valid(form)
 
     def get_context_data(self, **kwargs):
@@ -904,6 +930,9 @@ def form_valid(self, form):
             request=self.request,
             user=self.reset_user,
         )
+        adapter.send_notification_mail(
+            "account/email/password_reset", self.reset_user, {}
+        )
 
         if app_settings.LOGIN_ON_PASSWORD_RESET:
             return perform_login(

allauth/mfa/adapter.py

@@ -1,6 +1,7 @@
 from django.utils.translation import gettext_lazy as _
 
 from allauth import app_settings as allauth_settings
+from allauth.account.adapter import get_adapter as get_account_adapter
 from allauth.account.utils import user_email, user_username
 from allauth.core import context
 from allauth.mfa import app_settings
@@ -62,6 +63,9 @@ def decrypt(self, encrypted_text: str) -> str:
         text = encrypted_text
         return text
 
+    def send_notification_mail(self, *args, **kwargs):
+        return get_account_adapter().send_notification_mail(*args, **kwargs)
+
 
 def get_adapter():
     return import_attribute(app_settings.ADAPTER)()

allauth/mfa/migrations/0002_authenticator_timestamps.py

@@ -1,7 +1,7 @@
 # Generated by Django 3.2.22 on 2023-11-06 12:04
 
-from django.db import migrations, models
 import django.utils.timezone
+from django.db import migrations, models
 
 
 class Migration(migrations.Migration):

allauth/mfa/models.py

@@ -37,10 +37,7 @@ def wrap(self):
         from allauth.mfa.recovery_codes import RecoveryCodes
         from allauth.mfa.totp import TOTP
 
-        return {
-            self.Type.TOTP: TOTP,
-            self.Type.RECOVERY_CODES: RecoveryCodes,
-        }[
+        return {self.Type.TOTP: TOTP, self.Type.RECOVERY_CODES: RecoveryCodes,}[
             self.type
         ](self)
 

allauth/mfa/tests/test_views.py

@@ -1,12 +1,15 @@
+from unittest.mock import patch
+
 import django
 from django.conf import settings
+from django.core import mail
 from django.urls import reverse
 
 import pytest
 from pytest_django.asserts import assertFormError
 
 from allauth.account.models import EmailAddress
-from allauth.mfa import app_settings
+from allauth.mfa import app_settings, signals
 from allauth.mfa.adapter import get_adapter
 from allauth.mfa.models import Authenticator
 
@@ -57,6 +60,7 @@ def test_activate_totp_with_unverified_email(
     }
 
 
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
 def test_activate_totp_success(
     auth_client, totp_validation_bypass, user, reauthentication_bypass
 ):
@@ -68,7 +72,10 @@ def test_activate_totp_success(
                 {
                     "code": "123",
                 },
+                **{"HTTP_USER_AGENT": "test"},
             )
+    assert len(mail.outbox) == 1
+    print(mail.outbox[0].subject, mail.outbox[0].body)
     assert resp["location"] == reverse("mfa_view_recovery_codes")
     assert Authenticator.objects.filter(
         user=user, type=Authenticator.Type.TOTP
@@ -215,3 +222,53 @@ def test_totp_login_rate_limit(
                 if is_locked
                 else "Incorrect code.",
             )
+
+
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+def test_notification_on_mfa_activate_totp(
+    auth_client, reauthentication_bypass, totp_validation_bypass
+):
+    with reauthentication_bypass():
+        resp = auth_client.get(reverse("mfa_activate_totp"))
+        with totp_validation_bypass():
+            resp = auth_client.post(
+                reverse("mfa_activate_totp"),
+                {
+                    "code": "123",
+                },
+                **{"HTTP_USER_AGENT": "test"},
+            )
+    assert len(mail.outbox) == 1
+    assert "Totp activated" in mail.outbox[0].subject
+    assert "Totp has been activated." in mail.outbox[0].body
+
+
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+def test_notification_on_mfa_deactivate_totp(
+    auth_client, user_with_totp, user_password
+):
+    resp = auth_client.get(reverse("mfa_deactivate_totp"))
+    assert resp.status_code == 302
+    assert resp["location"].startswith(reverse("account_reauthenticate"))
+    resp = auth_client.post(resp["location"], {"password": user_password})
+    assert resp.status_code == 302
+    resp = auth_client.post(
+        reverse("mfa_deactivate_totp"), **{"HTTP_USER_AGENT": "test"}
+    )
+    assert len(mail.outbox) == 1
+    assert "Totp deactivated" in mail.outbox[0].subject
+    assert "Totp has been deactivated." in mail.outbox[0].body
+
+
+@patch("allauth.account.app_settings.ACCOUNT_EMAIL_NOTIFICATIONS", True)
+def test_notification_on_authenticator_reset(
+    auth_client, user_with_recovery_codes, user_password
+):
+    resp = auth_client.get(reverse("mfa_generate_recovery_codes"))
+    assert resp["location"].startswith(reverse("account_reauthenticate"))
+    resp = auth_client.post(resp["location"], {"password": user_password})
+    assert resp.status_code == 302
+    resp = auth_client.post(resp["location"], **{"HTTP_USER_AGENT": "test"})
+    assert len(mail.outbox) == 1
+    assert "Totp reset" in mail.outbox[0].subject
+    assert "Totp has been reset." in mail.outbox[0].body