feat(mfa): Signals

ChangeLog.rst

@@ -9,6 +9,9 @@ Note worthy changes
 
 - The MFA authenticator model is now registed with the Django admin.
 
+- Added MFA signals emitted when authenticators are added, removed or (in case
+  of recovery codes) reset.
+
 
 0.58.2 (2023-11-06)
 *******************

allauth/mfa/models.py

@@ -6,9 +6,12 @@
 
 class AuthenticatorManager(models.Manager):
     def delete_dangling_recovery_codes(self, user):
+        deleted_authenticator = None
         qs = Authenticator.objects.filter(user=user)
         if not qs.exclude(type=Authenticator.Type.RECOVERY_CODES).exists():
+            deleted_authenticator = qs.first()
             qs.delete()
+        return deleted_authenticator
 
 
 class Authenticator(models.Model):

allauth/mfa/signals.py

@@ -1,9 +1,23 @@
+from django.dispatch import Signal
 from django.forms import ValidationError
 
 from allauth.mfa.adapter import get_adapter
 from allauth.mfa.utils import is_mfa_enabled
 
 
+# Emitted when an authenticator is added.
+# Arguments: request, user, authenticator
+authenticator_added = Signal()
+
+# Emitted when an authenticator is removed.
+# Arguments: request, user, authenticator
+authenticator_removed = Signal()
+
+# Emitted when an authenticator is reset (e.g. recovery codes regenerated).
+# Arguments: request, user, authenticator
+authenticator_reset = Signal()
+
+
 def on_add_email(sender, email, user, **kwargs):
     if is_mfa_enabled(user):
         adapter = get_adapter()

allauth/mfa/totp.py

@@ -102,7 +102,6 @@ def activate(cls, user, secret):
 
     def deactivate(self):
         self.instance.delete()
-        Authenticator.objects.delete_dangling_recovery_codes(self.instance.user)
 
     def validate_code(self, code):
         secret = decrypt(self.instance.data["secret"])

allauth/mfa/views.py

@@ -14,7 +14,7 @@
 from allauth.account.adapter import get_adapter as get_account_adapter
 from allauth.account.decorators import reauthentication_required
 from allauth.account.stages import LoginStageController
-from allauth.mfa import app_settings, totp
+from allauth.mfa import app_settings, signals, totp
 from allauth.mfa.adapter import get_adapter
 from allauth.mfa.forms import ActivateTOTPForm, AuthenticateForm
 from allauth.mfa.models import Authenticator
@@ -103,8 +103,14 @@ def get_form_kwargs(self):
         return ret
 
     def form_valid(self, form):
-        totp.TOTP.activate(self.request.user, form.secret)
-        RecoveryCodes.activate(self.request.user)
+        totp_auth = totp.TOTP.activate(self.request.user, form.secret)
+        rc_auth = RecoveryCodes.activate(self.request.user)
+        for auth in [totp_auth, rc_auth]:
+            signals.authenticator_added.send(
+                sender=Authenticator,
+                user=self.request.user,
+                authenticator=auth.instance,
+            )
         adapter = get_account_adapter(self.request)
         adapter.add_message(
             self.request, messages.SUCCESS, "mfa/messages/totp_activated.txt"
@@ -141,6 +147,16 @@ def _dispatch(self, request, *args, **kwargs):
 
     def form_valid(self, form):
         self.authenticator.wrap().deactivate()
+        rc_auth = Authenticator.objects.delete_dangling_recovery_codes(
+            self.authenticator.user
+        )
+        for auth in [self.authenticator, rc_auth]:
+            if auth:
+                signals.authenticator_removed.send(
+                    sender=Authenticator,
+                    user=self.request.user,
+                    authenticator=auth,
+                )
         adapter = get_account_adapter(self.request)
         adapter.add_message(
             self.request, messages.SUCCESS, "mfa/messages/totp_deactivated.txt"
@@ -161,11 +177,14 @@ def form_valid(self, form):
         Authenticator.objects.filter(
             user=self.request.user, type=Authenticator.Type.RECOVERY_CODES
         ).delete()
-        RecoveryCodes.activate(self.request.user)
+        rc_auth = RecoveryCodes.activate(self.request.user)
         adapter = get_account_adapter(self.request)
         adapter.add_message(
             self.request, messages.SUCCESS, "mfa/messages/recovery_codes_generated.txt"
         )
+        signals.authenticator_reset.send(
+            sender=Authenticator, user=self.request.user, authenticator=rc_auth.instance
+        )
         return super().form_valid(form)
 
     def get_context_data(self, **kwargs):