Multi-Factor Authentication

ChangeLog.rst

@@ -4,14 +4,17 @@
 Note worthy changes
 -------------------
 
-- ...
+- Added builtin support for Two-Factor Authentication via the ``allauth.mfa`` app.
 
 
 Backwards incompatible changes
 ------------------------------
 
 - Dropped support for Django 3.1.
 
+- The ``"allauth.account.middleware.AccountMiddleware"`` middleware is required to be present
+  in your ``settings.MIDDLEWARE``.
+
 
 0.55.0 (2023-08-22)
 *******************

allauth/account/adapter.py

@@ -28,7 +28,7 @@
 from django.utils.encoding import force_str
 from django.utils.translation import gettext_lazy as _
 
-from allauth import ratelimit
+from allauth import app_settings as allauth_app_settings, ratelimit
 from allauth.account import signals
 from allauth.account.app_settings import EmailVerificationMethod
 from allauth.utils import (
@@ -53,6 +53,7 @@ class DefaultAccountAdapter(object):
             "Too many failed login attempts. Try again later."
         ),
         "email_taken": _("A user is already registered with this email address."),
+        "incorrect_password": _("Incorrect password."),
     }
 
     def __init__(self, request=None):
@@ -449,6 +450,8 @@ def post_login(
         return response
 
     def login(self, request, user):
+        from allauth.account.utils import record_authentication
+
         # HACK: This is not nice. The proper Django way is to use an
         # authentication backend
         if not hasattr(user, "backend"):
@@ -467,6 +470,7 @@ def login(self, request, user):
             backend_path = ".".join([backend.__module__, backend.__class__.__name__])
             user.backend = backend_path
         django_login(request, user)
+        record_authentication(request, user)
 
     def logout(self, request):
         django_logout(request)
@@ -667,6 +671,12 @@ def generate_emailconfirmation_key(self, email):
         key = get_random_string(64).lower()
         return key
 
+    def get_login_stages(self):
+        ret = []
+        if allauth_app_settings.MFA_ENABLED:
+            ret.append("allauth.mfa.stages.AuthenticateStage")
+        return ret
+
 
 def get_adapter(request=None):
     return import_attribute(app_settings.ADAPTER)(request)

allauth/account/app_settings.py

@@ -189,6 +189,8 @@ def RATE_LIMITS(self):
             "reset_password": "20/m",
             # Rate limit measured per individual email address
             "reset_password_email": "5/m",
+            # Reauthentication for users already logged in)
+            "reauthenticate": "10/m",
             # Password reset (the view the password reset email links to).
             "reset_password_from_key": "20/m",
             # Signups.
@@ -383,6 +385,10 @@ def PASSWORD_RESET_TOKEN_GENERATOR(self):
             token_generator = EmailAwarePasswordResetTokenGenerator
         return token_generator
 
+    @property
+    def REAUTHENTICATION_TIMEOUT(self):
+        return self._setting("REAUTHENTICATION_TIMEOUT", 300)
+
 
 _app_settings = AppSettings("ACCOUNT_")
 

allauth/account/apps.py

@@ -1,8 +1,17 @@
 from django.apps import AppConfig
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import gettext_lazy as _
 
 
 class AccountConfig(AppConfig):
     name = "allauth.account"
     verbose_name = _("Accounts")
     default_auto_field = "django.db.models.AutoField"
+
+    def ready(self):
+        required_mw = "allauth.account.middleware.AccountMiddleware"
+        if required_mw not in settings.MIDDLEWARE:
+            raise ImproperlyConfigured(
+                f"{required_mw} must be added to settings.MIDDLEWARE"
+            )

allauth/account/decorators.py

@@ -1,9 +1,14 @@
+from functools import wraps
+
 from django.contrib.auth import REDIRECT_FIELD_NAME
 from django.contrib.auth.decorators import login_required
+from django.http import HttpResponseRedirect
 from django.shortcuts import render
+from django.urls import reverse
+from django.utils.http import urlencode
 
 from .models import EmailAddress
-from .utils import send_email_confirmation
+from .utils import did_recently_authenticate, send_email_confirmation
 
 
 def verified_email_required(
@@ -36,3 +41,33 @@ def _wrapped_view(request, *args, **kwargs):
     if function:
         return decorator(function)
     return decorator
+
+
+def reauthentication_required(function=None, redirect_field_name=REDIRECT_FIELD_NAME):
+    def decorator(view_func):
+        @wraps(view_func)
+        def _wrapper_view(request, *args, **kwargs):
+            path = request.get_full_path()
+            if request.user.is_anonymous:
+                redirect_url = (
+                    reverse("account_login")
+                    + "?"
+                    + urlencode({redirect_field_name: path})
+                )
+                return HttpResponseRedirect(redirect_url)
+
+            if not did_recently_authenticate(request):
+                redirect_url = (
+                    reverse("account_reauthenticate")
+                    + "?"
+                    + urlencode({redirect_field_name: path})
+                )
+                return HttpResponseRedirect(redirect_url)
+
+            return view_func(request, *args, **kwargs)
+
+        return _wrapper_view
+
+    if function:
+        return decorator(function)
+    return decorator

allauth/account/forms.py

@@ -461,6 +461,8 @@ class AddEmailForm(UserForm):
     )
 
     def clean_email(self):
+        from allauth.account import signals
+
         value = self.cleaned_data["email"]
         value = get_adapter().clean_email(value)
         errors = {
@@ -478,6 +480,12 @@ def clean_email(self):
             raise forms.ValidationError(
                 errors["max_email_addresses"] % app_settings.MAX_EMAIL_ADDRESSES
             )
+
+        signals._add_email.send(
+            sender=self.user.__class__,
+            email=value,
+            user=self.user,
+        )
         return value
 
     def save(self, request):
@@ -646,3 +654,19 @@ def clean(self):
             raise forms.ValidationError(self.error_messages["token_invalid"])
 
         return cleaned_data
+
+
+class ReauthenticateForm(forms.Form):
+    password = PasswordField(label=_("Password"), autocomplete="current-password")
+
+    def __init__(self, *args, **kwargs):
+        self.user = kwargs.pop("user")
+        super().__init__(*args, **kwargs)
+
+    def clean_password(self):
+        password = self.cleaned_data.get("password")
+        if not self.user.check_password(password):
+            raise forms.ValidationError(
+                get_adapter().error_messages["incorrect_password"]
+            )
+        return password

allauth/account/middleware.py

@@ -0,0 +1,19 @@
+from django.conf import settings
+
+
+class AccountMiddleware:
+    def __init__(self, get_response):
+        self.get_response = get_response
+        # One-time configuration and initialization.
+
+    def __call__(self, request):
+        response = self.get_response(request)
+        self._remove_dangling_login(request, response)
+        return response
+
+    def _remove_dangling_login(self, request, response):
+        if request.path.startswith(settings.STATIC_URL):
+            return
+        if not getattr(request, "_account_login_accessed", False):
+            if "account_login" in request.session:
+                request.session.pop("account_login")

allauth/account/models.py

@@ -1,5 +1,6 @@
 import datetime
 
+from django.contrib.auth import get_user_model
 from django.core import signing
 from django.db import models
 from django.db.models import Index, Q
@@ -12,7 +13,6 @@
 from . import app_settings, signals
 from .adapter import get_adapter
 from .managers import EmailAddressManager, EmailConfirmationManager
-from .utils import user_email
 
 
 class EmailAddress(models.Model):
@@ -72,6 +72,8 @@ def set_as_primary(self, conditional=False):
         """Marks the email address as primary. In case of `conditional`, it is
         only marked as primary if there is no other primary email address set.
         """
+        from allauth.account.utils import user_email
+
         old_primary = EmailAddress.objects.get_primary(self.user)
         if old_primary:
             if conditional:
@@ -92,6 +94,8 @@ def send_confirmation(self, request=None, signup=False):
         return confirmation
 
     def remove(self):
+        from allauth.account.utils import user_email
+
         self.delete()
         if user_email(self.user) == self.email:
             alt = (
@@ -191,3 +195,81 @@ def from_key(cls, key):
         ):
             ret = None
         return ret
+
+
+class Login:
+    """
+    Represents a user that is in the process of logging in.
+    """
+
+    def __init__(
+        self,
+        user,
+        email_verification,
+        redirect_url=None,
+        signal_kwargs=None,
+        signup=False,
+        email=None,
+        state=None,
+    ):
+        self.user = user
+        self.email_verification = email_verification
+        self.redirect_url = redirect_url
+        self.signal_kwargs = signal_kwargs
+        self.signup = signup
+        self.email = email
+        self.state = {} if state is None else state
+
+    def serialize(self):
+        from allauth.account.utils import user_pk_to_url_str
+
+        # :-( Knowledge of the `socialaccount` is entering the `account` app.
+        signal_kwargs = self.signal_kwargs
+        if signal_kwargs is not None:
+            sociallogin = signal_kwargs.get("sociallogin")
+            if sociallogin is not None:
+                signal_kwargs = signal_kwargs.copy()
+                signal_kwargs["sociallogin"] = sociallogin.serialize()
+
+        data = {
+            "user_pk": user_pk_to_url_str(self.user),
+            "email_verification": self.email_verification,
+            "signup": self.signup,
+            "redirect_url": self.redirect_url,
+            "email": self.email,
+            "signal_kwargs": signal_kwargs,
+            "state": self.state,
+        }
+        return data
+
+    @classmethod
+    def deserialize(cls, data):
+        from allauth.account.utils import url_str_to_user_pk
+        from allauth.socialaccount.models import SocialLogin
+
+        user = (
+            get_user_model()
+            .objects.filter(pk=url_str_to_user_pk(data["user_pk"]))
+            .first()
+        )
+        if user is None:
+            raise ValueError()
+        try:
+            # :-( Knowledge of the `socialaccount` is entering the `account` app.
+            signal_kwargs = data["signal_kwargs"]
+            if signal_kwargs is not None:
+                sociallogin = signal_kwargs.get("sociallogin")
+                if sociallogin is not None:
+                    signal_kwargs = signal_kwargs.copy()
+                    signal_kwargs["sociallogin"] = SocialLogin.deserialize(sociallogin)
+
+            return Login(
+                user=user,
+                email_verification=data["email_verification"],
+                redirect_url=data["redirect_url"],
+                signup=data["signup"],
+                signal_kwargs=signal_kwargs,
+                state=data["state"],
+            )
+        except KeyError:
+            raise ValueError()

allauth/account/signals.py

@@ -28,3 +28,6 @@
 email_added = Signal()
 # Provides the arguments "request", "user", "email_address"
 email_removed = Signal()
+
+# Internal/private signal.
+_add_email = Signal()