Merge pull request #5322 from akvo/enforce-2fa

Implement enforcing 2FA

akvo/rest/serializers/user.py

@@ -71,6 +71,7 @@ class Meta:
             'organisations',
             'approved_employments',
             'api_key',
+            'enforce_2fa',
             'otp_keys',
             'legacy_org',
             'programs',

akvo/rsr/admin.py

@@ -264,7 +264,8 @@ class OrganisationAdmin(TimestampsAdminDisplayMixin, ObjectPermissionsModelAdmin
     fieldsets = (
         (_('General information'),
             {'fields': ('name', 'long_name', 'iati_org_id', 'description', 'new_organisation_type',
-                        'logo', 'language', 'currency', 'iati_prefixes', 'password_policy')}),
+                        'logo', 'language', 'currency', 'iati_prefixes', 'password_policy',
+                        'enforce_2fa')}),
         (_('Contact information'),
             {'fields': ('url', 'facebook', 'twitter', 'linkedin', 'phone', 'mobile', 'fax',
                         'contact_person', 'contact_email', )}),
@@ -541,7 +542,7 @@ class UserAdmin(DjangoUserAdmin):
         (None, {'fields': ('username', 'email', 'password')}),
         (_('Personal info'), {'fields': ('first_name', 'last_name')}),
         (_('Permissions'), {
-            'fields': ('is_active', 'is_staff', 'is_admin', 'is_support', 'is_superuser')
+            'fields': ('is_active', 'is_staff', 'is_admin', 'is_support', 'is_superuser', 'enforce_2fa')
         }),
         (_('Important dates'), {'fields': ('last_login', 'date_joined')}),
     )

akvo/rsr/decorators.py

@@ -11,6 +11,9 @@
 
 from django.shortcuts import get_object_or_404
 from django.http import HttpResponse
+from django.contrib.auth.decorators import user_passes_test
+from django.urls import reverse_lazy
+from two_factor.utils import default_device
 
 from akvo.rsr.models import Project
 
@@ -53,3 +56,24 @@ def wrapper(request, *args, **kwargs):
         return response
 
     return wrapper
+
+
+def two_factor_required(view=None, login_url=None):
+    """
+    If a user has enforced_2fa attribute as True, the user should enable 2FA
+    for accessing the page or user will be redirected to 2FA setup page.
+    """
+    if login_url is None:
+        login_url = reverse_lazy('two_factor:setup')
+
+    def test(user):
+        if not user.is_authenticated:
+            return True
+        has_default_device = bool(default_device(user))
+        if user.enforce_2fa and not has_default_device:
+            return False
+        return True
+
+    decorator = user_passes_test(test, login_url=login_url)
+
+    return decorator if (view is None) else decorator(view)

akvo/rsr/management/commands/org_related_users.py

@@ -0,0 +1,26 @@
+import sys
+from django.core.management.base import BaseCommand, CommandParser
+from tablib import Dataset
+from akvo.rsr.models import Organisation
+from akvo.rsr.usecases.toggle_org_enforce_2fa import find_related_users
+
+
+class Command(BaseCommand):
+    help = "Script to get list of users who are employed by or have role access to projects of the given organisation"
+
+    def add_arguments(self, parser: CommandParser):
+        parser.add_argument('org_id', type=int)
+
+    def handle(self, *args, **options):
+        try:
+            org = Organisation.objects.get(id=options['org_id'])
+        except Organisation.DoesNotExist:
+            self.stderr.write("Organisation not found")
+            sys.exit(1)
+
+        users = find_related_users(org)
+        tbl = Dataset()
+        tbl.headers = ['email', 'name']
+        for user in users:
+            tbl.append([user.email, user.get_full_name()])
+        self.stdout.write(tbl.export('csv'))

akvo/rsr/migrations/0229_organisation_enforce_2fa.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.18 on 2024-02-15 08:15
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('rsr', '0228_delete_emailreportjob'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='organisation',
+            name='enforce_2fa',
+            field=models.BooleanField(default=False, help_text='Enfore related users (through employment or project access) to enable their 2FA', verbose_name='Enforce 2-Factor-Authentication'),
+        ),
+    ]

akvo/rsr/migrations/0230_user_enforce_2fa.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.18 on 2024-02-19 07:42
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('rsr', '0229_organisation_enforce_2fa'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='user',
+            name='enforce_2fa',
+            field=models.BooleanField(default=False, help_text='Enfore related users (through employment or project access) to enable their 2FA', verbose_name='Enforce 2-Factor-Authentication'),
+        ),
+    ]

akvo/rsr/models/organisation.py

@@ -12,11 +12,13 @@
 from django.urls import reverse
 from django.utils.functional import cached_property
 from django.utils.translation import ugettext_lazy as _
+from django_q.tasks import async_task
 
 from sorl.thumbnail.fields import ImageField
 from akvo.password_policy.models import PolicyConfig
 
 from akvo.utils import codelist_choices, codelist_name, rsr_image_path
+from akvo.rsr.usecases.toggle_org_enforce_2fa import toggle_enfore_2fa
 
 from ..mixins import TimestampsMixin
 from ..fields import ValidXMLCharField, ValidXMLTextField
@@ -171,6 +173,10 @@ def org_type_from_iati_type(cls, iati_type):
     public_iati_file = models.BooleanField(
         _('Show latest exported IATI file on organisation page.'), default=True
     )
+    enforce_2fa = models.BooleanField(
+        _('Enforce 2-Factor-Authentication'), default=False,
+        help_text=_('Enfore related users (through employment or project access) to enable their 2FA'),
+    )
     password_policy = models.ForeignKey(
         PolicyConfig, null=True, blank=True, on_delete=models.SET_NULL,
         help_text='Password policies config for employees of this organization.'
@@ -202,6 +208,18 @@ def cacheable_url(self):
     def canonical_name(self):
         return self.long_name or self.name
 
+    __original_enforce_2fa = False
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.__original_enforce_2fa = self.enforce_2fa
+
+    def save(self, *args, **kwargs):
+        if self.enforce_2fa != self.__original_enforce_2fa:
+            async_task(toggle_enfore_2fa, self)
+        super().save(*args, **kwargs)
+        self.__original_enforce_2fa = self.enforce_2fa
+
     def clean(self):
         """Organisations can only be saved when we're sure that they do not exist already."""
         validation_errors = {}

akvo/rsr/models/user.py

@@ -90,6 +90,10 @@ class User(AbstractBaseUser, PermissionsMixin):
                     'willing to receive notifications when a new user registers for '
                     'their organisation.')
     )
+    enforce_2fa = models.BooleanField(
+        _('Enforce 2-Factor-Authentication'), default=False,
+        help_text=_('Enfore related users (through employment or project access) to enable their 2FA'),
+    )
     date_joined = models.DateTimeField(_('date joined'), default=timezone.now)
     organisations = models.ManyToManyField(
         'Organisation', verbose_name=_('organisations'), through='Employment',

akvo/rsr/spa/app/root.jsx

@@ -32,6 +32,10 @@ const isJWTView = () => {
   return reqToken !== null
 }
 
+const shouldEnable2FA = function (data) {
+  return data.enforce2fa && !data.otpKeys?.totp ? true : false
+}
+
 const Root = ({ dispatch }) => {
   const jwtView = isJWTView()
   const [data, loading] = useFetch('/me')
@@ -44,8 +48,13 @@ const Root = ({ dispatch }) => {
           scope.setUser({ id, email })
         })
       }
+      if (shouldEnable2FA(data)) {
+        window.location.href = `/en/account/two_factor/setup/?next=${window.location.href}`
+      }
+    }
+    else {
+      window.location.href = `/en/sign_in/?next=${window.location.href}`
     }
-    else window.location.href = `/en/sign_in/?next=${window.location.href}`
   }
   return (
     <Router basename="/my-rsr">

akvo/rsr/tests/test_decorators.py

@@ -0,0 +1,54 @@
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
+from django.http import HttpResponse
+from django.test import RequestFactory
+from django_otp.plugins.otp_totp.models import TOTPDevice
+from akvo.rsr.tests.base import BaseTestCase
+from akvo.rsr.decorators import two_factor_required
+
+
+def func_view(request):
+    return HttpResponse(request.user)
+
+
+class TwoFactorRequiredDecoratorTestCase(BaseTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.request = RequestFactory().get('')
+        self.user = self.create_user('test@akvo.org')
+        self.request.user = self.user
+
+    def make_default_device(self, user):
+        return TOTPDevice.objects.create(user=user, name='default')
+
+    def enforce_2fa(self, user):
+        user.enforce_2fa = True
+        user.save()
+        return get_user_model().objects.get(id=user.id)
+
+    def test_user_no_device_and_not_enforced_2fa(self):
+        response = two_factor_required(func_view)(self.request)
+        self.assertEqual(response.status_code, 200)
+
+    def test_user_no_device_and_enforced_2fa(self):
+        self.enforce_2fa(self.user)
+        response = two_factor_required(func_view)(self.request)
+        self.assertEqual(response.status_code, 302)
+
+    def test_user_with_device_not_and_enforced_2fa(self):
+        self.make_default_device(self.user)
+        response = two_factor_required(func_view)(self.request)
+        self.assertEqual(response.status_code, 200)
+
+    def test_user_with_device_and_enforced_2fa(self):
+        user = self.enforce_2fa(self.user)
+        self.make_default_device(user)
+        response = two_factor_required(func_view)(self.request)
+        self.assertEqual(response.status_code, 200)
+
+    def test_anonymous_user(self):
+        user = AnonymousUser()
+        request = RequestFactory().get('')
+        request.user = user
+        response = two_factor_required(func_view)(request)
+        self.assertEqual(response.status_code, 200)

akvo/rsr/tests/usecases/test_toggle_enforce_2fa.py

@@ -0,0 +1,56 @@
+from akvo.rsr.tests.base import BaseTestCase
+from akvo.rsr.models import User
+from akvo.rsr.usecases.toggle_org_enforce_2fa import find_related_users, toggle_enfore_2fa
+
+
+class FindRelatedUsersTest(BaseTestCase):
+    def setUp(self):
+        super().setUp()
+        self.org = self.create_organisation('Akvo')
+        self.user = self.create_user('test@akvo.org')
+        self.project = self.create_project('Acme')
+
+    def test_found_employee(self):
+        self.make_employment(self.user, self.org, 'Users')
+
+        found = find_related_users(self.org).values_list('email', flat=True)
+
+        self.assertIn(self.user.email, found)
+
+    def test_found_project_access(self):
+        self.make_project_role(self.user, self.project, 'Users')
+        self.make_partner(self.project, self.org)
+
+        found = find_related_users(self.org).values_list('email', flat=True)
+
+        self.assertIn(self.user.email, found)
+
+
+class ToggleEnforce2FATest(BaseTestCase):
+    def setUp(self):
+        super().setUp()
+        self.org = self.create_organisation('Akvo')
+        self.user = self.create_user('test@akvo.org')
+        self.make_employment(self.user, self.org, 'Users')
+
+    def test_toggle_true(self):
+        self.org.enforce_2fa = True
+        self.org.save()
+
+        toggle_enfore_2fa(self.org)
+
+        self.assertTrue(User.objects.get(id=self.user.id).enforce_2fa)
+
+    def test_toggle_false(self):
+        # Force changes
+        self.org.enforce_2fa = True
+        self.org.save()
+        self.org.enforce_2fa = False
+        self.org.save()
+
+        self.user.enforce_2fa = True
+        self.user.save()
+
+        toggle_enfore_2fa(self.org)
+
+        self.assertFalse(User.objects.get(id=self.user.id).enforce_2fa)

akvo/rsr/usecases/toggle_org_enforce_2fa.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+from django.apps import apps
+
+from django.db.models import QuerySet
+
+if TYPE_CHECKING:
+    from akvo.rsr.models import Organisation
+
+
+def toggle_enfore_2fa(org: Organisation):
+    User = apps.get_model('rsr.User')
+    user_ids = find_related_users(org).values_list('id', flat=True)
+    User.objects.filter(id__in=user_ids).update(enforce_2fa=org.enforce_2fa)
+
+
+def find_related_users(organisation: Organisation) -> QuerySet:
+    User = apps.get_model('rsr.User')
+    Project = apps.get_model('rsr.Project')
+    employees = User.objects.filter(employers__organisation=organisation)
+
+    programs = Project.objects.exclude(projecthierarchy__isnull=True).filter(
+        primary_organisation=organisation
+    )
+    program_projects = set(programs.values_list("id", flat=True))
+    for program in programs:
+        program_projects = program_projects | set(
+            program.descendants().values_list("id", flat=True)
+        )
+    program_users = User.objects.filter(projectrole__project__in=program_projects)
+    other_projects = Project.objects.filter(primary_organisation=organisation).exclude(
+        id__in=program_projects
+    )
+    other_users = User.objects.filter(projectrole__project__in=other_projects)
+
+    return employees.union(program_users).union(other_users)

akvo/urls.py

@@ -20,6 +20,7 @@
 from rest_framework_swagger.views import get_swagger_view
 from two_factor.urls import urlpatterns as two_factor_urls
 
+from akvo.rsr.decorators import two_factor_required
 from akvo.rsr import views
 from akvo.rsr.views import account
 from akvo.rsr.views import my_rsr
@@ -134,11 +135,11 @@
         my_rsr.my_details, name='my_details'),
 
     url(r'^myrsr/projects/$',
-        RedirectView.as_view(url='/my-rsr/'),
+        two_factor_required(RedirectView.as_view(url='/my-rsr/')),
         name='my_projects'),
 
     url(r'^myrsr/project_editor/(?P<project_id>\d+)/$',
-        RedirectView.as_view(url='/my-rsr/projects/%(project_id)s/'),
+        two_factor_required(RedirectView.as_view(url='/my-rsr/projects/%(project_id)s/')),
         name='project_editor'),
 
     url(r'^translations.json$', translations.myrsr, name='myrsr-translations'),