feat(mfa): replace trench based API endpoints, views and forms with allauth implementation DEV-749

dependencies/pip/dev_requirements.txt

@@ -108,6 +108,7 @@ cron-descriptor==1.4.3
 cryptography==42.0.5
     # via
     #   azure-storage-blob
+    #   fido2
     #   jwcrypto
     #   paramiko
     #   pyopenssl
@@ -172,7 +173,7 @@ django==4.2.24
     #   drf-spectacular
     #   jsonfield
     #   model-bakery
-django-allauth==65.11.2
+django-allauth[mfa]==65.11.2
     # via -r dependencies/pip/requirements.in
 django-amazon-ses==4.0.1
     # via -r dependencies/pip/requirements.in
@@ -276,6 +277,8 @@ fabric==3.2.2
     # via -r dependencies/pip/dev_requirements.in
 fakeredis==2.30.1
     # via -r dependencies/pip/dev_requirements.in
+fido2==2.0.0
+    # via django-allauth
 flake8==7.1.1
     # via
     #   -r dependencies/pip/dev_requirements.in
@@ -557,6 +560,8 @@ pyyaml==6.0.1
     # via
     #   drf-spectacular
     #   responses
+qrcode==8.2
+    # via django-allauth
 redis==5.0.3
     # via
     #   celery

dependencies/pip/requirements.in

@@ -28,7 +28,8 @@ dict2xml
 defusedxml
 dj-static
 dj-stripe
-django-allauth
+django-allauth>=65.11.2
+django-allauth[mfa]
 django-braces
 django-celery-beat
 django-constance
@@ -116,4 +117,4 @@ djangorestframework-jsonp
 pandas
 
 # Api Documentation
-drf-spectacular
+drf-spectacular

dependencies/pip/requirements.txt

@@ -92,6 +92,7 @@ cron-descriptor==1.4.3
 cryptography==42.0.5
     # via
     #   azure-storage-blob
+    #   fido2
     #   jwcrypto
     #   pyopenssl
 cssselect==1.2.0
@@ -140,7 +141,7 @@ django==4.2.24
     #   djangorestframework
     #   drf-spectacular
     #   jsonfield
-django-allauth==65.11.2
+django-allauth[mfa]==65.11.2
     # via -r dependencies/pip/requirements.in
 django-amazon-ses==4.0.1
     # via -r dependencies/pip/requirements.in
@@ -230,6 +231,8 @@ drf-spectacular==0.28.0
     # via -r dependencies/pip/requirements.in
 et-xmlfile==1.1.0
     # via openpyxl
+fido2==2.0.0
+    # via django-allauth
 flower==2.0.1
     # via -r dependencies/pip/requirements.in
 frozenlist==1.4.1
@@ -430,6 +433,8 @@ pyyaml==6.0.1
     # via
     #   drf-spectacular
     #   responses
+qrcode==8.2
+    # via django-allauth
 redis==5.0.3
     # via
     #   celery

hub/admin/extend_user.py

@@ -14,7 +14,7 @@
 from django.utils import timezone
 from django.utils.safestring import mark_safe
 
-from kobo.apps.accounts.mfa.models import MfaMethod
+from kobo.apps.accounts.mfa.models import MfaMethodsWrapper
 from kobo.apps.accounts.validators import (
     USERNAME_INVALID_MESSAGE,
     USERNAME_MAX_LENGTH,
@@ -27,7 +27,6 @@
 from kobo.apps.trash_bin.models.account import AccountTrash
 from kobo.apps.trash_bin.utils import move_to_trash
 from kpi.models.asset import AssetDeploymentStatus
-
 from .filters import UserAdvancedSearchFilter
 from .mixins import AdvancedSearchMixin
 
@@ -37,7 +36,7 @@ def validate_superuser_auth(obj) -> bool:
         obj.is_superuser
         and config.SUPERUSER_AUTH_ENFORCEMENT
         and obj.has_usable_password()
-        and not MfaMethod.objects.filter(user=obj, is_active=True).exists()
+        and not MfaMethodsWrapper.objects.filter(user=obj, is_active=True).exists()
     ):
         return False
     return True

hub/tests/test_admin_validators.py

@@ -2,7 +2,7 @@
 from django.test import TestCase
 
 from hub.admin.extend_user import validate_superuser_auth
-from kobo.apps.accounts.mfa.models import MfaMethod
+from kobo.apps.accounts.mfa.models import MfaMethodsWrapper
 from kobo.apps.kobo_auth.shortcuts import User
 
 
@@ -22,5 +22,5 @@ def test_superuser_with_unusable_password(self):
         self.assertTrue(validate_superuser_auth(self.superuser))
 
     def test_superuser_with_mfa_enabled(self):
-        MfaMethod.objects.create(user=self.superuser, is_active=True)
+        MfaMethodsWrapper.objects.create(user=self.superuser, is_active=True)
         self.assertTrue(validate_superuser_auth(self.superuser))

kobo/apps/__init__.py

@@ -1,33 +1,13 @@
-import trench
 from django.apps import AppConfig
 from django.core.checks import Tags, register
 
-import kpi.utils.monkey_patching  # noqa
 from kpi.utils.two_database_configuration_checker import TwoDatabaseConfigurationChecker
 
 
 class KpiConfig(AppConfig):
     name = 'kpi'
 
     def ready(self, *args, **kwargs):
-        # These imports cannot be at the top until the app is loaded.
-        from kobo.apps.accounts.mfa.command import (
-            create_mfa_method_command,
-            deactivate_mfa_method_command,
-        )
-
-        # Monkey-patch `django-trench` to avoid duplicating lots of code in views,
-        # and serializers just for few line changes.
-        # Changed behaviours:
-        # 1. Stop blocking deactivation of primary method
-        trench.command.deactivate_mfa_method.deactivate_mfa_method_command = (
-            deactivate_mfa_method_command
-        )
-        # 2. Resetting secret on reactivation
-        trench.command.create_mfa_method.create_mfa_method_command = (
-            create_mfa_method_command
-        )
-
         # Load all schema extension modules to register them
         import kpi.schema_extensions.imports  # noqa F401
 

kobo/apps/accounts/adapter.py

@@ -2,22 +2,11 @@
 from allauth.account.forms import SignupForm
 from constance import config
 from django.conf import settings
-from django.contrib.auth import REDIRECT_FIELD_NAME, login
 from django.db import transaction
-from django.shortcuts import resolve_url
-from django.template.response import TemplateResponse
 from django.utils import timezone
-from trench.utils import get_mfa_model, user_token_generator
-
-from .mfa.forms import MfaTokenForm
-from .mfa.models import MfaAvailableToUser
-from .mfa.permissions import mfa_allowed_for_user
-from .mfa.views import MfaTokenView
-from .utils import user_has_inactive_paid_subscription
 
 
 class AccountAdapter(DefaultAccountAdapter):
-
     def is_open_for_signup(self, request):
         return config.REGISTRATION_OPEN
 
@@ -26,44 +15,6 @@ def login(self, request, user):
         user.backend = settings.AUTHENTICATION_BACKENDS[0]
         super().login(request, user)
 
-    def pre_login(self, request, user, **kwargs):
-
-        if parent_response := super().pre_login(request, user, **kwargs):
-            # A response from the parent means the login process must be
-            # interrupted, e.g. due to the user being inactive or not having
-            # validated their email address
-            return parent_response
-
-        # If MFA is activated and allowed for the user, display the token form before letting them in
-        mfa_active = (
-            get_mfa_model().objects.filter(is_active=True, user=user).exists()
-        )
-        mfa_allowed = mfa_allowed_for_user(user)
-        inactive_subscription = user_has_inactive_paid_subscription(
-            user.username
-        )
-        if mfa_active and (mfa_allowed or inactive_subscription):
-            ephemeral_token_cache = user_token_generator.make_token(user)
-            mfa_token_form = MfaTokenForm(
-                initial={'ephemeral_token': ephemeral_token_cache}
-            )
-
-            next_url = kwargs.get('redirect_url') or resolve_url(
-                settings.LOGIN_REDIRECT_URL
-            )
-
-            context = {
-                REDIRECT_FIELD_NAME: next_url,
-                'view': MfaTokenView,
-                'form': mfa_token_form,
-            }
-
-            return TemplateResponse(
-                request=request,
-                template='mfa_token.html',
-                context=context,
-            )
-
     def save_user(self, request, user, form, commit=True):
         # Compare allauth SignupForm with our custom field
         standard_fields = set(SignupForm().fields.keys())

kobo/apps/accounts/mfa/adapter.py

@@ -0,0 +1,86 @@
+from typing import Optional
+
+from allauth.mfa.adapter import DefaultMFAAdapter
+from allauth.mfa.models import Authenticator
+from constance import config
+from django.conf import settings
+
+from ..utils import user_has_inactive_paid_subscription
+from .models import MfaMethod, MfaMethodsWrapper
+from .permissions import mfa_allowed_for_user
+
+
+class MfaAdapter(DefaultMFAAdapter):
+
+    def is_mfa_enabled(self, user, types=None) -> bool:
+        # NOTE: This is a temporary thing. We are migrating users to the allauth tables
+        # When the migration is done it won't be necessary.
+        self.migrate_user(user)
+        mfa_active_super = super().is_mfa_enabled(user, types)
+        mfa_active = (
+            mfa_active_super
+            and MfaMethodsWrapper.objects.filter(user=user, is_active=True).first()
+            is not None
+        )
+        mfa_allowed = mfa_allowed_for_user(user)
+        inactive_subscription = user_has_inactive_paid_subscription(user.username)
+        return mfa_active and (mfa_allowed or inactive_subscription)
+
+    def get_totp_label(self, user) -> str:
+        """Returns the label used for representing the given user in a TOTP QR
+        code.
+        """
+        return f'{config.MFA_ISSUER_NAME}-{user.username}'
+
+    def get_totp_issuer(self) -> str:
+        """Returns the TOTP issuer name that will be contained in the TOTP QR
+        code.
+        """
+        return config.MFA_ISSUER_NAME
+
+    def migrate_user(
+        self, user: settings.AUTH_USER_MODEL, mfa_method: MfaMethod = None
+    ) -> Optional[MfaMethodsWrapper]:
+        """Migrate user MFA data from trench tables to allauth tables"""
+        if not mfa_method:
+            mfa_method = (
+                MfaMethod.objects.filter(name='app', user=user, is_active=True)
+                .order_by('is_primary')
+                .first()
+            )
+        if not mfa_method:
+            return
+        authenticators = Authenticator.objects.filter(user_id=user.id)
+        types_ok = {a.type for a in authenticators} == {
+            Authenticator.Type.TOTP,
+            Authenticator.Type.RECOVERY_CODES,
+        }
+        # If allauth MFA Authenticators already exist, exit
+        if types_ok:
+            return
+        for authenticator in authenticators:
+            authenticator.delete()
+
+        encrypted_secret = self.encrypt(mfa_method.secret)
+        totp_authenticator = Authenticator.objects.create(
+            user_id=mfa_method.user_id,
+            type=Authenticator.Type.TOTP,
+            data={'secret': encrypted_secret},
+        )
+        recovery_codes = Authenticator.objects.create(
+            user_id=mfa_method.user_id,
+            type=Authenticator.Type.RECOVERY_CODES,
+            data={
+                'migrated_codes': [self.encrypt(c) for c in mfa_method.backup_codes],
+                'used_mask': 0,
+            },
+        )
+        mfa_method_wrapper = MfaMethodsWrapper.objects.create(
+            name='app',
+            user=user,
+            is_active=True,
+            totp=totp_authenticator,
+            recovery_codes=recovery_codes,
+            secret=encrypted_secret,
+        )
+        return mfa_method_wrapper

kobo/apps/accounts/mfa/admin.py

@@ -2,13 +2,13 @@
 from django.contrib import admin
 
 from .models import (
-    TrenchMFAMethod,
+    ExtendedTrenchMfaMethodAdmin,
     MfaAvailableToUser,
     MfaAvailableToUserAdmin,
     MfaMethod,
-    MfaMethodAdmin,
+    TrenchMFAMethod,
 )
 
 admin.site.unregister(TrenchMFAMethod)
-admin.site.register(MfaMethod, MfaMethodAdmin)
+admin.site.register(MfaMethod, ExtendedTrenchMfaMethodAdmin)
 admin.site.register(MfaAvailableToUser, MfaAvailableToUserAdmin)

kobo/apps/accounts/mfa/apps.py

@@ -4,6 +4,7 @@
 class MfaAppConfig(AppConfig):
     name = 'kobo.apps.accounts.mfa'
     verbose_name = 'Multi-factor authentication'
+    label = 'accounts_mfa'
 
     def ready(self):
         from . import signals  # noqa F401