add User2FA model + API

df_auth/drf/serializers.py

@@ -19,6 +19,7 @@
 from social_django.utils import load_backend
 
 from ..exceptions import Authentication2FARequiredError
+from ..models import User2FA
 from ..settings import api_settings
 from ..strategy import DRFStrategy
 from ..utils import (
@@ -39,7 +40,7 @@ def build_fields(fields: Dict[str, str], **kwargs: Any) -> Dict[str, serializers
 
 
 def check_user_2fa(user: Optional[AbstractBaseUser], otp: Optional[str]) -> None:
- if user and getattr(user, "is_2fa_enabled", False):
+ if user and hasattr(user, "user_2fa") and user.user_2fa.is_required:
  devices = [d for d in get_otp_devices(user) if d.confirmed]
 
  if not any(d.verify_token(otp) for d in devices):
@@ -420,3 +421,9 @@ def update(self, instance: User, validated_data: Dict[str, Any]) -> User:
  instance.set_password(validated_data["new_password"])
  instance.save()
  return instance
+
+
+class User2FASerializer(serializers.ModelSerializer):
+ class Meta:
+ model = User2FA
+ fields = ["is_required"]

df_auth/drf/urls.py

@@ -19,9 +19,9 @@
 
 router = DefaultRouter()
 router.register("token", TokenViewSet, basename="token")
-router.register("users", UserViewSet, basename="user")
+router.register("users", UserViewSet, basename="users")
 router.register("otp", OTPViewSet, basename="otp")
-router.register("otp-devices", OtpDeviceViewSet, basename="otp-device")
+router.register("otp-devices", OtpDeviceViewSet, basename="otp-devices")
 
 router.register("social", SocialTokenViewSet, basename="social")
 

df_auth/drf/viewsets.py

@@ -1,6 +1,8 @@
 from typing import Any, Iterable, List, Type
 
 from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.core.exceptions import FieldDoesNotExist
 from django.http import HttpRequest, HttpResponse
 from django_otp.models import Device
 from drf_spectacular.types import OpenApiTypes
@@ -16,6 +18,7 @@
 from ..exceptions import (
  DfAuthValidationError,
 )
+from ..models import User2FA
 from ..permissions import IsUnauthenticated, IsUserCreateAllowed
 from ..utils import get_otp_device_models, get_otp_devices
 from .serializers import (
@@ -26,6 +29,7 @@
  SocialTokenObtainSerializer,
  TokenObtainSerializer,
  TokenSerializer,
+ User2FASerializer,
  UserIdentitySerializer,
 )
 
@@ -174,9 +178,17 @@ def get_object(self) -> Any:
  return self.request.user
 
  def perform_create(self, serializer: UserIdentitySerializer) -> None:
- serializer.save(
- created_by=self.request.user if self.request.user.is_authenticated else None
- )
+ User = get_user_model()
+ try:
+ User._meta.get_field("created_by")
+ kwargs = {
+ "created_by": self.request.user
+ if self.request.user.is_authenticated
+ else None
+ }
+ except FieldDoesNotExist:
+ kwargs = {}
+ serializer.save(**kwargs)
 
  @action(
  detail=True,
@@ -188,3 +200,19 @@ def set_password(
  self, request: HttpRequest, *args: Any, **kwargs: Any
  ) -> HttpResponse:
  return super().update(request, *args, **kwargs)
+
+ @action(
+ detail=True,
+ methods=["GET", "PATCH"],
+ serializer_class=User2FASerializer,
+ url_path="two-fa",
+ )
+ def two_fa(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
+ instance = User2FA.objects.get_or_create(user=self.get_object())[0]
+ if request.method == "GET":
+ serializer = self.get_serializer(instance)
+ else:
+ serializer = self.get_serializer(instance, data=request.data, partial=True)
+ serializer.is_valid(raise_exception=True)
+ serializer.save()
+ return response.Response(serializer.data)

df_auth/managers.py

@@ -0,0 +1,41 @@
+from typing import Any
+
+from django.contrib.auth import get_user_model
+from django.contrib.auth.base_user import BaseUserManager
+
+
+class UserManager(BaseUserManager):
+ def create_superuser(self, **extra_fields: Any) -> Any:
+ """Create and save a SuperUser with the given email and password."""
+ extra_fields.setdefault("is_staff", True)
+ extra_fields.setdefault("is_superuser", True)
+
+ if extra_fields.get("is_staff") is not True:
+ raise ValueError("Superuser must have is_staff=True.")
+ if extra_fields.get("is_superuser") is not True:
+ raise ValueError("Superuser must have is_superuser=True.")
+
+ return self._create_user(**extra_fields)
+
+ def _create_user(self, **extra_fields: Any) -> Any:
+ """Create and save a User with the given email and password."""
+ User = get_user_model()
+ password = extra_fields.pop("password", None)
+
+ if not extra_fields.get(User.USERNAME_FIELD):
+ raise ValueError(f"The given {User.USERNAME_FIELD} must be set")
+ if User.EMAIL_FIELD in extra_fields:
+ extra_fields[User.EMAIL_FIELD] = self.normalize_email(
+ extra_fields[User.EMAIL_FIELD]
+ )
+
+ user = self.model(**extra_fields)
+ user.set_password(password)
+ user.save(using=self._db)
+ return user
+
+ def create_user(self, **extra_fields: Any) -> Any:
+ """Create and save a regular User with the given email and password."""
+ extra_fields.setdefault("is_staff", False)
+ extra_fields.setdefault("is_superuser", False)
+ return self._create_user(**extra_fields)

df_auth/migrations/0001_initial.py

@@ -0,0 +1,39 @@
+# Generated by Django 4.2.4 on 2023-09-19 02:36
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+ initial = True
+
+ dependencies = [
+ migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name="User2FA",
+ fields=[
+ (
+ "id",
+ models.AutoField(
+ auto_created=True,
+ primary_key=True,
+ serialize=False,
+ verbose_name="ID",
+ ),
+ ),
+ ("is_required", models.BooleanField(default=False)),
+ (
+ "user",
+ models.OneToOneField(
+ on_delete=django.db.models.deletion.CASCADE,
+ related_name="user_2fa",
+ to=settings.AUTH_USER_MODEL,
+ ),
+ ),
+ ],
+ ),
+ ]

df_auth/models.py

@@ -0,0 +1,11 @@
+from django.conf import settings
+from django.db import models
+
+
+class User2FA(models.Model):
+ user = models.OneToOneField(
+ settings.AUTH_USER_MODEL,
+ on_delete=models.CASCADE,
+ related_name="user_2fa",
+ )
+ is_required = models.BooleanField(default=False)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "django-df-auth"
-version = "1.0.0-alpha.10"
+version = "1.0.0-alpha.11"
 description = "Opinionated Django REST auth endpoints for JWT authentication and social accounts."
 readme = "README.md"
 authors = [{name = "Apexive OSS", email = "open-source@apexive.com"}]

tests/test_app/migrations/0012_alter_user_managers_remove_user_is_2fa_enabled.py

@@ -0,0 +1,20 @@
+# Generated by Django 4.2.4 on 2023-09-19 03:23
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ("test_app", "0011_rename_invited_by_user_created_by"),
+ ]
+
+ operations = [
+ migrations.AlterModelManagers(
+ name="user",
+ managers=[],
+ ),
+ migrations.RemoveField(
+ model_name="user",
+ name="is_2fa_enabled",
+ ),
+ ]

tests/test_app/models.py

@@ -1,48 +1,9 @@
-from typing import Any, List, Optional
+from typing import List
 
-from django.contrib.auth.base_user import BaseUserManager
 from django.contrib.auth.models import AbstractUser
 from django.db import models
 
-
-class UserManager(BaseUserManager):
- """Define a model manager for User model with no username field."""
-
- use_in_migrations = True
-
- @classmethod
- def normalize_email(cls, email: Optional[str]) -> str:
- return super().normalize_email(email).lower()
-
- def _create_user(self, **extra_fields: Any) -> AbstractUser:
- """Create and save a User with the given email and password."""
- user = self.model(**extra_fields)
- if extra_fields.get("email"):
- user.email = self.normalize_email(extra_fields.get("email"))
- if extra_fields.get("password") is not None:
- user.set_password(extra_fields.get("password"))
- user.save(using=self._db)
- return user # type: ignore
-
- def create_user(self, **extra_fields: Any) -> AbstractUser:
- """Create and save a regular User with the given email and password."""
- extra_fields.setdefault("is_staff", False)
- extra_fields.setdefault("is_superuser", False)
- return self._create_user(**extra_fields)
-
- def create_superuser(self, password: str, **extra_fields: Any) -> AbstractUser:
- """Create and save a SuperUser with the given email and password."""
- extra_fields.setdefault("is_staff", True)
- extra_fields.setdefault("is_superuser", True)
-
- if extra_fields.get("is_staff") is not True:
- raise ValueError("Superuser must have is_staff=True.")
- if extra_fields.get("is_superuser") is not True:
- raise ValueError("Superuser must have is_superuser=True.")
-
- extra_fields["password"] = password
-
- return self._create_user(**extra_fields)
+from df_auth.managers import UserManager
 
 
 class User(AbstractUser):
@@ -54,4 +15,3 @@ class User(AbstractUser):
  created_by = models.ForeignKey(
  "self", on_delete=models.SET_NULL, null=True, blank=True
  )
- is_2fa_enabled = models.BooleanField(default=False)

tests/test_app/tests.py

@@ -1,6 +1,7 @@
 import json
 
 import httpretty
+import pytest
 from django.db import IntegrityError
 from django_otp.plugins.otp_email.models import EmailDevice
 from django_otp.plugins.otp_totp.models import TOTPDevice
@@ -9,9 +10,12 @@
 from rest_framework.test import APIClient, APITestCase
 
 from df_auth.exceptions import UserDoesNotExistError
+from df_auth.models import User2FA
 from df_auth.settings import DEFAULTS, api_settings
 from tests.test_app.models import User
 
+pytestmark = pytest.mark.django_db
+
 
 class OtpDeviceViewSetAPITest(APITestCase):
  def setUp(self) -> None:
@@ -525,8 +529,8 @@ def setUp(self) -> None:
  username="testuser",
  password="testpass",
  email="test@te.st",
- is_2fa_enabled=True,
  )
+ User2FA.objects.create(user=self.user, is_required=True)
  self.device = EmailDevice.objects.create(
  user=self.user, name=self.user.email, confirmed=True, email=self.user.email
  )
@@ -619,12 +623,12 @@ def test_social_login_creates_new_user(self) -> None:
  self.assertEqual(user.last_name, self.last_name)
 
  def test_social_login_fails_if_2fa_enabled(self) -> None:
- User.objects.create_user(
+ user = User.objects.create_user(
  username="testuser",
  password="testpass",
  email=self.email,
- is_2fa_enabled=True,
  )
+ User2FA.objects.create(user=user, is_required=True)
  response = self.client.post(
  "/api/v1/auth/social/",
  {
@@ -642,8 +646,8 @@ def test_obtain_social_token_with_otp_for_2fa_user(self) -> None:
  password="testpass",
  email=self.email,
  phone_number="+31612345678",
- is_2fa_enabled=True,
  )
+ User2FA.objects.create(user=user, is_required=True)
  device = TwilioSMSDevice.objects.create(
  user=user, name=user.phone_number, confirmed=True, number=user.phone_number
  )
@@ -746,3 +750,39 @@ def test_invite_not_allowed(self) -> None:
  )
  self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
  self.assertEqual(response.data["errors"][0]["code"], "permission_denied")
+
+
+class User2FAAPITest(APITestCase):
+ def setUp(self) -> None:
+ self.client = APIClient()
+ self.user = User.objects.create_user(
+ username="testuser",
+ password="testpass",
+ )
+ self.client.force_authenticate(self.user)
+
+ def test_user_2fa_retrieve(self) -> None:
+ response = self.client.get(
+ "/api/v1/auth/users/0/two-fa/",
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertFalse(response.data["is_required"])
+
+ def test_user_2fa_update(self) -> None:
+ response = self.client.patch(
+ "/api/v1/auth/users/0/two-fa/",
+ {
+ "is_required": True,
+ },
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertTrue(response.data["is_required"])
+
+
+def test_create_superuser() -> None:
+ User.objects.create_superuser(
+ username="testuser",
+ password="testpass",
+ )