Custom User Model & Backends

$ python manage.py migrate --settings=config.settings_default   # built-in auth.User: fine
$ python manage.py migrate                                      # AUTH_USER_MODEL = "accounts.User"
django.db.migrations.exceptions.InconsistentMigrationHistory: Migration admin.0001_initial is applied before its dependency accounts.0001_initial on database 'default'.

import secrets
from io import StringIO
from unittest import mock

from django.contrib.auth import get_user_model
from django.db import IntegrityError
from django.test import TestCase

User = get_user_model()
PASSWORD = secrets.token_urlsafe(16)  # a throwaway value per test run


class ManagerTests(TestCase):
    def test_create_user_and_superuser(self):
        user = User.objects.create_user("Ann@Example.COM", PASSWORD)
        self.assertEqual(user.email, "Ann@example.com")  # only the domain is lowercased
        self.assertTrue(user.check_password(PASSWORD))
        self.assertFalse(user.is_staff)
        admin_user = User.objects.create_superuser("root@example.com", PASSWORD)
        self.assertTrue(admin_user.is_staff and admin_user.is_superuser)
        with self.assertRaises(ValueError):
            User.objects.create_superuser("x@example.com", PASSWORD, is_staff=False)
        with self.assertRaises(ValueError):
            User.objects.create_user("", PASSWORD)

    def test_no_username_field(self):
        self.assertIsNone(getattr(User, "username", None))
        self.assertEqual(User.USERNAME_FIELD, "email")
        self.assertEqual(User.get_email_field_name(), "email")

    def test_case_insensitive_uniqueness(self):
        User.objects.create_user("ann@example.com", PASSWORD)
        with self.assertRaises(IntegrityError):
            User.objects.create_user("ANN@example.com", PASSWORD)

    def test_unusable_password(self):
        user = User.objects.create_user("nopw@example.com")
        self.assertFalse(user.has_usable_password())

    def test_swapped_model(self):
        self.assertIs(User, __import__("accounts.models", fromlist=["User"]).User)
        self.assertEqual(User._meta.label, "accounts.User")

    def test_createsuperuser_command(self):
        from django.core.management import call_command

        with mock.patch.dict("os.environ", {"DJANGO_SUPERUSER_PASSWORD": PASSWORD}):
            call_command("createsuperuser", interactive=False, email="cli@example.com", stdout=StringIO())
        self.assertTrue(User.objects.get(email="cli@example.com").is_superuser)

$ python manage.py test accounts catalog
Found 32 test(s).
................................
Ran 32 tests in 0.1s

OK

import hashlib

from django.contrib.auth import get_user_model
from django.contrib.auth.backends import BaseBackend, ModelBackend
from django.core.cache import cache
from django.core.exceptions import PermissionDenied

from .models import ApiToken


def hash_token(token):
    return hashlib.sha256(token.encode()).hexdigest()


class ApiTokenBackend(BaseBackend):
    """Authenticate with authenticate(request, token=...); no password involved."""

    def user_can_authenticate(self, user):
        return getattr(user, "is_active", True)

    def authenticate(self, request, token=None):
        if token is None:
            return None
        try:
            record = ApiToken.objects.select_related("user").get(digest=hash_token(token))
        except ApiToken.DoesNotExist:
            return None
        return record.user if self.user_can_authenticate(record.user) else None

    def get_user(self, user_id):
        user = get_user_model()._default_manager.filter(pk=user_id).first()
        return user if user and self.user_can_authenticate(user) else None

    # Native async versions: without them Django wraps the sync ones in sync_to_async.
    async def aauthenticate(self, request, token=None):
        if token is None:
            return None
        try:
            record = await ApiToken.objects.select_related("user").aget(digest=hash_token(token))
        except ApiToken.DoesNotExist:
            return None
        return record.user if self.user_can_authenticate(record.user) else None

    async def aget_user(self, user_id):
        user = await get_user_model()._default_manager.filter(pk=user_id).afirst()
        return user if user and self.user_can_authenticate(user) else None


class ThrottledModelBackend(ModelBackend):
    """ModelBackend that gives up after too many failures for one identifier."""

    max_failures = 5
    window = 15 * 60  # seconds

    def _key(self, username, password, kwargs):
        identifier = username or kwargs.get(get_user_model().USERNAME_FIELD)
        if not identifier or password is None:
            return None  # not a username/password attempt (a token, say): not ours to count
        return "login-failures:" + hashlib.sha256(identifier.lower().encode()).hexdigest()

    def authenticate(self, request, username=None, password=None, **kwargs):
        key = self._key(username, password, kwargs)
        if key is None:
            return None
        if cache.get(key, 0) >= self.max_failures:
            raise PermissionDenied("Too many failed attempts")
        user = super().authenticate(request, username, password, **kwargs)
        if user is None:
            cache.add(key, 0, self.window)
            cache.incr(key)
        else:
            cache.delete(key)
        return user

    async def aauthenticate(self, request, username=None, password=None, **kwargs):
        key = self._key(username, password, kwargs)
        if key is None:
            return None
        if await cache.aget(key, 0) >= self.max_failures:
            raise PermissionDenied("Too many failed attempts")
        user = await super().aauthenticate(request, username, password, **kwargs)
        if user is None:
            await cache.aadd(key, 0, self.window)
            await cache.aincr(key)
        else:
            await cache.adelete(key)
        return user

from django.contrib.auth import authenticate
from django.http import JsonResponse


def whoami(request):
    header = request.headers.get("Authorization", "")
    scheme, _, token = header.partition(" ")
    user = authenticate(request, token=token) if scheme.lower() == "bearer" and token else None
    if user is None:
        return JsonResponse({"detail": "invalid token"}, status=401)
    return JsonResponse({"email": user.email, "backend": user.backend})

$ curl -si -H "Authorization: Bearer not-a-real-token" http://127.0.0.1:8740/api/whoami/
HTTP/1.1 200 OK
{"email": "dana@example.com", "backend": "accounts.backends.ApiTokenBackend"}

$ curl -si -H "Authorization: Bearer wrong" http://127.0.0.1:8740/api/whoami/
HTTP/1.1 401 Unauthorized
{"detail": "invalid token"}

import secrets

from django.test import TestCase
from django.urls import reverse

from accounts.backends import hash_token
from accounts.models import ApiToken, User

PASSWORD = secrets.token_urlsafe(16)  # a throwaway value per test run


class ApiTokenTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user("tok@example.com", PASSWORD)
        self.raw = "not-a-real-token"
        ApiToken.objects.create(user=self.user, digest=hash_token(self.raw))

    def test_whoami(self):
        ok = self.client.get(reverse("whoami"), headers={"Authorization": f"Bearer {self.raw}"})
        self.assertEqual(ok.status_code, 200)
        self.assertEqual(ok.json(), {"email": "tok@example.com", "backend": "accounts.backends.ApiTokenBackend"})
        bad = self.client.get(reverse("whoami"), headers={"Authorization": "Bearer nope"})
        self.assertEqual(bad.status_code, 401)

    def test_backend_signature_selects_backend(self):
        # username/password credentials never reach ApiTokenBackend, and vice versa
        self.assertIsNone(authenticate(token="nope"))
        self.assertEqual(authenticate(token=self.raw), self.user)

class ThrottledModelBackend(ModelBackend):
    """ModelBackend that gives up after too many failures for one identifier."""

    max_failures = 5
    window = 15 * 60  # seconds

    def _key(self, username, password, kwargs):
        identifier = username or kwargs.get(get_user_model().USERNAME_FIELD)
        if not identifier or password is None:
            return None  # not a username/password attempt (a token, say): not ours to count
        return "login-failures:" + hashlib.sha256(identifier.lower().encode()).hexdigest()

    def authenticate(self, request, username=None, password=None, **kwargs):
        key = self._key(username, password, kwargs)
        if key is None:
            return None
        if cache.get(key, 0) >= self.max_failures:
            raise PermissionDenied("Too many failed attempts")
        user = super().authenticate(request, username, password, **kwargs)
        if user is None:
            cache.add(key, 0, self.window)
            cache.incr(key)
        else:
            cache.delete(key)
        return user

import secrets

from django.contrib.auth import authenticate
from django.core.cache import cache
from django.test import TestCase
from django.urls import reverse

from accounts.models import User

PASSWORD = secrets.token_urlsafe(16)  # a throwaway value per test run


class AuthenticationTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.user = User.objects.create_user("bea@example.com", PASSWORD)

    def setUp(self):
        cache.clear()

    def test_login_by_email_any_case(self):
        self.assertEqual(authenticate(username="BEA@example.com", password=PASSWORD), self.user)
        self.assertEqual(authenticate(email="bea@example.com", password=PASSWORD), self.user)
        self.assertTrue(self.client.login(username="bea@EXAMPLE.com", password=PASSWORD))

    def test_throttling_raises_permission_denied(self):
        for _ in range(5):
            self.assertIsNone(authenticate(username="bea@example.com", password="wrong"))
        # the sixth attempt is refused even with the right password
        self.assertIsNone(authenticate(username="bea@example.com", password=PASSWORD))
        cache.clear()
        self.assertEqual(authenticate(username="bea@example.com", password=PASSWORD), self.user)

    def test_token_attempts_do_not_trip_the_password_throttle(self):
        for _ in range(10):
            self.assertIsNone(authenticate(token="nope"))
        self.assertEqual(authenticate(username="bea@example.com", password=PASSWORD), self.user)

    def test_login_view(self):
        response = self.client.post(reverse("login"), {"username": "bea@example.com", "password": PASSWORD})
        self.assertEqual(response.status_code, 302)
        self.assertEqual(self.client.session["_auth_user_backend"], "accounts.backends.ThrottledModelBackend")

    def test_login_needs_backend_when_several_are_configured(self):
        from django.contrib.auth import login
        from django.contrib.sessions.backends.db import SessionStore
        from django.test import RequestFactory

        request = RequestFactory().get("/")
        request.session = SessionStore()
        with self.assertRaises(ValueError):  # user was not returned by authenticate()
            login(request, self.user)
        login(request, self.user, backend="accounts.backends.ThrottledModelBackend")
        self.assertEqual(request.session["_auth_user_backend"], "accounts.backends.ThrottledModelBackend")

    def test_inactive_user_rejected(self):
        self.user.is_active = False
        self.user.save()
        self.assertIsNone(authenticate(username="bea@example.com", password=PASSWORD))

from django.contrib.auth.backends import BaseBackend

from .models import Review


class ReviewAuthorBackend(BaseBackend):
    """Object-level permissions: authors may change and delete their own reviews."""

    def has_perm(self, user_obj, perm, obj=None):
        if not user_obj.is_active or not isinstance(obj, Review):
            return False
        return perm in {"catalog.change_review", "catalog.delete_review"} and obj.author_id == user_obj.pk

import secrets

from django.test import TestCase

from accounts.models import User
from catalog.models import Review


class ReviewAuthorBackendTests(TestCase):
    def test_object_permissions(self):
        author = User.objects.create_user("a@example.com", secrets.token_urlsafe(16))
        other = User.objects.create_user("b@example.com", secrets.token_urlsafe(16))
        review = Review.objects.create(author=author, body="Great")
        self.assertTrue(author.has_perm("catalog.change_review", review))
        self.assertFalse(other.has_perm("catalog.change_review", review))
        self.assertFalse(author.has_perm("catalog.change_review"))  # model-level: no
        self.assertFalse(author.has_perm("catalog.view_review", review))

AUTH_USER_MODEL = "accounts.User"
AUTHENTICATION_BACKENDS = [
    "accounts.backends.ThrottledModelBackend",
    "accounts.backends.ApiTokenBackend",
    "catalog.backends.ReviewAuthorBackend",
]

import secrets

from asgiref.sync import sync_to_async
from django.contrib.auth import aauthenticate
from django.core.cache import cache
from django.test import AsyncClient, TestCase

from accounts.backends import hash_token
from accounts.models import ApiToken, User

PASSWORD = secrets.token_urlsafe(16)  # a throwaway value per test run


class AsyncTests(TestCase):
    def setUp(self):
        cache.clear()

    async def test_async_backends(self):
        user = await sync_to_async(User.objects.create_user)("as@example.com", PASSWORD)
        self.assertEqual(await aauthenticate(username="AS@example.com", password=PASSWORD), user)
        await ApiToken.objects.acreate(user=user, digest=hash_token("not-a-real-token"))
        found = await aauthenticate(token="not-a-real-token")
        self.assertEqual(found, user)
        self.assertEqual(found.backend, "accounts.backends.ApiTokenBackend")
        client = AsyncClient()
        self.assertTrue(await client.alogin(username="as@example.com", password=PASSWORD))

from django.contrib.auth.backends import RemoteUserBackend
from django.contrib.auth.middleware import PersistentRemoteUserMiddleware, RemoteUserMiddleware


class ProxyUserBackend(RemoteUserBackend):
    create_unknown_user = True

    def clean_username(self, username):
        return username.strip().lower()  # identifiers are emails here

    def configure_user(self, request, user, created=True):
        if created:
            user.set_unusable_password()  # the proxy is the only way in
            user.display_name = user.email.split("@")[0].title()
            user.save(update_fields=["password", "display_name"])
        return user


class ProxyHeaderMiddleware(RemoteUserMiddleware):
    header = "HTTP_X_PROXY_USER"


class SsoLoginMiddleware(PersistentRemoteUserMiddleware):
    header = "HTTP_X_PROXY_USER"

class ProxyHeaderMiddleware(RemoteUserMiddleware):
    header = "HTTP_X_PROXY_USER"


class SsoLoginMiddleware(PersistentRemoteUserMiddleware):
    header = "HTTP_X_PROXY_USER"

from asgiref.sync import sync_to_async
from django.test import AsyncClient, TestCase, override_settings
from django.urls import reverse

from accounts.models import User


class RemoteUserTests(TestCase):
    MW = [
        "django.contrib.sessions.middleware.SessionMiddleware",
        "django.contrib.auth.middleware.AuthenticationMiddleware",
        "django.contrib.auth.middleware.RemoteUserMiddleware",
    ]
    BACKENDS = ["accounts.remote.ProxyUserBackend", "accounts.backends.ThrottledModelBackend"]

    @override_settings(MIDDLEWARE=MW, AUTHENTICATION_BACKENDS=BACKENDS)
    def test_remote_user_wsgi(self):
        response = self.client.get(reverse("whoami"), REMOTE_USER="  Carl@Example.com ")
        self.assertEqual(response.status_code, 401)  # whoami wants a token...
        user = User.objects.get(email="carl@example.com")  # ...but the user was created
        self.assertEqual(user.display_name, "Carl")
        self.assertEqual(self.client.session["_auth_user_backend"], "accounts.remote.ProxyUserBackend")
        self.assertFalse(user.has_usable_password())

    @override_settings(MIDDLEWARE=MW, AUTHENTICATION_BACKENDS=BACKENDS)
    def test_header_is_not_trusted_under_wsgi(self):
        self.client.get(reverse("whoami"), headers={"Remote-User": "evil@example.com"})
        self.assertFalse(User.objects.filter(email="evil@example.com").exists())

    @override_settings(MIDDLEWARE=MW, AUTHENTICATION_BACKENDS=BACKENDS)
    def test_other_user_logs_out_previous(self):
        self.client.get("/api/whoami/", REMOTE_USER="a@example.com")
        self.client.get("/api/whoami/", REMOTE_USER="b@example.com")
        self.assertEqual(int(self.client.session["_auth_user_id"]), User.objects.get(email="b@example.com").pk)

    @override_settings(MIDDLEWARE=MW, AUTHENTICATION_BACKENDS=BACKENDS)
    def test_missing_header_logs_out_with_plain_middleware(self):
        self.client.get("/api/whoami/", REMOTE_USER="a@example.com")
        self.assertIn("_auth_user_id", self.client.session)
        self.client.get("/api/whoami/")
        self.assertNotIn("_auth_user_id", self.client.session)

    @override_settings(
        MIDDLEWARE=[MW[0], MW[1], "accounts.remote.SsoLoginMiddleware"], AUTHENTICATION_BACKENDS=BACKENDS
    )
    def test_persistent_middleware_keeps_session(self):
        self.client.get("/api/whoami/", headers={"X-Proxy-User": "p@example.com"})
        self.assertIn("_auth_user_id", self.client.session)
        self.client.get("/api/whoami/")  # no header: session survives
        self.assertIn("_auth_user_id", self.client.session)

    @override_settings(
        MIDDLEWARE=[MW[0], MW[1], "accounts.remote.ProxyHeaderMiddleware"], AUTHENTICATION_BACKENDS=BACKENDS
    )
    def test_custom_header(self):
        self.client.get("/api/whoami/", headers={"X-Proxy-User": "h@example.com"})
        self.assertTrue(User.objects.filter(email="h@example.com").exists())

    @override_settings(MIDDLEWARE=MW, AUTHENTICATION_BACKENDS=BACKENDS)
    async def test_remote_user_asgi_header(self):
        client = AsyncClient()
        await client.get("/api/whoami/", headers={"Remote-User": "asgi@example.com"})
        self.assertTrue(await User.objects.filter(email="asgi@example.com").aexists())

import os

from .settings import *  # noqa: F401,F403
from .settings import INSTALLED_APPS, MIDDLEWARE

SITE_ID = 1
INSTALLED_APPS = INSTALLED_APPS + [
    "django.contrib.sites",
    "allauth",
    "allauth.account",
    "allauth.socialaccount",
    "allauth.socialaccount.providers.google",
    "allauth.mfa",
]
MIDDLEWARE = MIDDLEWARE + ["allauth.account.middleware.AccountMiddleware"]
ROOT_URLCONF = "config.urls_allauth"

AUTHENTICATION_BACKENDS = ["allauth.account.auth_backends.AuthenticationBackend"]

ACCOUNT_USER_MODEL_USERNAME_FIELD = None
ACCOUNT_LOGIN_METHODS = {"email"}
ACCOUNT_SIGNUP_FIELDS = ["email*", "password1*", "password2*"]
ACCOUNT_EMAIL_VERIFICATION = "mandatory"
LOGIN_URL = "account_login"

SOCIALACCOUNT_PROVIDERS = {
    "google": {
        "APP": {
            "client_id": os.environ.get("GOOGLE_CLIENT_ID", ""),
            "secret": os.environ.get("GOOGLE_CLIENT_SECRET", ""),
        },
        "SCOPE": ["profile", "email"],
    }
}
MFA_SUPPORTED_TYPES = ["totp", "recovery_codes"]

from allauth.account.decorators import secure_admin_login
from django.contrib import admin
from django.urls import include, path

admin.autodiscover()
admin.site.login = secure_admin_login(admin.site.login)

urlpatterns = [
    path("admin/", admin.site.urls),
    path("accounts/", include("allauth.urls")),
]

$ pip install "django-allauth[mfa,socialaccount]==65.19.7"
$ python manage.py check --settings=config.settings_allauth
System check identified no issues (0 silenced).
$ python manage.py test accounts.test_allauth --settings=config.settings_allauth
Found 3 test(s).
...
Ran 3 tests in 0.1s

OK

import re
import secrets

from django.contrib.auth import get_user_model
from django.core import mail
from django.test import TestCase


class AllauthTests(TestCase):
    def test_signup_login_flow(self):
        r = self.client.get("/accounts/signup/")
        self.assertEqual(r.status_code, 200)
        self.assertNotContains(r, 'name="username"')
        self.assertContains(r, 'name="email"')
        password = secrets.token_urlsafe(16)
        r = self.client.post("/accounts/signup/", {"email": "Eve@Example.com", "password1": password, "password2": password})
        self.assertEqual(r.status_code, 302)
        user = get_user_model().objects.get(email__iexact="eve@example.com")
        self.assertEqual(len(mail.outbox), 1)
        link = re.search(r"http://testserver(/accounts/confirm-email/\S+/)", mail.outbox[0].body).group(1)
        self.assertEqual(self.client.get(link).status_code, 200)
        self.client.post(link)
        r = self.client.post("/accounts/login/", {"login": "eve@example.com", "password": password})
        self.assertEqual(r.status_code, 302)
        self.assertEqual(self.client.session["_auth_user_backend"], "allauth.account.auth_backends.AuthenticationBackend")

    def test_login_page_and_mfa_and_google(self):
        self.assertEqual(self.client.get("/accounts/login/").status_code, 200)
        self.assertEqual(self.client.get("/accounts/2fa/").status_code, 302)  # needs login
        r = self.client.get("/accounts/google/login/")
        self.assertIn(r.status_code, (200, 302))

    def test_admin_login_requires_allauth_login(self):
        r = self.client.get("/admin/login/")
        self.assertEqual(r.status_code, 302)
        self.assertIn("/accounts/login/", r["Location"])