feat(auth): switch to OIDC authentication

{{cookiecutter.project_name}}/docker-compose.override.yml

@@ -22,3 +22,5 @@ services:
       ]
     environment:
       - ENV=dev
+      #  - OIDC_VERIFY_SSL=False
+      - OIDC_OP_USER_ENDPOINT=https://{{cookiecutter.project_name}}.local/auth/realms/{{cookiecutter.project_name}}/protocol/openid-connect/userinfo

{{cookiecutter.project_name}}/pyproject.toml

@@ -16,6 +16,7 @@ djangorestframework = "^3.15.2"
 djangorestframework-jsonapi = "^7.0.2"
 djangorestframework-simplejwt = "^5.3.1"
 gunicorn = "^22.0.0"
+mozilla-django-oidc = "^4.0.1"
 psycopg2 = "^2.9.9"
 
 [tool.poetry.group.dev.dependencies]
@@ -32,6 +33,7 @@ pytest-factoryboy = "^2.7.0"
 pytest-freezer = "^0.4.8"
 pytest-mock = "^3.14.0"
 pytest-randomly = "^3.15.0"
+requests-mock = "^1.12.1"
 
 [tool.ruff]
 exclude = [
@@ -106,7 +108,7 @@ filterwarnings = [
     "error::DeprecationWarning",
     "error::PendingDeprecationWarning",
     "ignore:distutils Version classes are deprecated. Use packaging.version instead.:DeprecationWarning",  # issue in pytest-freezer
-
+    "ignore:CSR support in pyOpenSSL is deprecated. You should use the APIs in cryptography.:DeprecationWarning"
 ]
 
 [tool.coverage.run]

{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/conftest.py

@@ -1,14 +1,74 @@
+from functools import partial
+
 import pytest
+from django.core.cache import cache
 from pytest_factoryboy import register
 from rest_framework.test import APIClient
 
 from .{{cookiecutter.django_app}} import factories
+from .{{cookiecutter.django_app}}.models import OIDCUser
+
+register(factories.UserProfileFactory)
+
+
+def _get_claims(
+    settings,
+    id_claim="00000000-0000-0000-0000-000000000000",
+    groups_claim=None,
+    email_claim="test@example.com",
+    first_name_claim=None,
+    last_name_claim=None,
+):
+    groups_claim = groups_claim if groups_claim else []
+    return {
+        settings.OIDC_ID_CLAIM: id_claim,
+        settings.OIDC_GROUPS_CLAIM: groups_claim,
+        settings.OIDC_EMAIL_CLAIM: email_claim,
+        settings.OIDC_FIRST_NAME_CLAIM: first_name_claim,
+        settings.OIDC_LAST_NAME_CLAIM: last_name_claim,
+    }
+
+
+@pytest.fixture
+def get_claims(settings):
+    return partial(_get_claims, settings)
+
 
-register(factories.UserFactory)
+@pytest.fixture
+def claims(settings):
+    return _get_claims(settings)
+
+
+@pytest.fixture
+def admin_user(settings, get_claims):
+    return OIDCUser(
+        "sometoken",
+        get_claims(
+            id_claim="admin",
+            groups_claim=[settings.ADMIN_GROUP],
+            email_claim="admin@example.com",
+        ),
+    )
 
 
 @pytest.fixture
-def admin_client(db, admin_user):
+def user(get_claims):
+    return OIDCUser(
+        "sometoken",
+        get_claims(id_claim="user", groups_claim=[], email_claim="user@example.com"),
+    )
+
+
+@pytest.fixture(params=["admin"])
+def client(db, user, admin_user, request):
+    usermap = {"user": user, "admin": admin_user}
     client = APIClient()
-    client.force_authenticate(user=admin_user)
+    user = usermap[request.param]
+    client.force_authenticate(user=user)
+    client.user = user
     return client
+
+
+@pytest.fixture(autouse=True)
+def _autoclear_cache():
+    cache.clear()

{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/models.py

@@ -0,0 +1,16 @@
+import uuid
+
+from django.db import models
+
+
+class UUIDModel(models.Model):
+    """
+    Models which use uuid as primary key.
+
+    Defined as {{cookiecutter.project_name}} default
+    """
+
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+
+    class Meta:
+        abstract = True

{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/settings.py

@@ -1,4 +1,3 @@
-import datetime
 import os
 import re
 
@@ -85,8 +84,6 @@ def default(default_dev=env.NOTSET, default_prod=env.NOTSET):
 USE_I18N = True
 USE_TZ = True
 
-AUTH_USER_MODEL = "{{cookiecutter.django_app}}.User"
-
 REST_FRAMEWORK = {
     "EXCEPTION_HANDLER": "rest_framework_json_api.exceptions.exception_handler",
     "DEFAULT_PAGINATION_CLASS": "rest_framework_json_api.pagination.JsonApiPageNumberPagination",
@@ -102,7 +99,7 @@ def default(default_dev=env.NOTSET, default_prod=env.NOTSET):
     ),
     "DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),
     "DEFAULT_AUTHENTICATION_CLASSES": (
-        "rest_framework_simplejwt.authentication.JWTAuthentication",
+        "mozilla_django_oidc.contrib.drf.OIDCAuthentication",
     ),
     "DEFAULT_METADATA_CLASS": "rest_framework_json_api.metadata.JSONAPIMetadata",
     "DEFAULT_FILTER_BACKENDS": (
@@ -123,10 +120,31 @@ def default(default_dev=env.NOTSET, default_prod=env.NOTSET):
 JSON_API_FORMAT_TYPES = "dasherize"
 JSON_API_PLURALIZE_TYPES = True
 
-SIMPLE_AUTH = {
-    "ACCESS_TOKEN_LIFETIME": datetime.timedelta(days=2),
-    "REFRESH_TOKEN_LIFETIME": datetime.timedelta(days=7),
-}
+# Authentication
+OIDC_OP_USER_ENDPOINT = env.str("OIDC_OP_USER_ENDPOINT", default=None)
+OIDC_OP_TOKEN_ENDPOINT = "not supported in {{cookiecutter.project_name}}, but a value is needed"  # noqa: S105
+OIDC_VERIFY_SSL = env.bool("OIDC_VERIFY_SSL", default=True)
+OIDC_ID_CLAIM = env.str("OIDC_ID_CLAIM", default="sub")
+OIDC_EMAIL_CLAIM = env.str("OIDC_EMAIL_CLAIM", default="email")
+OIDC_FIRST_NAME_CLAIM = env.str("OIDC_FIRST_NAME_CLAIM", default="given_name")
+OIDC_LAST_NAME_CLAIM = env.str("OIDC_LAST_NAME_CLAIM", default="family_name")
+OIDC_GROUPS_CLAIM = env.str("OIDC_GROUPS_CLAIM", default="{{cookiecutter.project_name}}_groups")
+OIDC_CLIENT_GRANT_USERNAME_CLAIM = env.str(
+    "OIDC_CLIENT_GRANT_USERNAME_CLAIM",
+    default="preferred_username",
+)
+OIDC_BEARER_TOKEN_REVALIDATION_TIME = env.int(
+    "OIDC_BEARER_TOKEN_REVALIDATION_TIME",
+    default=300,
+)
+OIDC_DRF_AUTH_BACKEND = "{{cookiecutter.project_name}}.{{cookiecutter.django_app}}.authentication.OIDCAuthenticationBackend"
+
+
+# Needed to instantiate `mozilla_django_oidc.auth.OIDCAuthenticationBackend`
+OIDC_RP_CLIENT_ID = None
+OIDC_RP_CLIENT_SECRET = None
+
+ADMIN_GROUP = env.str("ADMIN_GROUP", default="admin")
 
 
 def parse_admins(admins):
@@ -140,8 +158,10 @@ def parse_admins(admins):
     for admin in admins:
         match = re.search(r"(.+) \<(.+@.+)\>", admin)
         if not match:  # pragma: no cover
-            msg = (f'In ADMINS admin "{admin}" is not in correct '
-                   '"Firstname Lastname <email@example.com>" format')
+            msg = (
+                f'In ADMINS admin "{admin}" is not in correct '
+                '"Firstname Lastname <email@example.com>" format'
+            )
             raise environ.ImproperlyConfigured(msg)
         result.append((match.group(1), match.group(2)))
     return result

{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/{{cookiecutter.django_app}}/authentication.py

@@ -0,0 +1,62 @@
+import functools
+import hashlib
+import warnings
+from typing import NamedTuple
+
+from django.conf import settings
+from django.core.cache import cache
+from django.core.exceptions import SuspiciousOperation
+from django.utils.encoding import force_bytes
+from mozilla_django_oidc.auth import OIDCAuthenticationBackend
+from urllib3.exceptions import InsecureRequestWarning
+
+from .models import OIDCUser
+
+
+class OIDCAuthenticationBackend(OIDCAuthenticationBackend):
+    class _HistoricalRequestUser(NamedTuple):
+        id: str
+
+    def verify_claims(self, claims):
+        # claims for human users
+        claims_to_verify = [
+            settings.OIDC_ID_CLAIM,
+            settings.OIDC_EMAIL_CLAIM,
+        ]
+
+        # # claims for application clients
+        # if claims.get(settings.OIDC_CLIENT_GRANT_USERNAME_CLAIM) in [
+        #     settings.OIDC_RP_CLIENT_USERNAME,
+        #     settings.OIDC_MONITORING_CLIENT_USERNAME,
+        # ]:
+        #     claims_to_verify = [
+        #         settings.OIDC_ID_CLAIM,
+        #     ]
+
+        for claim in claims_to_verify:
+            if claim not in claims:
+                msg = f'Couldn\'t find "{claim}" claim'
+                raise SuspiciousOperation(msg)
+
+    def get_or_create_user(self, access_token, id_token, payload):
+        """Verify claims and return user, otherwise raise an Exception."""
+
+        claims = self.cached_request(access_token, id_token, payload)
+
+        self.verify_claims(claims)
+
+        return OIDCUser(access_token, claims)
+
+    def cached_request(self, access_token, id_token, payload):
+        token_hash = hashlib.sha256(force_bytes(access_token)).hexdigest()
+
+        func = functools.partial(self.get_userinfo, access_token, id_token, payload)
+
+        with warnings.catch_warnings():
+            if settings.DEBUG:  # pragma: no cover
+                warnings.simplefilter("ignore", InsecureRequestWarning)
+            return cache.get_or_set(
+                f"auth.userinfo.{token_hash}",
+                func,
+                timeout=settings.OIDC_BEARER_TOKEN_REVALIDATION_TIME,
+            )

{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/{{cookiecutter.django_app}}/factories.py

@@ -4,11 +4,11 @@
 from . import models
 
 
-class UserFactory(DjangoModelFactory):
+class UserProfileFactory(DjangoModelFactory):
+    idp_id = Faker("uuid4")
+    email = Faker("email")
     first_name = Faker("first_name")
     last_name = Faker("last_name")
-    username = Faker("uuid")
-    email = Faker("safe_email")
 
     class Meta:
-        model = models.User
+        model = models.UserProfile

{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/{{cookiecutter.django_app}}/models.py

@@ -1,7 +1,99 @@
-from django.contrib.auth.models import AbstractUser
+import logging
 
-# from django.db import models
+from django.conf import settings
+from django.db import IntegrityError, models, transaction
+from django.db.models import Q
 
+from ..models import UUIDModel
 
-class User(AbstractUser):
-    pass
+logger = logging.getLogger(__name__)
+
+
+class UserProfile(UUIDModel):
+    idp_id = models.CharField(max_length=255, unique=True, null=True, blank=False)
+    email = models.EmailField(unique=True, null=True, blank=True)
+    first_name = models.CharField(max_length=255, null=True, blank=True)
+    last_name = models.CharField(max_length=255, null=True, blank=True)
+
+    class Meta:
+        ordering = ("last_name", "first_name", "email")
+
+
+class BaseOIDCUser:  # pragma: no cover
+    def __init__(self):
+        self.email = None
+        self.groups = []
+        self.group = None
+        self.token = None
+        self.claims = {}
+        self.is_authenticated = False
+
+    def __str__(self):
+        raise NotImplementedError
+
+    @property
+    def is_admin(self):
+        return settings.ADMIN_GROUP in self.groups
+
+
+class OIDCUser(BaseOIDCUser):
+    def __init__(self, token: str, claims: dict):
+        super().__init__()
+
+        self.claims = claims
+        self.id = self.claims[settings.OIDC_ID_CLAIM]
+        self.email = self.claims.get(settings.OIDC_EMAIL_CLAIM)
+        self.first_name = self.claims.get(settings.OIDC_FIRST_NAME_CLAIM)
+        self.last_name = self.claims.get(settings.OIDC_LAST_NAME_CLAIM)
+
+        self.groups = self.claims.get(settings.OIDC_GROUPS_CLAIM, [])
+        self.group = self.groups[0] if self.groups else None
+        self.token = token
+        self.is_authenticated = True
+        self.profile = self._update_or_create_profile()
+
+    def _update_or_create_profile(self):
+        """
+        Update or create UserProfile.
+
+        Analogous to QuerySet.get_or_create(), in order to handle race conditions as
+        gracefully as possible.
+        """
+        try:
+            profile = UserProfile.objects.get(
+                Q(idp_id=self.id) | Q(email__iexact=self.email),
+            )
+            # we only want to save if necessary in order to prevent adding historical
+            # records on every request
+            if profile.idp_id != self.id or profile.email != self.email:
+                profile.idp_id = self.id
+                profile.email = self.email
+                profile.save()
+        except UserProfile.MultipleObjectsReturned:
+            # TODO: trigger notification for staff members or admins
+            logger.warning(
+                "Found one UserProfile with same idp_id and one with same email. "
+                "Matching on idp_id.",
+            )
+            return UserProfile.objects.get(idp_id=self.id)
+        except UserProfile.DoesNotExist:
+            try:
+                with transaction.atomic(using=UserProfile.objects.db):
+                    return UserProfile.objects.create(
+                        idp_id=self.id,
+                        email=self.email,
+                        first_name=self.first_name,
+                        last_name=self.last_name,
+                    )
+            except IntegrityError:  # pragma: no cover
+                # race condition happened
+                try:
+                    return UserProfile.objects.get(idp_id=self.id)
+                except UserProfile.DoesNotExist:
+                    pass
+                raise
+        else:
+            return profile
+
+    def __str__(self):
+        return f"{self.email} - {self.id}"

{{cookiecutter.project_name}}/{{cookiecutter.project_name}}/{{cookiecutter.django_app}}/serializers.py

@@ -1,8 +1,9 @@
-from django.contrib.auth import get_user_model
 from rest_framework_json_api import serializers
 
+from . import models
+
 
 class UserSerializer(serializers.ModelSerializer):
     class Meta:
-        model = get_user_model()
-        fields = [*get_user_model().REQUIRED_FIELDS, get_user_model().USERNAME_FIELD]
+        model = models.UserProfile
+        fields = "__all__"